♻️ Refactor: Sort backend imports (unit test) #1037

Author: WMC001

backend/services/agent_service.py

@@ -729,8 +729,7 @@ def _memory_token(message_text: str) -> str:
             ):
                 yield data_chunk
         except Exception as run_exc:
-            logger.error(
-                f"Agent run error after memory failure: {str(run_exc)}")
+            logger.error(f"Agent run error after memory failure: {str(run_exc)}")
             raise AgentRunException(f"Agent run error: {str(run_exc)}")
     except Exception as e:
         logger.error(f"Generate stream with memory error: {str(e)}")

backend/services/conversation_management_service.py

@@ -69,8 +69,7 @@ def save_message(request: MessageRequest, authorization: Optional[str] = Header(
         # Validate conversation_id
         conversation_id = message_data.get('conversation_id')
         if not conversation_id:
-            raise Exception(
-                "conversation_id is required, please call /conversation/create to create a conversation first")
+            raise Exception("conversation_id is required, please call /conversation/create to create a conversation first")
 
         # Process different types of message units
         message_units = message_data['message']
@@ -303,8 +302,7 @@ def update_conversation_title(conversation_id: int, title: str, user_id: str = N
     """
     success = rename_conversation(conversation_id, title, user_id)
     if not success:
-        raise Exception(
-            f"Conversation {conversation_id} does not exist or has been deleted")
+        raise Exception(f"Conversation {conversation_id} does not exist or has been deleted")
     return success
 
 
@@ -357,8 +355,7 @@ def rename_conversation_service(conversation_id: int, name: str, user_id: str) -
     try:
         success = rename_conversation(conversation_id, name, user_id)
         if not success:
-            raise Exception(
-                f"Conversation {conversation_id} does not exist or has been deleted")
+            raise Exception(f"Conversation {conversation_id} does not exist or has been deleted")
         return True
     except Exception as e:
         logging.error(f"Failed to rename conversation: {str(e)}")
@@ -379,8 +376,7 @@ def delete_conversation_service(conversation_id: int, user_id: str) -> bool:
     try:
         success = delete_conversation(conversation_id, user_id)
         if not success:
-            raise Exception(
-                f"Conversation {conversation_id} does not exist or has been deleted")
+            raise Exception(f"Conversation {conversation_id} does not exist or has been deleted")
         return True
     except Exception as e:
         logging.error(f"Failed to delete conversation: {str(e)}")

backend/services/data_process_service.py

@@ -65,8 +65,7 @@ def _init_redis_client(self):
                     connection_pool=self.redis_pool)
                 logger.info("Redis client initialized successfully.")
             else:
-                logger.warning(
-                    "REDIS_BACKEND_URL not set, Redis client not initialized.")
+                logger.warning("REDIS_BACKEND_URL not set, Redis client not initialized.")
         except Exception as e:
             logger.error(f"Failed to initialize Redis client: {str(e)}")
 
@@ -83,8 +82,7 @@ def _init_clip_model(self):
             self.clip_available = True
             logger.info("CLIP model loaded successfully")
         except Exception as e:
-            logger.warning(
-                f"Failed to load CLIP model, size-only filtering will be used: {str(e)}")
+            logger.warning(f"Failed to load CLIP model, size-only filtering will be used: {str(e)}")
             self.clip_available = False
 
     async def start(self):
@@ -104,8 +102,7 @@ def _get_celery_inspector(self):
             if not celery_app.conf.broker_url or not celery_app.conf.result_backend:
                 celery_app.conf.broker_url = REDIS_URL
                 celery_app.conf.result_backend = REDIS_BACKEND_URL
-                logger.warning(
-                    f"Celery broker URL is not configured properly, reconfiguring to {celery_app.conf.broker_url}")
+                logger.warning(f"Celery broker URL is not configured properly, reconfiguring to {celery_app.conf.broker_url}")
             try:
                 inspector = celery_app.control.inspect()
                 inspector.ping()
@@ -179,10 +176,8 @@ def get_reserved():
                     # Add to the set, duplicates will be handled
                     task_ids.add(task_id)
             except Exception as redis_error:
-                logger.warning(
-                    f"Failed to query Redis for stored task IDs: {str(redis_error)}")
-            logger.debug(
-                f"Total unique task IDs collected (inspector + Redis): {len(task_ids)}")
+                logger.warning(f"Failed to query Redis for stored task IDs: {str(redis_error)}")
+            logger.debug(f"Total unique task IDs collected (inspector + Redis): {len(task_ids)}")
             tasks = [get_task_info(task_id) for task_id in task_ids]
             all_task_infos = await asyncio.gather(*tasks, return_exceptions=True)
             for task_info in all_task_infos:

backend/services/model_health_service.py

@@ -18,14 +18,14 @@
 
 
 async def _embedding_dimension_check(
-        model_name: str,
-        model_type: str,
-        model_base_url: str,
-        model_api_key: str):
+    model_name: str,
+    model_type: str,
+    model_base_url: str,
+    model_api_key: str):
 
     # Test connectivity based on different model types
     if model_type == "embedding":
-        embedding = await OpenAICompatibleEmbedding(
+        embedding =await OpenAICompatibleEmbedding(
             model_name=model_name,
             base_url=model_base_url,
             api_key=model_api_key,
@@ -34,7 +34,7 @@ async def _embedding_dimension_check(
         if len(embedding) > 0:
             return len(embedding[0])
     elif model_type == "multi_embedding":
-        embedding = await JinaEmbedding(
+        embedding =await JinaEmbedding(
             model_name=model_name,
             base_url=model_base_url,
             api_key=model_api_key,
@@ -65,8 +65,7 @@ async def _perform_connectivity_check(
         bool: Connectivity check result
     """
     if "localhost" in model_base_url or "127.0.0.1" in model_base_url:
-        model_base_url = model_base_url.replace(
-            "localhost", "host.docker.internal").replace("127.0.0.1", "host.docker.internal")
+        model_base_url = model_base_url.replace("localhost", "host.docker.internal").replace("127.0.0.1", "host.docker.internal")
 
     connectivity: bool
 
@@ -139,19 +138,16 @@ async def check_model_connectivity(display_name: str, authorization: Optional[st
                 model_name, model_type, model_base_url, model_api_key
             )
         except Exception as e:
-            update_data = {
-                "connect_status": ModelConnectStatusEnum.UNAVAILABLE.value}
+            update_data = {"connect_status": ModelConnectStatusEnum.UNAVAILABLE.value}
             logger.error(f"Error checking model connectivity: {str(e)}")
             update_model_record(model["model_id"], update_data)
             return ModelResponse(code=400, message=str(e),
                                  data={"connectivity": False, "connect_status": ModelConnectStatusEnum.UNAVAILABLE.value})
 
         if connectivity:
-            logger.info(
-                f"CONNECTED: {model_name}; Base URL: {model.get('base_url')}; API Key: {model.get('api_key')}")
+            logger.info(f"CONNECTED: {model_name}; Base URL: {model.get('base_url')}; API Key: {model.get('api_key')}")
         else:
-            logger.warning(
-                f"UNCONNECTED: {model_name}; Base URL: {model.get('base_url')}; API Key: {model.get('api_key')}")
+            logger.warning(f"UNCONNECTED: {model_name}; Base URL: {model.get('base_url')}; API Key: {model.get('api_key')}")
         connect_status = ModelConnectStatusEnum.AVAILABLE.value if connectivity else ModelConnectStatusEnum.UNAVAILABLE.value
         update_data = {"connect_status": connect_status}
         update_model_record(model["model_id"], update_data)
@@ -160,8 +156,7 @@ async def check_model_connectivity(display_name: str, authorization: Optional[st
     except Exception as e:
         logger.error(f"Error checking model connectivity: {str(e)}")
         if 'model' in locals() and model:
-            update_data = {
-                "connect_status": ModelConnectStatusEnum.UNAVAILABLE.value}
+            update_data = {"connect_status": ModelConnectStatusEnum.UNAVAILABLE.value}
             update_model_record(model["model_id"], update_data)
         return ModelResponse(code=500, message=f"Connectivity test error: {str(e)}",
                              data={"connectivity": False, "connect_status": ModelConnectStatusEnum.UNAVAILABLE.value})
@@ -177,8 +172,7 @@ async def check_me_model_connectivity(model_name: str):
             result = response.json()['data']
 
             # Find model
-            model_data = next(
-                (item for item in result if item['id'] == model_name), None)
+            model_data = next((item for item in result if item['id'] == model_name), None)
             if not model_data:
                 return ModelResponse(code=404, message="Specified model not found",
                                      data={"connectivity": False, "message": "Specified model not found", "connect_status": ""})
@@ -187,8 +181,7 @@ async def check_me_model_connectivity(model_name: str):
 
             # Test model based on type
             if model_type == 'llm':
-                payload = {"model": model_name, "messages": [
-                    {"role": "user", "content": "hello"}]}
+                payload = {"model": model_name, "messages": [{"role": "user", "content": "hello"}]}
                 api_response = await client.post(
                     f"{MODEL_ENGINE_HOST}/open/router/v1/chat/completions",
                     headers=headers,
@@ -238,17 +231,15 @@ async def verify_model_config_connectivity(model_config: dict):
         model_type = model_config["model_type"]
         model_base_url = model_config["base_url"]
         model_api_key = model_config["api_key"]
-        embedding_dim = model_config.get(
-            "embedding_dim", model_config.get("max_tokens", 1024))
+        embedding_dim = model_config.get("embedding_dim", model_config.get("max_tokens", 1024))
 
         try:
             # Use the common connectivity check function
             connectivity = await _perform_connectivity_check(
                 model_name, model_type, model_base_url, model_api_key, embedding_dim
             )
         except ValueError as e:
-            logger.warning(
-                f"UNCONNECTED: {model_name}; Base URL: {model_base_url}; API Key: {model_api_key}; Error: {str(e)}")
+            logger.warning(f"UNCONNECTED: {model_name}; Base URL: {model_base_url}; API Key: {model_api_key}; Error: {str(e)}")
             return ModelResponse(
                 code=400,
                 message=str(e),
@@ -275,8 +266,7 @@ async def verify_model_config_connectivity(model_config: dict):
         )
     except Exception as e:
         error_message = str(e)
-        logger.warning(
-            f"UNCONNECTED: {model_name}; Base URL: {model_base_url}; API Key: {model_api_key}; Error: {error_message}")
+        logger.warning(f"UNCONNECTED: {model_name}; Base URL: {model_base_url}; API Key: {model_api_key}; Error: {error_message}")
         return ModelResponse(
             code=500,
             message="",

backend/services/northbound_service.py

@@ -67,13 +67,11 @@ async def idempotency_start(key: str, ttl_seconds: Optional[int] = None) -> None
     async with _IDEMPOTENCY_LOCK:
         # purge expired
         now = _now_seconds()
-        expired = [k for k, v in _IDEMPOTENCY_RUNNING.items(
-        ) if now - v > (ttl_seconds or _IDEMPOTENCY_TTL_SECONDS_DEFAULT)]
+        expired = [k for k, v in _IDEMPOTENCY_RUNNING.items() if now - v > (ttl_seconds or _IDEMPOTENCY_TTL_SECONDS_DEFAULT)]
         for k in expired:
             _IDEMPOTENCY_RUNNING.pop(k, None)
         if key in _IDEMPOTENCY_RUNNING:
-            raise LimitExceededError(
-                "Duplicate request is still running, please wait.")
+            raise LimitExceededError("Duplicate request is still running, please wait.")
         _IDEMPOTENCY_RUNNING[key] = now
 
 
@@ -93,8 +91,7 @@ async def check_and_consume_rate_limit(tenant_id: str) -> None:
         state = _RATE_STATE.setdefault(tenant_id, {})
         count = state.get(bucket, 0)
         if count >= _RATE_LIMIT_PER_MINUTE:
-            raise LimitExceededError(
-                "Query rate exceeded limit. Please try again later")
+            raise LimitExceededError("Query rate exceeded limit. Please try again later")
         state[bucket] = count + 1
         # cleanup old buckets, keep only current
         for b in list(state.keys()):
@@ -123,20 +120,17 @@ def _build_idempotency_key(*parts: Any) -> str:
 async def to_external_conversation_id(internal_id: int) -> str:
     if not internal_id:
         raise Exception("invalid internal conversation id")
-    external_id = get_external_id_by_internal(
-        internal_id=internal_id, mapping_type="CONVERSATION")
+    external_id = get_external_id_by_internal(internal_id=internal_id, mapping_type="CONVERSATION")
     if not external_id:
-        logger.error(
-            f"cannot find external id for conversation_id: {internal_id}")
+        logger.error(f"cannot find external id for conversation_id: {internal_id}")
         raise Exception("cannot find external id")
     return external_id
 
 
 async def to_internal_conversation_id(external_id: str) -> int:
     if not external_id:
         raise Exception("invalid external conversation id")
-    internal_id = get_internal_id_by_external(
-        external_id=external_id, mapping_type="CONVERSATION")
+    internal_id = get_internal_id_by_external(external_id=external_id, mapping_type="CONVERSATION")
     return internal_id
 
 
@@ -147,8 +141,7 @@ async def get_agent_info_by_name(agent_name: str, tenant_id: str) -> int:
     try:
         return await get_agent_id_by_name(agent_name=agent_name, tenant_id=tenant_id)
     except Exception as _:
-        raise Exception(
-            f"Failed to get agent id for agent_name: {agent_name} in tenant_id: {tenant_id}")
+        raise Exception(f"Failed to get agent id for agent_name: {agent_name} in tenant_id: {tenant_id}")
 
 
 async def start_streaming_chat(
@@ -166,22 +159,18 @@ async def start_streaming_chat(
         internal_conversation_id = await to_internal_conversation_id(external_conversation_id)
         # Add mapping to postgres database
         if internal_conversation_id is None:
-            logging.info(
-                f"Conversation {external_conversation_id} not found, creating a new conversation")
+            logging.info(f"Conversation {external_conversation_id} not found, creating a new conversation")
             # Create a new conversation and get its internal ID
-            new_conversation = create_new_conversation(
-                title="New Conversation", user_id=ctx.user_id)
+            new_conversation = create_new_conversation(title="New Conversation", user_id=ctx.user_id)
             internal_conversation_id = new_conversation["conversation_id"]
             # Add the new mapping to the database
-            add_mapping_id(internal_id=internal_conversation_id,
-                           external_id=external_conversation_id, tenant_id=ctx.tenant_id, user_id=ctx.user_id)
+            add_mapping_id(internal_id=internal_conversation_id, external_id=external_conversation_id, tenant_id=ctx.tenant_id, user_id=ctx.user_id)
 
         # Get history according to internal_conversation_id
         history = await get_conversation_history(ctx, external_conversation_id)
         agent_id = await get_agent_id_by_name(agent_name=agent_name, tenant_id=ctx.tenant_id)
         # Idempotency: only prevent concurrent duplicate starts
-        composed_key = idempotency_key or _build_idempotency_key(
-            ctx.tenant_id, external_conversation_id, agent_id, query)
+        composed_key = idempotency_key or _build_idempotency_key(ctx.tenant_id, external_conversation_id, agent_id, query)
         await idempotency_start(composed_key)
         agent_request = AgentRequest(
             conversation_id=internal_conversation_id,
@@ -193,13 +182,11 @@ async def start_streaming_chat(
         )
 
     except LimitExceededError as _:
-        raise LimitExceededError(
-            "Query rate exceeded limit. Please try again later.")
+        raise LimitExceededError("Query rate exceeded limit. Please try again later.")
     except UnauthorizedError as _:
         raise UnauthorizedError("Cannot authenticate.")
     except Exception as e:
-        raise Exception(
-            f"Failed to start streaming chat for external conversation id {external_conversation_id}: {str(e)}")
+        raise Exception(f"Failed to start streaming chat for external conversation id {external_conversation_id}: {str(e)}")
 
     try:
         response = await run_agent_stream(
@@ -226,8 +213,7 @@ async def stop_chat(ctx: NorthboundContext, external_conversation_id: str) -> Di
         stop_result = stop_agent_tasks(internal_id)
         return {"message": stop_result.get("message", "success"), "data": external_conversation_id, "requestId": ctx.request_id}
     except Exception as e:
-        raise Exception(
-            f"Failed to stop chat for external conversation id {external_conversation_id}: {str(e)}")
+        raise Exception(f"Failed to stop chat for external conversation id {external_conversation_id}: {str(e)}")
 
 
 async def list_conversations(ctx: NorthboundContext) -> Dict[str, Any]:
@@ -266,8 +252,7 @@ async def get_agent_info_list(ctx: NorthboundContext) -> Dict[str, Any]:
             agent_info.pop("agent_id", None)
         return {"message": "success", "data": agent_info_list, "requestId": ctx.request_id}
     except Exception as e:
-        raise Exception(
-            f"Failed to get agent info list for tenant {ctx.tenant_id}: {str(e)}")
+        raise Exception(f"Failed to get agent info list for tenant {ctx.tenant_id}: {str(e)}")
 
 
 async def update_conversation_title(ctx: NorthboundContext, external_conversation_id: str, title: str, idempotency_key: Optional[str] = None) -> Dict[str, Any]:
@@ -276,8 +261,7 @@ async def update_conversation_title(ctx: NorthboundContext, external_conversatio
         internal_id = await to_internal_conversation_id(external_conversation_id)
 
         # Idempotency: avoid concurrent duplicate title update for same conversation
-        composed_key = idempotency_key or _build_idempotency_key(
-            ctx.tenant_id, external_conversation_id, title)
+        composed_key = idempotency_key or _build_idempotency_key(ctx.tenant_id, external_conversation_id, title)
         await idempotency_start(composed_key)
 
         update_conversation_title_service(internal_id, title, ctx.user_id)
@@ -288,11 +272,9 @@ async def update_conversation_title(ctx: NorthboundContext, external_conversatio
             "idempotency_key": composed_key,
         }
     except LimitExceededError as _:
-        raise LimitExceededError(
-            "Duplicate request is still running, please wait.")
+        raise LimitExceededError("Duplicate request is still running, please wait.")
     except Exception as e:
-        raise Exception(
-            f"Failed to update conversation title for external conversation id {external_conversation_id}: {str(e)}")
+        raise Exception(f"Failed to update conversation title for external conversation id {external_conversation_id}: {str(e)}")
     finally:
         if composed_key:
             asyncio.create_task(_release_idempotency_after_delay(composed_key))