analytics worker refactor

app/tasks/analytics_worker.py

@@ -202,15 +202,71 @@ class AnalyticsWorker:
     
     # app/tasks/analytics_worker.py - Replace _sync_load_dataframe
 
+    # def _sync_load_dataframe(self, entity_type: str) -> pd.DataFrame:
+    #     """
+    #     Load data with entity context (receives entity_type from STEP 2)
+    #     """
+    #     try:
+    #         conn = get_conn(self.org_id)
+    #         table_name = f"main.{entity_type}_canonical"
+        
+    #         # Verify table exists first
+    #         table_exists = conn.execute(
+    #             "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'main' AND table_name = ?",
+    #             [entity_type + "_canonical"]
+    #         ).fetchone()[0] > 0
+        
+    #         if not table_exists:
+    #             logger.error(f"[LOAD] Table {table_name} does not exist")
+    #             return pd.DataFrame()
+        
+    #         # Load with time window
+    #         cutoff = datetime.now() - timedelta(hours=self.hours_window)
+    #         df = conn.execute(
+    #             f"SELECT * FROM {table_name} WHERE timestamp >= ? ORDER BY timestamp DESC LIMIT 10000",
+    #             [cutoff]
+    #         ).df()
+        
+    #         if not df.empty:
+    #             logger.info(f"[LOAD] Success: {len(df)} rows × {len(df.columns)} cols (time-filtered)")
+    #             return df
+        
+    #         # Fallback to recent data
+    #         logger.warning(f"[LOAD] No data in {self.hours_window}h window, returning recent rows")
+    #         df = conn.execute(f"SELECT * FROM {table_name} ORDER BY timestamp DESC LIMIT 1000").df()
+        
+    #         if df.empty:
+    #             logger.error(f"[LOAD] Table exists but contains no rows")
+            
+    #         return df
+        
+    #     except Exception as e:
+    #         logger.error(f"[LOAD] Fatal error: {e}")
+    #         return pd.DataFrame()
+    
+    # # app/tasks/analytics_worker.py - Add these inside AnalyticsWorker class
+
+    async def _load_dataframe(self) -> pd.DataFrame:
+        """
+        Load data asynchronously (non-blocking)
+        Requires: self._entity_type must be set from Redis first
+        """
+        if not hasattr(self, '_entity_type') or not self._entity_type:
+            raise ValueError("entity_type must be loaded from Redis first")
+    
+        # Run sync DB operation in thread pool
+        return await asyncio.to_thread(self._sync_load_dataframe, self._entity_type)
+
     def _sync_load_dataframe(self, entity_type: str) -> pd.DataFrame:
         """
-        Load data with entity context (receives entity_type from STEP 2)
+        Synchronous data loader (runs in thread pool)
+        Receives entity_type from STEP 2 (_load_entity_from_redis)
         """
         try:
             conn = get_conn(self.org_id)
             table_name = f"main.{entity_type}_canonical"
         
-            # Verify table exists first
+            # Verify table exists
             table_exists = conn.execute(
                 "SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = 'main' AND table_name = ?",
                 [entity_type + "_canonical"]
@@ -243,7 +299,7 @@ class AnalyticsWorker:
         except Exception as e:
             logger.error(f"[LOAD] Fatal error: {e}")
             return pd.DataFrame()
-    
+
     async def _load_entity_from_redis(self) -> dict:
         """Instantly load entity/industry from Redis (source of truth)"""
         try: