fix: 使用Union替代类型注解中的管道符号，使python3.9版本不报错

app/domain/gemini_models.py

@@ -45,7 +45,7 @@ class GenerationConfig(BaseModel):
 
 class SystemInstruction(BaseModel):
     role: str = "system"
-    parts: List[Dict[str, Any]] | Dict[str, Any]
+    parts: Union[List[Dict[str, Any]], Dict[str, Any]]
 
 
 class GeminiContent(BaseModel):

app/domain/image_models.py

@@ -1,5 +1,8 @@
+from typing import Union
+
+
 class ImageMetadata:
-    def __init__(self, width: int, height: int, filename: str, size: int, url: str, delete_url: str | None = None):
+    def __init__(self, width: int, height: int, filename: str, size: int, url: str, delete_url: Union[str, None] = None):
         self.width = width
         self.height = height
         self.filename = filename
@@ -19,5 +22,3 @@ class UploadResponse:
 class ImageUploader:
     def upload(self, file: bytes, filename: str) -> UploadResponse:
         raise NotImplementedError
-    
-    

app/service/key/key_manager.py

@@ -1,6 +1,6 @@
 import asyncio
 from itertools import cycle
-from typing import Dict
+from typing import Dict, Union
 
 from app.config.config import settings
 from app.log.logger import get_key_manager_logger
@@ -178,19 +178,20 @@ class KeyManager:
         if self.api_keys:
             return self.api_keys[0]
         if not self.api_keys:
-            logger.warning("API key list is empty, cannot get first valid key.")
+            logger.warning(
+                "API key list is empty, cannot get first valid key.")
             return ""
         return self.api_keys[0]
 
 
 _singleton_instance = None
 _singleton_lock = asyncio.Lock()
-_preserved_failure_counts: Dict[str, int] | None = None
+_preserved_failure_counts: Union[Dict[str, int], None] = None
-_preserved_vertex_failure_counts: Dict[str, int] | None = None
+_preserved_vertex_failure_counts: Union[Dict[str, int], None] = None
-_preserved_old_api_keys_for_reset: list | None = None
+_preserved_old_api_keys_for_reset: Union[list, None] = None
-_preserved_vertex_old_api_keys_for_reset: list | None = None
+_preserved_vertex_old_api_keys_for_reset: Union[list, None] = None
-_preserved_next_key_in_cycle: str | None = None
+_preserved_next_key_in_cycle: Union[str, None] = None
-_preserved_vertex_next_key_in_cycle: str | None = None
+_preserved_vertex_next_key_in_cycle: Union[str, None] = None
 
 
 async def get_key_manager_instance(
@@ -252,7 +253,8 @@ async def get_key_manager_instance(
                 _singleton_instance.vertex_key_failure_counts = (
                     current_vertex_failure_counts
                 )
-                logger.info("Inherited failure counts for applicable Vertex keys.")
+                logger.info(
+                    "Inherited failure counts for applicable Vertex keys.")
             _preserved_vertex_failure_counts = None
 
             # 2. 调整 key_cycle 的起始点
@@ -429,7 +431,8 @@ async def reset_key_manager_instance():
                 )
                 _preserved_next_key_in_cycle = None
             except Exception as e:
-                logger.error(f"Error preserving next key hint during reset: {e}")
+                logger.error(
+                    f"Error preserving next key hint during reset: {e}")
                 _preserved_next_key_in_cycle = None
 
             # 4. 保存 vertex_key_cycle 的下一个 key 提示
@@ -446,10 +449,10 @@ async def reset_key_manager_instance():
                 )
                 _preserved_vertex_next_key_in_cycle = None
             except Exception as e:
-                logger.error(f"Error preserving next key hint during reset: {e}")
+                logger.error(
+                    f"Error preserving next key hint during reset: {e}")
                 _preserved_vertex_next_key_in_cycle = None
 
-
             _singleton_instance = None
             logger.info(
                 "KeyManager instance has been reset. State (failure counts, old keys, next key hint) preserved for next instantiation."

app/service/stats/stats_service.py

@@ -1,6 +1,7 @@
 # app/service/stats_service.py
 
 import datetime
+from typing import Union
 
 from sqlalchemy import and_, case, func, or_, select
 
@@ -195,10 +196,11 @@ class StatsService:
             return details
 
         except Exception as e:
-            logger.error(f"Failed to get API call details for period '{period}': {e}")
+            logger.error(
+                f"Failed to get API call details for period '{period}': {e}")
             raise
 
-    async def get_key_usage_details_last_24h(self, key: str) -> dict | None:
+    async def get_key_usage_details_last_24h(self, key: str) -> Union[dict, None]:
         """
         获取指定 API 密钥在过去 24 小时内按模型统计的调用次数。
 
@@ -218,7 +220,8 @@ class StatsService:
         try:
             query = (
                 select(
-                    RequestLog.model_name, func.count(RequestLog.id).label("call_count")
+                    RequestLog.model_name, func.count(
+                        RequestLog.id).label("call_count")
                 )
                 .where(
                     RequestLog.api_key == key,
@@ -237,7 +240,8 @@ class StatsService:
                 )
                 return {}
 
-            usage_details = {row["model_name"]: row["call_count"] for row in results}
+            usage_details = {row["model_name"]: row["call_count"]
+                             for row in results}
             logger.info(
                 f"Successfully fetched usage details for key ending in ...{key[-4:]}: {usage_details}"
             )