feat: 秘钥凭证管理

app/utils/api_secret_utils.py

@@ -0,0 +1,34 @@
+from enum import Enum
+
+
+class ApiSecretStatus(Enum):
+  valid = "valid"  # 秘钥有效
+  invalid = "invalid"  # 秘钥无效
+  not_exist = "not_exist"  # 秘钥不存在
+
+
+class ApiSecretUtils():
+  def __init__(self):
+    self.cache = {}
+
+  # 验证秘钥状态
+  async def verify_secret(self, secret: str) -> ApiSecretStatus:
+    api_secret_status = self.cache.get(secret, None)
+    print("verify_secret", self.cache)
+    return api_secret_status or ApiSecretStatus.not_exist
+
+  # 保存秘钥状态
+  async def save_secret(self, secret: str, status: ApiSecretStatus):
+    print("save_secret", self.cache)
+    self.cache[secret] = status
+
+  # 移除秘钥缓存
+  async def remove_secret(self, secret: str):
+    # self.cache.pop(secret, None)
+    if secret in self.cache:
+      del self.cache[secret]
+    print("remove_secret", secret)
+    print(await self.verify_secret(secret))
+
+
+api_secret_utils = ApiSecretUtils()

app/utils/create_module_service.py

@@ -309,11 +309,11 @@ def create_model_service(
       row_dict: dict = Body(..., description=f"插入的数据，字段参考{Cls.__name__}"),
       user: ModelPublicUser = None,
     ):
+      await self.check_insert_row_dict(row_dict, user)
+
       if before_insert is not None:
         await before_insert(row_dict, session)
 
-      await self.check_insert_row_dict(row_dict, user)
-
       try:
         # 使用模型类验证数据并创建实例（校验字段类型和约束）
         insert_cls = Cls.model_validate(row_dict)
@@ -342,9 +342,6 @@ def create_model_service(
       user: ModelPublicUser = None,
     ):
 
-      if before_batch_insert is not None:
-        await before_batch_insert(row_dict_list, session)
-
       # 筛选出没有id的记录（需要自动生成id）
       row_dict_list_without_id = []
 
@@ -363,6 +360,9 @@ def create_model_service(
       for row_dict in row_dict_list_without_id:
         await self.check_insert_row_dict(row_dict, user)
 
+      if before_batch_insert is not None:
+        await before_batch_insert(row_dict_list, session)
+
       try:
         # 验证所有记录并转换为模型实例列表
         insert_cls_list = [Cls.model_validate(row_dict) for row_dict in row_dict_list]