♻️ Backend code cleanup and import organization (data_process/) #1037

Author: WMC001

backend/data_process/app.py

@@ -2,10 +2,12 @@
 Celery application configuration for data processing tasks
 """
 import logging
+
 from celery import Celery
-from consts.const import REDIS_URL, REDIS_BACKEND_URL, ELASTICSEARCH_SERVICE
 from celery.backends.base import DisabledBackend
 
+from consts.const import ELASTICSEARCH_SERVICE, REDIS_BACKEND_URL, REDIS_URL
+
 # Configure logging
 logger = logging.getLogger("data_process.app")
 

backend/data_process/ray_actors.py

@@ -1,12 +1,13 @@
 import logging
+from typing import Any, Dict, List, Optional
+
 import ray
-from typing import List, Dict, Any, Optional
 
-from nexent.data_process import DataProcessCore
-from database.attachment_db import get_file_stream
 from consts.const import RAY_ACTOR_NUM_CPUS
+from database.attachment_db import get_file_stream
+from nexent.data_process import DataProcessCore
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger("data_process.ray_actors")
 # This now controls the number of CPUs requested by each DataProcessorRayActor instance.
 # It allows a single file processing task to potentially use more than one core if the
 # underlying processing library (e.g., unstructured) can leverage it.
@@ -22,7 +23,14 @@ def __init__(self):
         logger.info(f"Ray actor initialized using {RAY_ACTOR_NUM_CPUS} CPU cores...")
         self._processor = DataProcessCore()
 
-    def process_file(self, source: str, chunking_strategy: str, destination: str, task_id: Optional[str] = None, **params) -> List[Dict[str, Any]]:
+    def process_file(
+        self,
+        source: str,
+        chunking_strategy: str,
+        destination: str,
+        task_id: Optional[str] = None,
+        **params
+    ) -> List[Dict[str, Any]]:
         """
         Process a file, auto-detecting its type using DataProcessCore.file_process.
 
@@ -62,4 +70,4 @@ def process_file(self, source: str, chunking_strategy: str, destination: str, ta
             return []
 
         logger.debug(f"[RayActor] file_process returned {len(chunks)} chunks, returning as is.")
-        return chunks 
+        return chunks

backend/data_process/ray_config.py

@@ -2,32 +2,41 @@
 Ray configuration management module
 """
 
-import os
 import logging
+import os
+from typing import Any, Dict, Optional
+
 import ray
-from typing import Dict, Any, Optional
-from consts.const import RAY_PLASMA_DIRECTORY, RAY_OBJECT_STORE_MEMORY_GB, RAY_TEMP_DIR, RAY_NUM_CPUS
 
-logger = logging.getLogger(__name__)
+from consts.const import (
+    RAY_NUM_CPUS,
+    RAY_OBJECT_STORE_MEMORY_GB,
+    RAY_PLASMA_DIRECTORY,
+    RAY_TEMP_DIR,
+)
+
+logger = logging.getLogger("data_process.ray_config")
 
 # Forward declaration variable so runtime references succeed before instantiation
 ray_config: Optional["RayConfig"] = None
 
 
 class RayConfig:
     """Ray configuration manager"""
-    
+
     def __init__(self):
         self.plasma_directory = RAY_PLASMA_DIRECTORY
         self.object_store_memory_gb = RAY_OBJECT_STORE_MEMORY_GB
         self.temp_dir = RAY_TEMP_DIR
-    
-    def get_init_params(self, 
-                       address: Optional[str] = None,
-                       num_cpus: Optional[int] = None,
-                       include_dashboard: bool = False,
-                       dashboard_host: str = "0.0.0.0",
-                       dashboard_port: int = 8265) -> Dict[str, Any]:
+
+    def get_init_params(
+            self,
+            address: Optional[str] = None,
+            num_cpus: Optional[int] = None,
+            include_dashboard: bool = False,
+            dashboard_host: str = "0.0.0.0",
+            dashboard_port: int = 8265
+    ) -> Dict[str, Any]:
         """
         Get Ray initialization parameters
         
@@ -45,29 +54,29 @@ def get_init_params(self,
             "ignore_reinit_error": True,
             "_plasma_directory": self.plasma_directory,
         }
-        
+
         if address:
             params["address"] = address
         else:
             # Local cluster configuration
             if num_cpus:
                 params["num_cpus"] = num_cpus
-            
+
             # Object store memory configuration (convert to bytes)
             object_store_memory = int(self.object_store_memory_gb * 1024 * 1024 * 1024)
             params["object_store_memory"] = object_store_memory
-            
+
             # Temp directory configuration
             params["_temp_dir"] = self.temp_dir
-            
+
             # Dashboard configuration
             if include_dashboard:
                 params["include_dashboard"] = True
                 params["dashboard_host"] = dashboard_host
                 params["dashboard_port"] = dashboard_port
-        
+
         return params
-    
+
     def init_ray(self, **kwargs) -> bool:
         """
         Initialize Ray
@@ -82,11 +91,8 @@ def init_ray(self, **kwargs) -> bool:
             if ray.is_initialized():
                 logger.info("Ray already initialized, skipping...")
                 return True
-            
+
             params = self.get_init_params(**kwargs)
-            
-            # Get Ray configuration from environment
-            num_cpus = int(RAY_NUM_CPUS) if RAY_NUM_CPUS else None  # None lets Ray decide
 
             # Log the attempt to initialize
             logger.debug("Initializing Ray cluster...")
@@ -95,27 +101,27 @@ def init_ray(self, **kwargs) -> bool:
                 if key.startswith('_'):
                     logger.debug(f"  {key}: {value}")
                 elif key == 'object_store_memory':
-                    logger.debug(f"  {key}: {value / (1024**3):.1f} GB")
+                    logger.debug(f"  {key}: {value / (1024 ** 3):.1f} GB")
                 else:
                     logger.debug(f"  {key}: {value}")
-            
+
             ray.init(**params)
             logger.info("✅ Ray initialization successful")
-            
+
             # Display cluster information
             try:
                 if hasattr(ray, 'cluster_resources'):
                     resources = ray.cluster_resources()
                     logger.debug(f"Ray cluster resources: {resources}")
             except Exception as e:
                 logger.error(f"Failed to get cluster resources information: {e}")
-            
+
             return True
-            
+
         except Exception as e:
             logger.error(f"❌ Ray initialization failed: {str(e)}")
             return False
-    
+
     def connect_to_cluster(self, address: str = "auto") -> bool:
         """
         Connect to existing Ray cluster
@@ -130,23 +136,25 @@ def connect_to_cluster(self, address: str = "auto") -> bool:
             if ray.is_initialized():
                 logger.debug("Ray already initialized, skipping...")
                 return True
-            
+
             params = self.get_init_params(address=address)
-            
+
             logger.debug(f"Connecting to Ray cluster: {address}")
             ray.init(**params)
             logger.info("✅ Successfully connected to Ray cluster")
-            
+
             return True
-            
+
         except Exception as e:
             logger.info(f"Cannot connect to Ray cluster: {str(e)}")
             return False
-    
-    def start_local_cluster(self, 
-                          num_cpus: Optional[int] = None,
-                          include_dashboard: bool = True,
-                          dashboard_port: int = 8265) -> bool:
+
+    def start_local_cluster(
+            self,
+            num_cpus: Optional[int] = None,
+            include_dashboard: bool = True,
+            dashboard_port: int = 8265
+    ) -> bool:
         """
         Start local Ray cluster
         
@@ -160,13 +168,13 @@ def start_local_cluster(self,
         """
         if num_cpus is None:
             num_cpus = os.cpu_count()
-        
+
         return self.init_ray(
             num_cpus=num_cpus,
             include_dashboard=include_dashboard,
             dashboard_port=dashboard_port
         )
-    
+
     def log_configuration(self):
         """Log current configuration information"""
         logger.debug("Ray Configuration:")
@@ -182,11 +190,13 @@ def init_ray_for_worker(cls, address: str = "auto") -> bool:
         return ray_config.connect_to_cluster(address)
 
     @classmethod
-    def init_ray_for_service(cls,
-                             num_cpus: Optional[int] = None,
-                             dashboard_port: int = 8265,
-                             try_connect_first: bool = True,
-                             include_dashboard: bool = True) -> bool:
+    def init_ray_for_service(
+            cls,
+            num_cpus: Optional[int] = None,
+            dashboard_port: int = 8265,
+            try_connect_first: bool = True,
+            include_dashboard: bool = True
+    ) -> bool:
         """Initialize Ray for data processing service (class method wrapper)."""
         ray_config.log_configuration()
 
@@ -203,7 +213,3 @@ def init_ray_for_service(cls,
             include_dashboard=include_dashboard,
             dashboard_port=dashboard_port
         )
-
-# Create a global RayConfig instance accessible throughout the module
-ray_config = RayConfig()
-