Ignore AI Buddy downloaded static files (#1262)

* Optimize Redis connection pool and batch operations

* Refactor token session index and online status

README.md

@@ -44,6 +44,20 @@ springBoot...). ), but a self-righteous directory structure that you can give it
 For more details, please check
 the [official documentation](https://fastapi-practices.github.io/fastapi_best_architecture_docs/)
 
+## Sponsors
+
+<div align="center">
+  <table>
+    <tr>
+      <td align="center">
+        <a href="https://claude.uy/home">
+          <img src="https://purple-sun-4f5a.wuyao1243.workers.dev/" alt="Claude.uy" width="400">
+        </a>
+      </td>
+    </tr>
+  </table>
+</div>
+
 ## Contributors
 
 <a href="https://github.com/fastapi-practices/fastapi_best_architecture/graphs/contributors">

README.zh-CN.md

@@ -41,6 +41,20 @@ mvc 架构作为常规设计模式，在 python web 中很常见，但是三层
 
 更多详情请查看 [官方文档](https://fastapi-practices.github.io/fastapi_best_architecture_docs/)
 
+## 赞助商
+
+<div align="center">
+  <table>
+    <tr>
+      <td align="center">
+        <a href="https://claude.uy/home">
+          <img src="https://purple-sun-4f5a.wuyao1243.workers.dev/" alt="Claude.uy" width="400">
+        </a>
+      </td>
+    </tr>
+  </table>
+</div>
+
 ## 贡献者
 
 <a href="https://github.com/fastapi-practices/fastapi_best_architecture/graphs/contributors">

backend/.gitignore

@@ -2,5 +2,6 @@ __pycache__/
 .env
 alembic/versions/
 static/media/
+static/ai_buddy/
 *.log
 celerybeat-schedule.*

backend/app/admin/api/v1/monitor/online.py

@@ -1,82 +1,108 @@
 import json
 
-from typing import Annotated
+from typing import TYPE_CHECKING, Annotated, Any
 
 from fastapi import APIRouter, Path, Query
 
 from backend.app.admin.schema.token import GetTokenDetail
 from backend.common.enums import StatusType
+from backend.common.exception import errors
 from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
-from backend.common.security.jwt import DependsSuperUser, jwt_decode, revoke_token
+from backend.common.security.jwt import DependsSuperUser, jwt_decode
+from backend.common.security.token import revoke_token
 from backend.core.conf import settings
 from backend.database.redis import redis_client
 
+if TYPE_CHECKING:
+    from backend.common.dataclasses import TokenPayload
+
 router = APIRouter()
 
 
 @router.get('', summary='获取在线用户', dependencies=[DependsSuperUser])
-async def get_sessions(
+async def get_sessions(  # ruff:ignore[complex-structure]
     username: Annotated[str | None, Query(description='用户名')] = None,
 ) -> ResponseSchemaModel[list[GetTokenDetail]]:
-    token_keys = await redis_client.get_by_prefix(settings.TOKEN_REDIS_PREFIX)
+    users_key = f'{settings.TOKEN_SESSION_REDIS_PREFIX}:users'
-    online_clients = await redis_client.smembers(settings.TOKEN_ONLINE_REDIS_PREFIX)
+    user_ids = list(await redis_client.smembers(users_key))
-    data: list[GetTokenDetail] = []
+    if not user_ids:
-    if not token_keys:
+        return response_base.success(data=[])
-        return response_base.success(data=data)
 
-    def append_token_detail() -> None:
+    session_sets = await redis_client.smembers_many([
-        data.append(
+        f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}' for user_id in user_ids
-            token_detail.model_copy(
+    ])
-                update={
+    session_refs: list[tuple[str, str]] = []
-                    'username': extra_info.get('username', '未知'),
+    for user_id, members in zip(user_ids, session_sets, strict=True):
-                    'nickname': extra_info.get('nickname', '未知'),
+        session_refs.extend((user_id, session_uuid) for session_uuid in members)
-                    'ip': extra_info.get('ip', '未知'),
+    if not session_refs:
-                    'os': extra_info.get('os', '未知'),
+        await redis_client.srem(users_key, *user_ids)
-                    'browser': extra_info.get('browser', '未知'),
+        return response_base.success(data=[])
-                    'device': extra_info.get('device', '未知'),
-                    'last_login_time': extra_info.get('last_login_time', '未知'),
-                },
-            ),
-        )
 
-    token_values = await redis_client.mget(*token_keys)
+    tokens = await redis_client.mget_batched([
-    token_details: list[GetTokenDetail] = []
+        f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}' for user_id, session_uuid in session_refs
-    extra_info_keys: list[str] = []
+    ])
-    for token in token_values:
+    token_payloads: list[TokenPayload] = []
+    live_user_ids: set[str] = set()
+    for (user_id, _session_uuid), token in zip(session_refs, tokens, strict=True):
         if not token:
             continue
-        token_payload = jwt_decode(token)
+        try:
-        user_id = token_payload.user_id
+            token_payloads.append(jwt_decode(token))
-        session_uuid = token_payload.session_uuid
+        except errors.TokenError:
-        token_detail = GetTokenDetail(
+            continue
-            id=user_id,
+        live_user_ids.add(user_id)
-            session_uuid=session_uuid,
+    stale_user_ids = [user_id for user_id in user_ids if user_id not in live_user_ids]
-            username='未知',
+    if stale_user_ids:
-            nickname='未知',
+        await redis_client.srem(users_key, *stale_user_ids)
-            ip='未知',
+    if not token_payloads:
-            os='未知',
+        return response_base.success(data=[])
-            browser='未知',
-            device='未知',
-            status=StatusType.enable if session_uuid in online_clients else StatusType.disable,
-            last_login_time='未知',
-            expire_time=token_payload.expire_time,
-        )
-        token_details.append(token_detail)
-        extra_info_keys.append(f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{user_id}:{session_uuid}')
 
-    extra_infos = await redis_client.mget(*extra_info_keys) if extra_info_keys else []
+    extra_infos = await redis_client.mget_batched([
-    for token_detail, extra_info in zip(token_details, extra_infos, strict=True):
+        f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{item.user_id}:{item.session_uuid}' for item in token_payloads
+    ])
+    sid_sets = await redis_client.smembers_many([
+        f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:session:{item.session_uuid}' for item in token_payloads
+    ])
+    sid_list = [sid for members in sid_sets for sid in members]
+    sid_values = await redis_client.mget_batched([
+        f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{sid}' for sid in sid_list
+    ])
+    sid_session_map = dict(zip(sid_list, sid_values, strict=True))
+    online_sessions = {
+        item.session_uuid
+        for item, members in zip(token_payloads, sid_sets, strict=True)
+        if any(sid_session_map.get(sid) == item.session_uuid for sid in members)
+    }
+    data: list[GetTokenDetail] = []
+    for token_payload, extra_info in zip(token_payloads, extra_infos, strict=True):
+        info: dict[str, Any] = {}
         if extra_info:
-            extra_info = json.loads(extra_info)
+            try:
-            # 排除 swagger 登录生成的 token
+                parsed = json.loads(extra_info)
-            if extra_info.get('swagger') is None:
+            except (json.JSONDecodeError, TypeError):
-                if username is not None:
+                parsed = None
-                    if username == extra_info.get('username'):
+            if isinstance(parsed, dict):
-                        append_token_detail()
+                info = parsed
-                else:
+        if info.get('swagger') is not None:
-                    append_token_detail()
+            continue
-        else:
+        if username is not None and username != info.get('username'):
-            data.append(token_detail)
+            continue
+        data.append(
+            GetTokenDetail(
+                id=token_payload.user_id,
+                session_uuid=token_payload.session_uuid,
+                username=info.get('username', '未知'),
+                nickname=info.get('nickname', '未知'),
+                ip=info.get('ip', '未知'),
+                os=info.get('os', '未知'),
+                browser=info.get('browser', '未知'),
+                device=info.get('device', '未知'),
+                status=StatusType.enable if token_payload.session_uuid in online_sessions else StatusType.disable,
+                last_login_time=info.get('last_login_time', '未知'),
+                expire_time=token_payload.expire_time,
+            )
+        )
+    data.sort(key=lambda item: (item.id, item.session_uuid))
     return response_base.success(data=data)
 
 

backend/app/admin/crud/crud_opera_log.py

@@ -7,7 +7,7 @@ from backend.app.admin.model import OperaLog
 from backend.app.admin.schema.opera_log import CreateOperaLogParam
 
 
-class CRUDOperaLogDao(CRUDPlus[OperaLog]):
+class CRUDOperaLog(CRUDPlus[OperaLog]):
     """操作日志数据库操作类"""
 
     async def get_select(self, username: str | None, status: int | None, ip: str | None) -> Select:
@@ -71,4 +71,4 @@ class CRUDOperaLogDao(CRUDPlus[OperaLog]):
         await db.execute(sa_delete(OperaLog))
 
 
-opera_log_dao: CRUDOperaLogDao = CRUDOperaLogDao(OperaLog)
+opera_log_dao: CRUDOperaLog = CRUDOperaLog(OperaLog)

backend/app/admin/schema/data_scope.py

@@ -7,18 +7,18 @@ from backend.common.enums import StatusType
 from backend.common.schema import SchemaBase
 
 
-class DataScopeBase(SchemaBase):
+class DataScopeSchemaBase(SchemaBase):
     """数据范围基础模型"""
 
     name: str = Field(description='名称')
     status: StatusType = Field(description='状态')
 
 
-class CreateDataScopeParam(DataScopeBase):
+class CreateDataScopeParam(DataScopeSchemaBase):
     """创建数据范围参数"""
 
 
-class UpdateDataScopeParam(DataScopeBase):
+class UpdateDataScopeParam(DataScopeSchemaBase):
     """更新数据范围参数"""
 
 
@@ -41,7 +41,7 @@ class DeleteDataScopeParam(SchemaBase):
     pks: list[int] = Field(description='数据范围 ID 列表')
 
 
-class GetDataScopeDetail(DataScopeBase):
+class GetDataScopeDetail(DataScopeSchemaBase):
     """数据范围详情"""
 
     model_config = ConfigDict(from_attributes=True)

backend/app/admin/schema/user_password_history.py

@@ -3,12 +3,12 @@ from pydantic import Field
 from backend.common.schema import SchemaBase
 
 
-class UserPasswordHistoryBase(SchemaBase):
+class UserPasswordHistorySchemaBase(SchemaBase):
     """用户历史密码记录基础模型"""
 
     user_id: int = Field(description='用户 ID')
     password: str = Field(description='历史密码')
 
 
-class CreateUserPasswordHistoryParam(UserPasswordHistoryBase):
+class CreateUserPasswordHistoryParam(UserPasswordHistorySchemaBase):
     """创建用户历史密码记录"""

backend/app/admin/service/auth_service.py

@@ -17,12 +17,14 @@ from backend.common.exception import errors
 from backend.common.i18n import t
 from backend.common.log import log
 from backend.common.response.response_code import CustomErrorCode
-from backend.common.security.jwt import (
+from backend.common.security.jwt import jwt_decode
+from backend.common.security.token import (
     create_access_token,
     create_new_token,
     create_refresh_token,
     get_token,
-    jwt_decode,
+    get_user_sessions,
+    revoke_token,
 )
 from backend.core.conf import settings
 from backend.database.db import uuid4_str
@@ -35,7 +37,7 @@ class AuthService:
     """认证服务类"""
 
     @staticmethod
-    async def user_verify(db: AsyncSession, username: str, password: str) -> tuple[User, int | None]:
+    async def user_verify(*, db: AsyncSession, username: str, password: str) -> tuple[User, int | None]:
         """
         验证用户名和密码
 
@@ -47,9 +49,7 @@ class AuthService:
         user = await user_dao.get_by_username(db, username)
         if not user:
             raise errors.NotFoundError(msg='用户名或密码有误')
-
         await password_security_service.check_status(user.id, user.status)
-
         if user.password is None or not password_verify(password, user.password):
             await password_security_service.handle_login_failure(db, user.id)
             raise errors.AuthorizationError(msg='用户名或密码有误')
@@ -57,7 +57,6 @@ class AuthService:
         days_remaining = await password_security_service.check_password_expiry_status(
             db, user.last_password_changed_time
         )
-
         await password_security_service.handle_login_success(user.id)
 
         return user, days_remaining
@@ -70,12 +69,11 @@ class AuthService:
         :param obj: 登录凭证
         :return:
         """
-        user, _ = await self.user_verify(db, obj.username, obj.password)
+        user, _ = await self.user_verify(db=db, username=obj.username, password=obj.password)
         await user_dao.update_login_time(db, obj.username)
         access_token_data = await create_access_token(
             user.id,
             multi_login=user.is_multi_login,
-            # extra info
             swagger=True,
         )
         return access_token_data.access_token, user
@@ -103,14 +101,15 @@ class AuthService:
             if settings.LOGIN_CAPTCHA_ENABLED:
                 if not obj.uuid or not obj.captcha:
                     raise errors.RequestError(msg=t('error.captcha.invalid'))
-                captcha_code = await redis_client.get(f'{settings.LOGIN_CAPTCHA_REDIS_PREFIX}:{obj.uuid}')
+                captcha_key = f'{settings.LOGIN_CAPTCHA_REDIS_PREFIX}:{obj.uuid}'
+                captcha_code = await redis_client.get(captcha_key)
                 if not captcha_code:
                     raise errors.RequestError(msg=t('error.captcha.expired'))
                 if captcha_code.lower() != obj.captcha.lower():
                     raise errors.CustomError(error=CustomErrorCode.CAPTCHA_ERROR)
-                await redis_client.delete(f'{settings.LOGIN_CAPTCHA_REDIS_PREFIX}:{obj.uuid}')
+                await redis_client.delete(captcha_key)
 
-            user, days_remaining = await self.user_verify(db, obj.username, obj.password)
+            user, days_remaining = await self.user_verify(db=db, username=obj.username, password=obj.password)
             await user_dao.update_login_time(db, obj.username)
             await db.refresh(user)
             access_token_data = await create_access_token(
@@ -218,11 +217,9 @@ class AuthService:
             raise errors.NotFoundError(msg='用户不存在')
         if not user.status:
             raise errors.AuthorizationError(msg='用户已被锁定, 请联系统管理员')
-        token_keys = await redis_client.get_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}')
+        if not user.is_multi_login and await get_user_sessions(user.id) - {token_payload.session_uuid}:
-        if not user.is_multi_login and [
-            key for key in token_keys if not key.endswith(f':{token_payload.session_uuid}')
-        ]:
             raise errors.ForbiddenError(msg='此用户已在异地登录，请重新登录并及时修改密码')
+
         new_token = await create_new_token(
             refresh_token,
             token_payload.session_uuid,
@@ -244,6 +241,7 @@ class AuthService:
             expires=timezone.to_utc(new_token.new_refresh_token_expire_time),
             httponly=True,
         )
+
         data = GetNewToken(
             access_token=new_token.new_access_token,
             access_token_expire_time=new_token.new_access_token_expire_time,
@@ -263,18 +261,12 @@ class AuthService:
         try:
             token = get_token(request)
             token_payload = jwt_decode(token)
-            user_id = token_payload.user_id
-            session_uuid = token_payload.session_uuid
-            refresh_token = request.cookies.get(settings.COOKIE_REFRESH_TOKEN_KEY)
         except errors.TokenError:
             return
         finally:
             response.delete_cookie(settings.COOKIE_REFRESH_TOKEN_KEY)
 
-        await redis_client.delete(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}')
+        await revoke_token(token_payload.user_id, token_payload.session_uuid)
-        await redis_client.delete(f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{user_id}:{session_uuid}')
-        if refresh_token:
-            await redis_client.delete(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}')
 
 
 auth_service: AuthService = AuthService()

backend/app/admin/service/user_password_history_service.py

@@ -28,16 +28,15 @@ class UserPasswordHistoryService:
         if not user_status:
             raise errors.AuthorizationError(msg='用户已被锁定, 请联系统管理员')
 
-        locked_until_str = await redis_client.get(f'{settings.USER_LOCK_REDIS_PREFIX}:{user_id}')
+        lock_key = f'{settings.USER_LOCK_REDIS_PREFIX}:{user_id}'
-
+        locked_until_str = await redis_client.get(lock_key)
         if locked_until_str:
             locked_until = timezone.from_str(locked_until_str)
             now = timezone.now()
             if locked_until > now:
                 remaining_minutes = math.ceil((locked_until - now).total_seconds() / 60)
                 raise errors.AuthorizationError(msg=f'账号已被锁定，请在 {remaining_minutes} 分钟后重试')
-
+            await redis_client.delete(lock_key)
-            await redis_client.delete(f'{settings.USER_LOCK_REDIS_PREFIX}:{user_id}')
             await redis_client.delete(f'{settings.LOGIN_FAILURE_PREFIX}:{user_id}')
 
     @staticmethod
@@ -54,14 +53,11 @@ class UserPasswordHistoryService:
         if settings.USER_LOCK_THRESHOLD == 0:
             return
 
-        failure_count = await redis_client.get(f'{settings.LOGIN_FAILURE_PREFIX}:{user_id}')
+        failure_key = f'{settings.LOGIN_FAILURE_PREFIX}:{user_id}'
+        failure_count = await redis_client.get(failure_key)
         failure_count = int(failure_count) if failure_count else 0
         failure_count += 1
-        await redis_client.set(
+        await redis_client.set(failure_key, str(failure_count), ex=settings.USER_LOCK_SECONDS)
-            f'{settings.LOGIN_FAILURE_PREFIX}:{user_id}',
-            str(failure_count),
-            ex=settings.USER_LOCK_SECONDS,
-        )
 
         if failure_count >= settings.USER_LOCK_THRESHOLD:
             locked_until = timezone.now() + timedelta(seconds=settings.USER_LOCK_SECONDS)

backend/app/admin/service/user_service.py

@@ -21,7 +21,8 @@ from backend.common.enums import UserPermissionType
 from backend.common.exception import errors
 from backend.common.pagination import paging_data
 from backend.common.response.response_code import CustomErrorCode
-from backend.common.security.jwt import get_token, jwt_decode
+from backend.common.security.jwt import jwt_decode
+from backend.common.security.token import get_token, revoke_user_tokens
 from backend.core.conf import settings
 from backend.database.redis import redis_client
 from backend.utils.serializers import select_join_serialize
@@ -169,24 +170,19 @@ class UserService:
                 user = await user_dao.get(db, pk)
                 if not user:
                     raise errors.NotFoundError(msg='用户不存在')
-                multi_login = user.is_multi_login if pk != user.id else request.user.is_multi_login
+                multi_login = user.is_multi_login if pk != request.user.id else request.user.is_multi_login
                 new_multi_login = not multi_login
                 count = await user_dao.set_multi_login(db, pk, multi_login=new_multi_login)
                 token = get_token(request)
                 token_payload = jwt_decode(token)
-                if pk == user.id:
+                if pk == request.user.id:
                     # 系统管理员修改自身时，除当前 token 外，其他 token 失效
                     if not new_multi_login:
-                        key_prefix = f'{settings.TOKEN_REDIS_PREFIX}:{user.id}'
+                        await revoke_user_tokens(user.id, exclude_session_uuid=token_payload.session_uuid)
-                        await redis_client.delete_by_prefix(
-                            key_prefix,
-                            exclude_keys=f'{key_prefix}:{token_payload.session_uuid}',
-                        )
                 else:
                     # 系统管理员修改他人时，他人 token 全部失效
                     if not new_multi_login:
-                        key_prefix = f'{settings.TOKEN_REDIS_PREFIX}:{user.id}'
+                        await revoke_user_tokens(user.id)
-                        await redis_client.delete_by_prefix(key_prefix)
             case _:
                 raise errors.RequestError(msg='权限类型不存在')
 
@@ -209,13 +205,12 @@ class UserService:
 
         await validate_new_password(db, user.id, password)
         count = await user_dao.reset_password(db, user.id, password)
-
         history_obj = CreateUserPasswordHistoryParam(user_id=user.id, password=user.password)
         await password_security_service.save_password_history(db, history_obj)
         await user_dao.update_password_changed_time(db, user.id)
-        await redis_client.delete_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}')
+        await revoke_user_tokens(user.id)
-        await redis_client.delete_by_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user.id}')
+        await redis_client.delete(f'{settings.JWT_USER_REDIS_PREFIX}:{user.id}')
-        await redis_client.delete_by_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user.id}')
+
         return count
 
     @staticmethod
@@ -257,7 +252,8 @@ class UserService:
         :param email: 邮箱
         :return:
         """
-        captcha_code = await redis_client.get(f'{settings.EMAIL_CAPTCHA_REDIS_PREFIX}:{ctx.ip}')
+        captcha_key = f'{settings.EMAIL_CAPTCHA_REDIS_PREFIX}:{ctx.ip}'
+        captcha_code = await redis_client.get(captcha_key)
         if not captcha_code:
             raise errors.RequestError(msg='验证码已失效，请重新获取')
         if captcha != captcha_code:
@@ -265,7 +261,7 @@ class UserService:
         email_user = await user_dao.check_email(db, email)
         if email_user and email_user.id != user_id:
             raise errors.ConflictError(msg='邮箱已被绑定')
-        await redis_client.delete(f'{settings.EMAIL_CAPTCHA_REDIS_PREFIX}:{ctx.ip}')
+        await redis_client.delete(captcha_key)
         count = await user_dao.update_email(db, user_id, email)
         await redis_client.delete(f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}')
         return count
@@ -281,22 +277,19 @@ class UserService:
         :return:
         """
         user = await user_dao.get(db, user_id)
-
         if user.password and not password_verify(obj.old_password, user.password):
             raise errors.RequestError(msg='原密码错误')
-
         if obj.new_password != obj.confirm_password:
             raise errors.RequestError(msg='两次密码输入不一致')
 
         await validate_new_password(db, user_id, obj.new_password)
         count = await user_dao.reset_password(db, user_id, obj.new_password)
-
         history_obj = CreateUserPasswordHistoryParam(user_id=user.id, password=user.password)
         await password_security_service.save_password_history(db, history_obj)
         await user_dao.update_password_changed_time(db, user.id)
-        await redis_client.delete_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}')
+        await revoke_user_tokens(user_id)
-        await redis_client.delete_by_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}')
+        await redis_client.delete(f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}')
-        await redis_client.delete_by_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}')
+
         return count
 
     @staticmethod
@@ -311,10 +304,11 @@ class UserService:
         user = await user_dao.get(db, pk)
         if not user:
             raise errors.NotFoundError(msg='用户不存在')
+
         count = await user_dao.delete(db, user.id)
-        await redis_client.delete_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user.id}')
+        await revoke_user_tokens(user.id)
-        await redis_client.delete_by_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user.id}')
+        await redis_client.delete(f'{settings.JWT_USER_REDIS_PREFIX}:{user.id}')
-        await redis_client.delete_by_prefix(f'{settings.JWT_USER_REDIS_PREFIX}:{user.id}')
+
         return count
 
 

backend/app/task/api/v1/control.py

@@ -4,7 +4,7 @@ from fastapi import APIRouter, Depends, Path
 from starlette.concurrency import run_in_threadpool
 
 from backend.app.task import celery_app
-from backend.app.task.schema.control import TaskRegisteredDetail
+from backend.app.task.schema.control import GetTaskRegisteredDetail
 from backend.common.exception import errors
 from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
 from backend.common.security.jwt import DependsJwtAuth
@@ -15,7 +15,7 @@ router = APIRouter()
 
 
 @router.get('/registered', summary='获取已注册的任务', dependencies=[DependsJwtAuth])
-async def get_task_registered() -> ResponseSchemaModel[list[TaskRegisteredDetail]]:
+async def get_task_registered() -> ResponseSchemaModel[list[GetTaskRegisteredDetail]]:
     inspector = celery_app.control.inspect(timeout=0.5)
     registered = await run_in_threadpool(inspector.registered)
     if not registered:
@@ -26,9 +26,9 @@ async def get_task_registered() -> ResponseSchemaModel[list[TaskRegisteredDetail
         for task in tasks:
             task_ins = celery_app_tasks.get(task)
             if task_ins:
-                task_registered.append(TaskRegisteredDetail(name=task_ins.__doc__ or task, task=task))
+                task_registered.append(GetTaskRegisteredDetail(name=task_ins.__doc__ or task, task=task))
             else:
-                task_registered.append(TaskRegisteredDetail(name=task, task=task))
+                task_registered.append(GetTaskRegisteredDetail(name=task, task=task))
     return response_base.success(data=task_registered)
 
 

backend/app/task/crud/crud_scheduler.py

@@ -85,7 +85,7 @@ class CRUDTaskScheduler(CRUDPlus[TaskScheduler]):
         TaskScheduler.no_changes = False
         return 1
 
-    async def set_status(self, db: AsyncSession, pk: int, *, status: bool) -> int:
+    async def set_status(self, db: AsyncSession, pk: int, *, status: int) -> int:
         """
         设置任务调度状态
 
@@ -95,7 +95,7 @@ class CRUDTaskScheduler(CRUDPlus[TaskScheduler]):
         :return:
         """
         task_scheduler = await self.get(db, pk)
-        task_scheduler.enabled = status
+        task_scheduler.status = status
         TaskScheduler.no_changes = False
         return 1
 

backend/app/task/model/scheduler.py

@@ -1,12 +1,14 @@
 import asyncio
 
 from datetime import datetime
+from typing import ClassVar
 
 import sqlalchemy as sa
 
 from sqlalchemy import event
 from sqlalchemy.orm import Mapped, mapped_column
 
+from backend.common.enums import StatusType
 from backend.common.exception import errors
 from backend.common.model import Base, TimeZone, UniversalText, id_key
 from backend.core.conf import settings
@@ -39,12 +41,14 @@ class TaskScheduler(Base):
     interval_period: Mapped[str | None] = mapped_column(sa.String(256), comment='任务运行之间的周期类型')
     crontab: Mapped[str | None] = mapped_column(sa.String(64), default='* * * * *', comment='Crontab 表达式')
     one_off: Mapped[bool] = mapped_column(default=False, comment='是否仅运行一次')
-    enabled: Mapped[bool] = mapped_column(default=True, comment='是否启用任务')
+    status: Mapped[int] = mapped_column(default=StatusType.enable.value, comment='状态（0停用 1正常）')
     total_run_count: Mapped[int] = mapped_column(default=0, comment='任务触发的总次数')
     last_run_time: Mapped[datetime | None] = mapped_column(TimeZone, default=None, comment='任务最后触发的时间')
     remark: Mapped[str | None] = mapped_column(UniversalText, default=None, comment='备注')
 
     no_changes: bool = False
+    # 持有后台任务引用，避免 create_task 返回的任务在执行前被回收
+    _update_tasks: ClassVar[set[asyncio.Task]] = set()
 
     @staticmethod
     def before_insert_or_update(mapper, connection, target) -> None:  # ruff:ignore[missing-type-function-argument]
@@ -63,7 +67,9 @@ class TaskScheduler(Base):
 
     @classmethod
     def update_changed(cls, mapper, connection, target) -> None:  # ruff:ignore[missing-type-function-argument]
-        asyncio.create_task(cls.update_changed_async())
+        task = asyncio.create_task(cls.update_changed_async())
+        cls._update_tasks.add(task)
+        task.add_done_callback(cls._update_tasks.discard)
 
 
 # 事件监听器

backend/app/task/schema/control.py

@@ -1,6 +1,10 @@
+from pydantic import Field
+
 from backend.common.schema import SchemaBase
 
 
-class TaskRegisteredDetail(SchemaBase):
+class GetTaskRegisteredDetail(SchemaBase):
-    name: str
+    """已注册任务详情"""
-    task: str
+
+    name: str = Field(description='任务名称')
+    task: str = Field(description='任务函数')

backend/app/task/schema/scheduler.py

@@ -4,45 +4,46 @@ from pydantic import ConfigDict, Field
 from pydantic.types import JsonValue
 
 from backend.app.task.enums import PeriodType, TaskSchedulerType
+from backend.common.enums import StatusType
 from backend.common.schema import SchemaBase
 
 
-class TaskSchedulerSchemeBase(SchemaBase):
+class TaskSchedulerSchemaBase(SchemaBase):
     """任务调度参数"""
 
     name: str = Field(description='任务名称')
     task: str = Field(description='要运行的 Celery 任务')
-    args: JsonValue | None = Field(default=None, description='任务可接收的位置参数')
+    args: JsonValue | None = Field(None, description='任务可接收的位置参数')
-    kwargs: JsonValue | None = Field(default=None, description='任务可接收的关键字参数')
+    kwargs: JsonValue | None = Field(None, description='任务可接收的关键字参数')
-    queue: str | None = Field(default=None, description='CELERY_TASK_QUEUES 中定义的队列')
+    queue: str | None = Field(None, description='CELERY_TASK_QUEUES 中定义的队列')
-    exchange: str | None = Field(default=None, description='低级别 AMQP 路由的交换机')
+    exchange: str | None = Field(None, description='低级别 AMQP 路由的交换机')
-    routing_key: str | None = Field(default=None, description='低级别 AMQP 路由的路由密钥')
+    routing_key: str | None = Field(None, description='低级别 AMQP 路由的路由密钥')
-    start_time: datetime | None = Field(default=None, description='任务开始触发的时间')
+    start_time: datetime | None = Field(None, description='任务开始触发的时间')
-    expire_time: datetime | None = Field(default=None, description='任务不再触发的截止时间')
+    expire_time: datetime | None = Field(None, description='任务不再触发的截止时间')
-    expire_seconds: int | None = Field(default=None, description='任务不再触发的秒数时间差')
+    expire_seconds: int | None = Field(None, description='任务不再触发的秒数时间差')
     type: TaskSchedulerType = Field(description='任务调度类型（0间隔 1定时）')
-    interval_every: int | None = Field(default=None, description='任务再次运行前的间隔周期数')
+    interval_every: int | None = Field(None, description='任务再次运行前的间隔周期数')
-    interval_period: PeriodType | None = Field(default=None, description='任务运行之间的周期类型')
+    interval_period: PeriodType | None = Field(None, description='任务运行之间的周期类型')
     crontab: str = Field(default='* * * * *', description='Crontab 表达式')
     one_off: bool = Field(default=False, description='是否仅运行一次')
-    remark: str | None = Field(default=None, description='备注')
+    remark: str | None = Field(None, description='备注')
 
 
-class CreateTaskSchedulerParam(TaskSchedulerSchemeBase):
+class CreateTaskSchedulerParam(TaskSchedulerSchemaBase):
     """创建任务调度参数"""
 
 
-class UpdateTaskSchedulerParam(TaskSchedulerSchemeBase):
+class UpdateTaskSchedulerParam(TaskSchedulerSchemaBase):
     """更新任务调度参数"""
 
 
-class GetTaskSchedulerDetail(TaskSchedulerSchemeBase):
+class GetTaskSchedulerDetail(TaskSchedulerSchemaBase):
     """任务调度详情"""
 
     model_config = ConfigDict(from_attributes=True)
 
     id: int = Field(description='任务调度 ID')
-    enabled: bool = Field(description='是否启用任务')
+    status: StatusType = Field(description='状态')
     total_run_count: int = Field(description='已运行总次数')
     last_run_time: datetime | None = Field(None, description='最后运行时间')
     created_time: datetime = Field(description='创建时间')

backend/app/task/service/result_service.py

@@ -10,6 +10,8 @@ from backend.common.pagination import paging_data
 
 
 class TaskResultService:
+    """任务结果服务类"""
+
     @staticmethod
     async def get(*, db: AsyncSession, pk: int) -> TaskResult:
         """

backend/app/task/service/scheduler_service.py

@@ -12,6 +12,7 @@ from backend.app.task.enums import TaskSchedulerType
 from backend.app.task.model import TaskScheduler
 from backend.app.task.schema.scheduler import CreateTaskSchedulerParam, UpdateTaskSchedulerParam
 from backend.app.task.utils.tzcrontab import crontab_verify
+from backend.common.enums import StatusType
 from backend.common.exception import errors
 from backend.common.pagination import paging_data
 
@@ -110,7 +111,8 @@ class TaskSchedulerService:
         task_scheduler = await task_scheduler_dao.get(db, pk)
         if not task_scheduler:
             raise errors.NotFoundError(msg='任务调度不存在')
-        count = await task_scheduler_dao.set_status(db, pk, status=not task_scheduler.enabled)
+        next_status = StatusType.disable if task_scheduler.status == StatusType.enable else StatusType.enable
+        count = await task_scheduler_dao.set_status(db, pk, status=next_status)
         return count
 
     @staticmethod

backend/app/task/utils/schedulers.py

@@ -19,6 +19,7 @@ from backend.app.task.enums import PeriodType, TaskSchedulerType
 from backend.app.task.model.scheduler import TaskScheduler
 from backend.app.task.schema.scheduler import CreateTaskSchedulerParam
 from backend.app.task.utils.tzcrontab import TzAwareCrontab, crontab_verify
+from backend.common.enums import StatusType
 from backend.common.exception import errors
 from backend.core.conf import settings
 from backend.database.db import async_db_session
@@ -91,22 +92,24 @@ class ModelEntry(ScheduleEntry):
         self.last_run_at = timezone.from_datetime(model.last_run_time)
         self.options['periodic_task_name'] = model.name
         self.model = model
+        self.enabled = model.status == StatusType.enable
 
     async def _disable(self, model: TaskScheduler) -> None:
         """禁用任务"""
         model.no_changes = True
-        self.model.enabled = self.enabled = model.enabled = False
+        self.model.status = model.status = StatusType.disable
+        self.enabled = False
         async with async_db_session.begin() as db:
             stmt = select(TaskScheduler).where(TaskScheduler.id == model.id, TaskScheduler.deleted == 0)
             query = await db.execute(stmt)
             task = query.scalars().first()
             if task:
                 task.no_changes = True
-                task.enabled = False
+                task.status = StatusType.disable
 
     def is_due(self) -> tuple[bool, int | float | datetime]:
         """任务到期状态"""
-        if not self.model.enabled:
+        if self.model.status != StatusType.enable:
             # 重新启用时延迟 5 秒
             return schedules.schedstate(is_due=False, next=5)
 
@@ -119,11 +122,11 @@ class ModelEntry(ScheduleEntry):
                 return schedules.schedstate(is_due=False, next=delay)
 
         # 一次性任务
-        if self.model.one_off and self.model.enabled and self.model.total_run_count > 0:
+        if self.model.one_off and self.model.status == StatusType.enable and self.model.total_run_count > 0:
-            self.model.enabled = False
+            self.model.status = StatusType.disable
             self.model.total_run_count = 0
             self.model.no_changes = False
-            save_fields = ('enabled',)
+            save_fields = ('status',)
             run_await(self.save)(save_fields)
             return schedules.schedstate(is_due=False, next=1000000000)  # 高延迟，避免重新检查
 
@@ -237,6 +240,9 @@ class ModelEntry(ScheduleEntry):
             **cls._unpack_options(**options or {}),
             **entry,
         )
+        if 'enabled' in model_dict:
+            enabled = model_dict.pop('enabled')
+            model_dict['status'] = StatusType.enable if enabled else StatusType.disable
         return model_dict
 
     @classmethod
@@ -359,7 +365,7 @@ class DatabaseScheduler(Scheduler):
         try:
             for name, entry_fields in beat_dict.items():
                 entry = run_await(self.Entry.from_entry)(name, app=self.app, **entry_fields)
-                if entry.model.enabled:
+                if entry.model.status == StatusType.enable:
                     s[name] = entry
         except Exception:
             logger.error(f'添加任务 {name} 到数据库失败')
@@ -371,9 +377,10 @@ class DatabaseScheduler(Scheduler):
     def schedule_changed(self) -> bool | None:
         """任务调度变更状态"""
         now = timezone.now()
-        last_update = run_await(redis_client.get)(f'{settings.CELERY_REDIS_PREFIX}:last_update')
+        last_update_key = f'{settings.CELERY_REDIS_PREFIX}:last_update'
+        last_update = run_await(redis_client.get)(last_update_key)
         if not last_update:
-            run_await(redis_client.set)(f'{settings.CELERY_REDIS_PREFIX}:last_update', timezone.to_str(now))
+            run_await(redis_client.set)(last_update_key, timezone.to_str(now))
             return False
 
         last, ts = self._last_update, timezone.from_str(last_update)
@@ -388,7 +395,7 @@ class DatabaseScheduler(Scheduler):
         async with async_db_session() as db:
             logger.debug('DatabaseScheduler: Fetching database schedule')
             stmt = select(TaskScheduler).where(
-                TaskScheduler.enabled.is_(True),
+                TaskScheduler.status == StatusType.enable,
                 TaskScheduler.deleted == 0,
             )
             query = await db.execute(stmt)

backend/cli.py

@@ -261,8 +261,10 @@ async def init(db: AsyncSession, redis: RedisCli) -> None:
             for prefix in [
                 settings.JWT_USER_REDIS_PREFIX,
                 settings.TOKEN_EXTRA_INFO_REDIS_PREFIX,
+                settings.TOKEN_ONLINE_REDIS_PREFIX,
                 settings.TOKEN_REDIS_PREFIX,
                 settings.TOKEN_REFRESH_REDIS_PREFIX,
+                settings.TOKEN_SESSION_REDIS_PREFIX,
             ]:
                 await redis.delete_by_prefix(prefix)
 
@@ -318,6 +320,9 @@ def run(host: str, port: int, reload: bool, workers: int) -> None:  # ruff:ignor
     panel_content.append('\n🌐 架构官方文档: ', style='bold magenta')
     panel_content.append('https://docs.fba.wu-clan.cc/fastapi_best_architecture_docs/')
 
+    panel_content.append('\n\n赞助商：', style='bold yellow')
+    panel_content.append('Claude.uy', style='link https://claude.uy/home')
+
     console.print(Panel(panel_content, title=f'fba (v{__version__})', border_style='purple', padding=(1, 2)))
     granian.Granian(
         target='backend.main:app',
@@ -594,15 +599,15 @@ async def import_table(
         raise cappa.Exit('代码生成仅在开发环境可用', code=1)
 
     try:
-        from backend.plugin.code_generator.schema.gen import ImportParam
+        from backend.plugin.code_generator.schema.code_gen import ImportParam
-        from backend.plugin.code_generator.service.gen_service import gen_service
+        from backend.plugin.code_generator.service.code_gen_service import code_gen_service
     except ImportError:
         raise cappa.Exit('代码生成插件用法导入失败，请联系系统管理员', code=1)
 
     try:
         obj = ImportParam(app=app, table_schema=table_schema, table_name=table_name)
         async with async_db_session.begin() as db:
-            await gen_service.import_business_and_model(db=db, obj=obj)
+            await code_gen_service.import_business_and_model(db=db, obj=obj)
         console.tip('代码生成业务和模型列导入成功')
         console.log('\n快试试 [bold cyan]fba codegen[/bold cyan] 生成代码吧~')
     except Exception as e:
@@ -615,15 +620,15 @@ async def generate(*, preview: bool = False) -> None:
         raise cappa.Exit('代码生成仅在开发环境可用', code=1)
 
     try:
-        from backend.plugin.code_generator.service.business_service import gen_business_service
+        from backend.plugin.code_generator.service.business_service import code_gen_business_service
-        from backend.plugin.code_generator.service.gen_service import gen_service
+        from backend.plugin.code_generator.service.code_gen_service import code_gen_service
     except ImportError:
         raise cappa.Exit('代码生成插件用法导入失败，请联系系统管理员', code=1)
 
     try:
         ids = []
         async with async_db_session() as db:
-            results = await gen_business_service.get_all(db=db)
+            results = await code_gen_business_service.get_all(db=db)
 
         if not results:
             raise cappa.Exit('[red]暂无可用的代码生成业务！请先通过 import 命令导入！[/]')
@@ -648,7 +653,7 @@ async def generate(*, preview: bool = False) -> None:
 
         # 预览
         async with async_db_session() as db:
-            preview_data = await gen_service.preview(db=db, pk=business)
+            preview_data = await code_gen_service.preview(db=db, pk=business)
 
         console.print('\n[bold yellow]将要生成以下文件：[/]')
         file_table = Table(show_header=True, header_style='bold cyan')
@@ -672,7 +677,7 @@ async def generate(*, preview: bool = False) -> None:
 
         if ok.lower() == 'y':
             async with async_db_session.begin() as db:
-                gen_path = await gen_service.generate(db=db, pk=business)
+                gen_path = await code_gen_service.generate(db=db, pk=business)
 
             console.print()
             console.tip('代码已生成完成')

backend/common/cache/pubsub.py

@@ -38,26 +38,33 @@ class CachePubSubManager:
 
             try:
                 # 使用独立连接
-                pubsub_client = RedisCli(socket_timeout=None)
+                pubsub_client = RedisCli(max_connections=1)
                 pubsub = pubsub_client.pubsub()
                 await pubsub.subscribe(settings.CACHE_PUBSUB_CHANNEL)
 
                 # 发布订阅成功
                 reconnect_attempts = 0
 
-                async for message in pubsub.listen():
+                # 带超时轮询而不是 listen()，每次进入读取都会触发健康检查 PING，
-                    if message['type'] == 'message':
+                # 避免连接被静默断开后协程永久挂起
-                        try:
+                while True:
-                            data = json.loads(message['data'])
+                    message = await pubsub.get_message(
-                            cache_key = data['cache_key']
+                        ignore_subscribe_messages=True,
-                            if not data['delete_by_prefix']:
+                        timeout=settings.CACHE_PUBSUB_POLL_TIMEOUT,
-                                local_cache_manager.delete(cache_key)
+                    )
-                            else:
+                    if message is None or message['type'] != 'message':
-                                local_cache_manager.delete_by_prefix(cache_key)
+                        continue
-                        except json.JSONDecodeError as e:
+                    try:
-                            log.warning(f'[CachePubSub] 消息格式错误 {e}')
+                        data = json.loads(message['data'])
-                        except Exception as e:
+                        cache_key = data['cache_key']
-                            log.error(f'[CachePubSub] 处理通知失败: {e}')
+                        if not data['delete_by_prefix']:
+                            local_cache_manager.delete(cache_key)
+                        else:
+                            local_cache_manager.delete_by_prefix(cache_key)
+                    except json.JSONDecodeError as e:
+                        log.warning(f'[CachePubSub] 消息格式错误 {e}')
+                    except Exception as e:
+                        log.error(f'[CachePubSub] 处理通知失败: {e}')
 
             except asyncio.CancelledError:
                 break

backend/common/pagination.py

@@ -34,6 +34,12 @@ class _CustomPageParams(BaseModel, AbstractParams):
         )
 
 
+class _CustomCursorParams(CursorParams):
+    """自定义游标分页参数"""
+
+    size: int = Query(50, ge=0, le=200, description='每页数量')
+
+
 class _Links(BaseModel):
     """分页链接"""
 
@@ -98,7 +104,7 @@ class _CustomPage(_PageDetails, AbstractPage[T], Generic[T]):
 class _CustomCursorPage(_CursorPageDetails, AbstractPage[T], Generic[T]):
     """自定义游标分页类"""
 
-    __params_type__ = CursorParams
+    __params_type__ = _CustomCursorParams
 
     @classmethod
     def create(

backend/common/security/jwt.py

@@ -1,12 +1,7 @@
-import json
-import uuid
-
-from datetime import timedelta
 from typing import Annotated, Any
 
 from fastapi import Depends, Request
 from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
-from fastapi.security.utils import get_authorization_scheme_param
 from jose import ExpiredSignatureError, JWTError, jwt
 from pydantic_core import from_json
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -15,7 +10,7 @@ from starlette.authentication import UnauthenticatedUser
 from backend.app.admin.model import User
 from backend.app.admin.schema.user import GetUserInfoWithRelationDetail
 from backend.common.context import ctx
-from backend.common.dataclasses import AccessToken, NewToken, RefreshToken, TokenPayload
+from backend.common.dataclasses import TokenPayload
 from backend.common.exception import errors
 from backend.core.conf import settings
 from backend.database.db import async_db_session
@@ -63,132 +58,6 @@ def jwt_decode(token: str) -> TokenPayload:
     )
 
 
-async def create_access_token(user_id: int, *, multi_login: bool, **kwargs) -> AccessToken:
-    """
-    生成加密 token
-
-    :param user_id: 用户 ID
-    :param multi_login: 是否允许多端登录
-    :param kwargs: token 额外信息
-    :return:
-    """
-    expire = timezone.now() + timedelta(seconds=settings.TOKEN_EXPIRE_SECONDS)
-    session_uuid = str(uuid.uuid4())
-    access_token = jwt_encode({
-        'session_uuid': session_uuid,
-        'exp': timezone.to_utc(expire).timestamp(),
-        'sub': str(user_id),
-    })
-
-    if not multi_login:
-        await redis_client.delete_by_prefix(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}')
-
-    await redis_client.set(
-        f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}',
-        access_token,
-        ex=settings.TOKEN_EXPIRE_SECONDS,
-    )
-
-    # Token 附加信息单独存储
-    if kwargs:
-        await redis_client.set(
-            f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{user_id}:{session_uuid}',
-            json.dumps(kwargs, ensure_ascii=False),
-            ex=settings.TOKEN_EXPIRE_SECONDS,
-        )
-
-    return AccessToken(access_token=access_token, access_token_expire_time=expire, session_uuid=session_uuid)
-
-
-async def create_refresh_token(session_uuid: str, user_id: int, *, multi_login: bool) -> RefreshToken:
-    """
-    生成加密刷新 token，仅用于创建新的 token
-
-    :param session_uuid: 会话 UUID
-    :param user_id: 用户 ID
-    :param multi_login: 是否允许多端登录
-    :return:
-    """
-    expire = timezone.now() + timedelta(seconds=settings.TOKEN_REFRESH_EXPIRE_SECONDS)
-    refresh_token = jwt_encode({
-        'session_uuid': session_uuid,
-        'exp': timezone.to_utc(expire).timestamp(),
-        'sub': str(user_id),
-    })
-
-    if not multi_login:
-        await redis_client.delete_by_prefix(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}')
-
-    await redis_client.set(
-        f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}',
-        refresh_token,
-        ex=settings.TOKEN_REFRESH_EXPIRE_SECONDS,
-    )
-    return RefreshToken(refresh_token=refresh_token, refresh_token_expire_time=expire)
-
-
-async def create_new_token(
-    refresh_token: str,
-    session_uuid: str,
-    user_id: int,
-    *,
-    multi_login: bool,
-    **kwargs,
-) -> NewToken:
-    """
-    生成新的 token
-
-    :param refresh_token: 刷新 token
-    :param session_uuid: 会话 UUID
-    :param user_id: 用户 ID
-    :param multi_login: 是否允许多端登录
-    :param kwargs: token 附加信息
-    :return:
-    """
-    redis_refresh_token = await redis_client.get(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}')
-    if not redis_refresh_token or redis_refresh_token != refresh_token:
-        raise errors.TokenError(msg='Refresh Token 已过期，请重新登录')
-
-    await redis_client.delete(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}')
-    await redis_client.delete(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}')
-
-    new_access_token = await create_access_token(user_id, multi_login=multi_login, **kwargs)
-    new_refresh_token = await create_refresh_token(new_access_token.session_uuid, user_id, multi_login=multi_login)
-    return NewToken(
-        new_access_token=new_access_token.access_token,
-        new_access_token_expire_time=new_access_token.access_token_expire_time,
-        new_refresh_token=new_refresh_token.refresh_token,
-        new_refresh_token_expire_time=new_refresh_token.refresh_token_expire_time,
-        session_uuid=new_access_token.session_uuid,
-    )
-
-
-async def revoke_token(user_id: int, session_uuid: str) -> None:
-    """
-    撤销 token
-
-    :param user_id: 用户 ID
-    :param session_uuid: 会话 ID
-    :return:
-    """
-    await redis_client.delete(f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}')
-    await redis_client.delete(f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{user_id}:{session_uuid}')
-
-
-def get_token(request: Request) -> str:
-    """
-    获取请求头中的 token
-
-    :param request: FastAPI 请求对象
-    :return:
-    """
-    authorization = request.headers.get('Authorization')
-    scheme, token = get_authorization_scheme_param(authorization)
-    if not authorization or scheme.lower() != 'bearer':
-        raise errors.TokenError(msg='Token 无效')
-    return token
-
-
 async def get_current_user(db: AsyncSession, pk: int) -> User:
     """
     获取当前用户
@@ -219,16 +88,17 @@ async def get_jwt_user(user_id: int) -> GetUserInfoWithRelationDetail:
     """
     获取 JWT 用户
 
-    :param user_id:
+    :param user_id: 用户 ID
     :return:
     """
-    cache_user = await redis_client.get(f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}')
+    user_key = f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}'
+    cache_user = await redis_client.get(user_key)
     if not cache_user:
         async with async_db_session() as db:
             current_user = await get_current_user(db, user_id)
             user = GetUserInfoWithRelationDetail.model_validate(current_user)
             await redis_client.set(
-                f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}',
+                user_key,
                 user.model_dump_json(),
                 ex=settings.TOKEN_EXPIRE_SECONDS,
             )
@@ -251,7 +121,6 @@ async def jwt_authentication(token: str) -> GetUserInfoWithRelationDetail:
     redis_token = await redis_client.get(f'{settings.TOKEN_REDIS_PREFIX}:{ctx.user_id}:{token_payload.session_uuid}')
     if not redis_token:
         raise errors.TokenError(msg='Token 已过期')
-
     if token != redis_token:
         raise errors.TokenError(msg='Token 已失效')
 
@@ -292,7 +161,6 @@ def superuser_verify(request: Request, _token: str = DependsJwtAuth) -> bool:
     """
     if isinstance(request.user, UnauthenticatedUser):
         raise errors.TokenError
-
     superuser = request.user.is_superuser
     if not superuser or not request.user.is_staff:
         raise errors.AuthorizationError

backend/common/security/token.py

@@ -0,0 +1,329 @@
+import asyncio
+import json
+import uuid
+
+from datetime import timedelta
+from typing import Any
+
+from fastapi import Request
+from fastapi.security.utils import get_authorization_scheme_param
+
+from backend.common.dataclasses import AccessToken, NewToken, RefreshToken
+from backend.common.exception import errors
+from backend.common.security.jwt import jwt_encode
+from backend.core.conf import settings
+from backend.database.redis import redis_client
+from backend.utils.timezone import timezone
+
+_socket_disconnect_tasks: set[asyncio.Task[None]] = set()
+_REDIS_BATCH_SIZE = 1000
+
+
+def get_token(request: Request) -> str:
+    """
+    获取请求头中的 token
+
+    :param request: FastAPI 请求对象
+    :return:
+    """
+    authorization = request.headers.get('Authorization')
+    scheme, token = get_authorization_scheme_param(authorization)
+    if not authorization or scheme.lower() != 'bearer':
+        raise errors.TokenError(msg='Token 无效')
+    return token
+
+
+async def _srem_members(key: str, members: list[str] | set[str]) -> None:
+    """
+    分批从集合中移除成员
+
+    :param key: 集合 key
+    :param members: 要移除的成员
+    :return:
+    """
+    ordered = list(members)
+    if not ordered:
+        return
+    for index in range(0, len(ordered), _REDIS_BATCH_SIZE):
+        await redis_client.srem(key, *ordered[index : index + _REDIS_BATCH_SIZE])
+
+
+async def get_user_sessions(user_id: int) -> set[str]:
+    """
+    读取有效正式会话
+
+    :param user_id: 用户 ID
+    :return:
+    """
+    index_key = f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}'
+    users_key = f'{settings.TOKEN_SESSION_REDIS_PREFIX}:users'
+    swagger_key = f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}:swagger'
+    indexed = set(await redis_client.smembers(index_key))
+
+    live_sessions: set[str] = set()
+    swagger_sessions: set[str] = set()
+    if indexed:
+        ordered = list(indexed)
+        access_tokens, refresh_tokens, extras = await asyncio.gather(
+            redis_client.mget_batched([
+                f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}' for session_uuid in ordered
+            ]),
+            redis_client.mget_batched([
+                f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}' for session_uuid in ordered
+            ]),
+            redis_client.mget_batched([
+                f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{user_id}:{session_uuid}' for session_uuid in ordered
+            ]),
+        )
+        for session_uuid, access, refresh, extra in zip(ordered, access_tokens, refresh_tokens, extras, strict=True):
+            extra_info = None
+            if extra:
+                try:
+                    extra_info = json.loads(extra)
+                except (json.JSONDecodeError, TypeError):
+                    extra_info = None
+            if isinstance(extra_info, dict) and extra_info.get('swagger') is not None:
+                swagger_sessions.add(session_uuid)
+                continue
+            if access or refresh:
+                live_sessions.add(session_uuid)
+        if swagger_sessions:
+            await _srem_members(index_key, swagger_sessions)
+            async with redis_client.pipeline(transaction=False) as pipe:
+                pipe.sadd(swagger_key, *swagger_sessions)
+                pipe.expire(swagger_key, settings.TOKEN_EXPIRE_SECONDS)
+                await pipe.execute()
+
+    drop_from_index = indexed - live_sessions
+    session_ttl = max(settings.TOKEN_EXPIRE_SECONDS, settings.TOKEN_REFRESH_EXPIRE_SECONDS)
+    async with redis_client.pipeline(transaction=False) as pipe:
+        if drop_from_index:
+            pipe.srem(index_key, *drop_from_index)
+        if live_sessions:
+            pipe.expire(index_key, session_ttl)
+            pipe.sadd(users_key, str(user_id))
+        else:
+            pipe.delete(index_key)
+            pipe.srem(users_key, str(user_id))
+        await pipe.execute()
+    return live_sessions
+
+
+async def create_access_token(
+    user_id: int,
+    *,
+    multi_login: bool,
+    session_uuid: str | None = None,
+    swagger: bool = False,
+    **kwargs: Any,
+) -> AccessToken:
+    """
+    生成加密 token
+
+    :param user_id: 用户 ID
+    :param multi_login: 是否允许多端登录
+    :param session_uuid: 复用已有会话 UUID，刷新令牌时保持在线状态连续
+    :param swagger: 是否为 swagger 调试 token，不写入会话索引且不踢其他端
+    :param kwargs: token 额外信息
+    :return:
+    """
+    expire = timezone.now() + timedelta(seconds=settings.TOKEN_EXPIRE_SECONDS)
+    session_uuid = session_uuid or str(uuid.uuid4())
+    access_token = jwt_encode({
+        'session_uuid': session_uuid,
+        'jti': str(uuid.uuid4()),
+        'exp': timezone.to_utc(expire).timestamp(),
+        'sub': str(user_id),
+    })
+
+    if not swagger and not multi_login:
+        await revoke_user_tokens(user_id, exclude_session_uuid=session_uuid, include_swagger=False)
+
+    extra_info = {'swagger': True, **kwargs} if swagger else kwargs
+    extra_ttl = (
+        settings.TOKEN_EXPIRE_SECONDS
+        if swagger
+        else max(settings.TOKEN_EXPIRE_SECONDS, settings.TOKEN_REFRESH_EXPIRE_SECONDS)
+    )
+    session_ttl = max(settings.TOKEN_EXPIRE_SECONDS, settings.TOKEN_REFRESH_EXPIRE_SECONDS)
+    async with redis_client.pipeline(transaction=False) as pipe:
+        pipe.set(
+            f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}',
+            access_token,
+            ex=settings.TOKEN_EXPIRE_SECONDS,
+        )
+        if extra_info:
+            pipe.set(
+                f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{user_id}:{session_uuid}',
+                json.dumps(extra_info, ensure_ascii=False),
+                ex=extra_ttl,
+            )
+        if swagger:
+            swagger_key = f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}:swagger'
+            pipe.sadd(swagger_key, session_uuid)
+            pipe.expire(swagger_key, settings.TOKEN_EXPIRE_SECONDS)
+        else:
+            index_key = f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}'
+            pipe.sadd(index_key, session_uuid)
+            pipe.expire(index_key, session_ttl)
+            pipe.sadd(
+                f'{settings.TOKEN_SESSION_REDIS_PREFIX}:users',
+                str(user_id),
+            )
+        await pipe.execute()
+
+    return AccessToken(access_token=access_token, access_token_expire_time=expire, session_uuid=session_uuid)
+
+
+async def create_refresh_token(session_uuid: str, user_id: int, *, multi_login: bool) -> RefreshToken:
+    """
+    生成加密刷新 token，仅用于创建新的 token
+
+    :param session_uuid: 会话 UUID
+    :param user_id: 用户 ID
+    :param multi_login: 是否允许多端登录
+    :return:
+    """
+    expire = timezone.now() + timedelta(seconds=settings.TOKEN_REFRESH_EXPIRE_SECONDS)
+    refresh_token = jwt_encode({
+        'session_uuid': session_uuid,
+        'jti': str(uuid.uuid4()),
+        'exp': timezone.to_utc(expire).timestamp(),
+        'sub': str(user_id),
+    })
+
+    session_ttl = max(settings.TOKEN_EXPIRE_SECONDS, settings.TOKEN_REFRESH_EXPIRE_SECONDS)
+    index_key = f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}'
+    async with redis_client.pipeline(transaction=False) as pipe:
+        pipe.set(
+            f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}',
+            refresh_token,
+            ex=settings.TOKEN_REFRESH_EXPIRE_SECONDS,
+        )
+        pipe.sadd(index_key, session_uuid)
+        pipe.expire(index_key, session_ttl)
+        pipe.sadd(f'{settings.TOKEN_SESSION_REDIS_PREFIX}:users', str(user_id))
+        await pipe.execute()
+
+    return RefreshToken(refresh_token=refresh_token, refresh_token_expire_time=expire)
+
+
+async def create_new_token(
+    refresh_token: str,
+    session_uuid: str,
+    user_id: int,
+    *,
+    multi_login: bool,
+    **kwargs: Any,
+) -> NewToken:
+    """
+    生成新的 token
+
+    :param refresh_token: 刷新 token
+    :param session_uuid: 会话 UUID
+    :param user_id: 用户 ID
+    :param multi_login: 是否允许多端登录
+    :param kwargs: token 附加信息
+    :return:
+    """
+    redis_refresh_token = await redis_client.get(f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}')
+    if not redis_refresh_token or redis_refresh_token != refresh_token:
+        raise errors.TokenError(msg='Refresh Token 已过期，请重新登录')
+
+    new_access_token = await create_access_token(
+        user_id,
+        multi_login=multi_login,
+        session_uuid=session_uuid,
+        **kwargs,
+    )
+    new_refresh_token = await create_refresh_token(session_uuid, user_id, multi_login=multi_login)
+
+    return NewToken(
+        new_access_token=new_access_token.access_token,
+        new_access_token_expire_time=new_access_token.access_token_expire_time,
+        new_refresh_token=new_refresh_token.refresh_token,
+        new_refresh_token_expire_time=new_refresh_token.refresh_token_expire_time,
+        session_uuid=session_uuid,
+    )
+
+
+async def _revoke_sessions(user_id: int, session_uuids: set[str]) -> None:
+    """
+    批量删除会话相关 key，并异步断开 socket
+
+    :param user_id: 用户 ID
+    :param session_uuids: 要撤销的会话 UUID
+    :return:
+    """
+    if not session_uuids:
+        return
+
+    ordered = list(session_uuids)
+    sid_sets = await redis_client.smembers_many([
+        f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:session:{session_uuid}' for session_uuid in ordered
+    ])
+    sids = [sid for members in sid_sets for sid in members]
+    delete_keys: list[str] = []
+    for session_uuid in ordered:
+        delete_keys.extend((
+            f'{settings.TOKEN_REDIS_PREFIX}:{user_id}:{session_uuid}',
+            f'{settings.TOKEN_EXTRA_INFO_REDIS_PREFIX}:{user_id}:{session_uuid}',
+            f'{settings.TOKEN_REFRESH_REDIS_PREFIX}:{user_id}:{session_uuid}',
+            f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:session:{session_uuid}',
+        ))
+    delete_keys.extend(f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{sid}' for sid in sids)
+    await redis_client.delete_batched(delete_keys)
+    await asyncio.gather(
+        _srem_members(
+            f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}',
+            ordered,
+        ),
+        _srem_members(
+            f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}:swagger',
+            ordered,
+        ),
+    )
+    if not await redis_client.smembers(f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}'):
+        await redis_client.srem(f'{settings.TOKEN_SESSION_REDIS_PREFIX}:users', str(user_id))
+
+    if sids:
+        from backend.common.socketio.server import sio
+
+        for sid in sids:
+            for namespace in ('/', '/ws'):
+                task = sio.start_background_task(sio.disconnect, sid, namespace=namespace)
+                _socket_disconnect_tasks.add(task)
+                task.add_done_callback(_socket_disconnect_tasks.discard)
+
+
+async def revoke_token(user_id: int, session_uuid: str) -> None:
+    """
+    撤销 token
+
+    :param user_id: 用户 ID
+    :param session_uuid: 会话 ID
+    :return:
+    """
+    await _revoke_sessions(user_id, {session_uuid})
+
+
+async def revoke_user_tokens(
+    user_id: int,
+    *,
+    exclude_session_uuid: str | None = None,
+    include_swagger: bool = True,
+) -> None:
+    """
+    撤销用户全部会话，可保留当前会话
+
+    :param user_id: 用户 ID
+    :param exclude_session_uuid: 需要保留的会话 UUID
+    :param include_swagger: 是否同时撤销 swagger 调试 token
+    :return:
+    """
+    session_uuids = await get_user_sessions(user_id)
+    if include_swagger:
+        session_uuids |= set(await redis_client.smembers(f'{settings.TOKEN_SESSION_REDIS_PREFIX}:{user_id}:swagger'))
+    if exclude_session_uuid:
+        session_uuids.discard(exclude_session_uuid)
+    await _revoke_sessions(user_id, session_uuids)

backend/common/socketio/server.py

@@ -1,14 +1,17 @@
 import urllib.parse
 import uuid
 
+from typing import Any
+
 import socketio
 
 from starlette_context import request_cycle_context
 
 from backend.common.log import log
-from backend.common.security.jwt import jwt_authentication
+from backend.common.security.jwt import jwt_authentication, jwt_decode
 from backend.core.conf import settings
 from backend.database.redis import redis_client
+from backend.utils.timezone import timezone
 
 # 创建 Socket.IO 服务器实例
 sio = socketio.AsyncServer(
@@ -22,20 +25,29 @@ sio = socketio.AsyncServer(
     async_mode='asgi',
     cors_allowed_origins=settings.CORS_ALLOWED_ORIGINS,
     cors_credentials=True,
-    namespaces=['/ws'],
+    namespaces=['/', '/ws'],
 )
 
 
-@sio.event
+@sio.event(namespace='*')
-async def connect(sid, environ, auth) -> bool:
+async def connect(namespace: str, sid: str, _environ: dict[str, Any], auth: dict[str, Any] | None) -> bool:
-    """Socket 连接事件"""
+    """
-    if not auth:
+    Socket 连接事件
+
+    :param namespace: 命名空间
+    :param sid: 连接 ID
+    :param _environ: 连接环境
+    :param auth: 授权信息
+    :return:
+    """
+    if namespace not in sio.namespaces:
+        return False
+    if not isinstance(auth, dict):
         log.error('WebSocket 连接失败：无授权')
         return False
-
     session_uuid = auth.get('session_uuid')
     token = auth.get('token')
-    if not token or not session_uuid:
+    if not isinstance(token, str) or not token or not isinstance(session_uuid, str) or not session_uuid:
         log.error('WebSocket 连接失败：授权失败，请检查')
         return False
 
@@ -44,34 +56,63 @@ async def connect(sid, environ, auth) -> bool:
         if settings.ENVIRONMENT == 'prod':
             log.error('WebSocket 连接失败：生产环境禁止免授权直连')
             return False
-        await redis_client.set(f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{sid}', session_uuid)
+        expire = settings.TOKEN_EXPIRE_SECONDS
-        await redis_client.sadd(f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:session:{session_uuid}', sid)
+    else:
-        await redis_client.sadd(settings.TOKEN_ONLINE_REDIS_PREFIX, session_uuid)
+        try:
-        return True
+            with request_cycle_context({settings.TRACE_ID_REQUEST_HEADER_KEY: uuid.uuid4().hex}):
+                await jwt_authentication(token)
+            token_payload = jwt_decode(token)
+        except Exception as e:
+            log.info(f'WebSocket 连接失败：{e!s}')
+            return False
+        session_uuid = token_payload.session_uuid
+        expire = int((token_payload.expire_time - timezone.now()).total_seconds())
+        if expire <= 0:
+            log.info('WebSocket 连接失败：Token 已过期')
+            return False
 
-    try:
+    await sio.save_session(sid, {'session_uuid': session_uuid}, namespace=namespace)
-        with request_cycle_context({settings.TRACE_ID_REQUEST_HEADER_KEY: uuid.uuid4().hex}):
+    sid_key = f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{sid}'
-            await jwt_authentication(token)
+    session_key = f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:session:{session_uuid}'
-    except Exception as e:
+    await redis_client.set(sid_key, session_uuid, ex=expire)
-        log.info(f'WebSocket 连接失败：{e!s}')
+    await redis_client.sadd(session_key, sid)
-        return False
+    session_ttl = await redis_client.ttl(session_key)
-
+    # 新集合尚无 TTL 时按本次连接设置；多连接取最长剩余寿命
-    await redis_client.set(f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{sid}', session_uuid)
+    new_ttl = expire if session_ttl < 0 else max(session_ttl, expire)
-    await redis_client.sadd(f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:session:{session_uuid}', sid)
+    await redis_client.expire(session_key, new_ttl)
-    await redis_client.sadd(settings.TOKEN_ONLINE_REDIS_PREFIX, session_uuid)
     return True
 
 
-@sio.event
+@sio.event(namespace='*')
-async def disconnect(sid) -> None:
+async def disconnect(namespace: str, sid: str, _reason: str | None = None) -> None:
-    """Socket 断开连接事件"""
+    """
-    session_uuid = await redis_client.get(f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{sid}')
+    Socket 断开连接事件
+
+    :param namespace: 命名空间
+    :param sid: 连接 ID
+    :param _reason: 断开原因
+    :return:
+    """
+    sid_key = f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{sid}'
+    session_uuid = await redis_client.get(sid_key)
+    if not session_uuid:
+        try:
+            session_data = await sio.get_session(sid, namespace=namespace)
+        except KeyError:
+            return
+        session_uuid = session_data.get('session_uuid')
     if not session_uuid:
         return
 
     session_key = f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:session:{session_uuid}'
-    await redis_client.delete(f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{sid}')
+    await redis_client.delete(sid_key)
     await redis_client.srem(session_key, sid)
-    if await redis_client.scard(session_key) == 0:
+    remaining = list(await redis_client.smembers(session_key))
-        await redis_client.delete(session_key)
+    if not remaining:
-        await redis_client.srem(settings.TOKEN_ONLINE_REDIS_PREFIX, session_uuid)
+        return
+    mappings = await redis_client.mget_batched([
+        f'{settings.TOKEN_ONLINE_REDIS_PREFIX}:sid:{other_sid}' for other_sid in remaining
+    ])
+    stale_sids = [other_sid for other_sid, mapping in zip(remaining, mappings, strict=True) if mapping != session_uuid]
+    if stale_sids:
+        await redis_client.srem(session_key, *stale_sids)

backend/conftest.py

@@ -27,7 +27,7 @@ def client() -> Generator:
 
 
 @pytest.fixture(scope='module')
-def token_headers(client: TestClient) -> dict[str, str]:
+def token_headers(client: TestClient) -> Generator[dict[str, str], None, None]:
     params = {
         'username': PYTEST_USERNAME,
         'password': PYTEST_PASSWORD,
@@ -37,4 +37,5 @@ def token_headers(client: TestClient) -> dict[str, str]:
     token_type = response.json()['token_type']
     access_token = response.json()['access_token']
     headers = {'Authorization': f'{token_type} {access_token}'}
-    return headers
+    yield headers
+    client.post('/auth/logout', headers=headers)

backend/core/conf.py

@@ -68,6 +68,8 @@ class Settings(BaseSettings):
 
     # Redis
     REDIS_TIMEOUT: int = 5
+    REDIS_MAX_CONNECTIONS: int = 100  # 连接池上限
+    REDIS_POOL_TIMEOUT: int = 20  # 等待空闲连接超时（秒）
 
     # 缓存
     CACHE_LOCAL_ENABLED: bool = True
@@ -79,6 +81,7 @@ class Settings(BaseSettings):
     CACHE_PUBSUB_CHANNEL: str = 'fba:cache:invalidate'
     CACHE_PUBSUB_RECONNECT_DELAY: int = 5  # 重连延迟（秒）
     CACHE_PUBSUB_MAX_RECONNECT_ATTEMPTS: int = 10  # 最大重连次数
+    CACHE_PUBSUB_POLL_TIMEOUT: float = 1.0  # 订阅消息轮询超时（秒）
 
     # .env Snowflake
     SNOWFLAKE_ENABLED: bool = False
@@ -101,6 +104,7 @@ class Settings(BaseSettings):
     TOKEN_EXTRA_INFO_REDIS_PREFIX: str = 'fba:token_extra_info'
     TOKEN_ONLINE_REDIS_PREFIX: str = 'fba:token_online'
     TOKEN_REFRESH_REDIS_PREFIX: str = 'fba:refresh_token'
+    TOKEN_SESSION_REDIS_PREFIX: str = 'fba:token_session'
     TOKEN_REQUEST_UNDERLYING_SECURITY: bool = True
     TOKEN_REQUEST_PATH_EXCLUDE: list[str] = [  # JWT / RBAC 路由白名单
         f'{FASTAPI_API_V1_PATH}/auth/login',

backend/database/redis.py

@@ -1,6 +1,6 @@
 import sys
 
-from redis.asyncio import Redis
+from redis.asyncio import BlockingConnectionPool, Redis
 from redis.exceptions import AuthenticationError, TimeoutError
 
 from backend.common.log import log
@@ -22,6 +22,8 @@ class RedisCli(Redis):
         socket_keepalive: bool = True,
         health_check_interval: int = 30,
         decode_responses: bool = True,
+        max_connections: int = settings.REDIS_MAX_CONNECTIONS,
+        pool_timeout: int = settings.REDIS_POOL_TIMEOUT,
     ) -> None:
         """
         初始化 Redis 客户端
@@ -35,8 +37,12 @@ class RedisCli(Redis):
         :param socket_keepalive: 是否开启 TCP Keepalive 探测
         :param health_check_interval: 健康检查间隔时间（秒）
         :param decode_responses: 是否自动将 Redis 返回的字节流（bytes）解码为字符串（utf-8）
+        :param max_connections: 连接池最大连接数，超出后排队等待而不是无限新建
+        :param pool_timeout: 等待空闲连接的超时时间（秒）
         """
-        super().__init__(
+        pool = BlockingConnectionPool(
+            max_connections=max_connections,
+            timeout=pool_timeout,
             host=host,
             port=port,
             password=password,
@@ -47,6 +53,9 @@ class RedisCli(Redis):
             health_check_interval=health_check_interval,
             decode_responses=decode_responses,
         )
+        super().__init__(connection_pool=pool)
+        # 连接池由客户端独占，aclose 时一并释放
+        self.auto_close_connection_pool = True
 
     async def init(self) -> None:
         """初始化 Redis 服务器"""
@@ -67,13 +76,15 @@ class RedisCli(Redis):
         key_prefix: str,
         exclude_keys: str | list[str] | None = None,
         batch_size: int = 1000,
+        count: int = 1000,
     ) -> None:
         """
         删除指定前缀的所有 key
 
         :param key_prefix: 要删除的键前缀
         :param exclude_keys: 要排除的键或键列表
-        :param batch_size: 批量删除的大小，避免一次性删除过多键导致 Redis 阻塞
+        :param batch_size: 批量删除的大小
+        :param count: 每次扫描批次的数量
         :return:
         """
         exclude_set = (
@@ -84,22 +95,18 @@ class RedisCli(Redis):
             else set()
         )
         batch_keys = []
-
         if key_prefix not in exclude_set and await self.exists(key_prefix):
             batch_keys.append(key_prefix)
-
+        async for key in self.scan_iter(match=f'{key_prefix}:*', count=count):
-        async for key in self.scan_iter(match=f'{key_prefix}:*'):
             if key not in exclude_set:
                 batch_keys.append(key)
-
                 if len(batch_keys) >= batch_size:
                     await self.delete(*batch_keys)
                     batch_keys.clear()
-
         if batch_keys:
             await self.delete(*batch_keys)
 
-    async def get_by_prefix(self, key_prefix: str, count: int = 100) -> list[str]:
+    async def get_by_prefix(self, key_prefix: str, count: int = 1000) -> list[str]:
         """
         获取指定前缀的所有 key
 
@@ -109,6 +116,75 @@ class RedisCli(Redis):
         """
         return [key async for key in self.scan_iter(match=f'{key_prefix}:*', count=count)]
 
+    async def mget_batched(self, keys: list[str], batch_size: int = 1000) -> list[str | None]:
+        """
+        分批获取多个 key 的值
+
+        :param keys: 键列表
+        :param batch_size: 每批数量
+        :return:
+        """
+        if batch_size <= 0:
+            raise ValueError('batch_size 必须大于 0')
+        if not keys:
+            return []
+        values: list[str | None] = []
+        for index in range(0, len(keys), batch_size):
+            values.extend(await self.mget(keys[index : index + batch_size]))
+        return values
+
+    async def exists_batched(self, keys: list[str], batch_size: int = 1000) -> list[bool]:
+        """
+        分批判断多个 key 是否存在
+
+        :param keys: 键列表
+        :param batch_size: 每批数量
+        :return:
+        """
+        if batch_size <= 0:
+            raise ValueError('batch_size 必须大于 0')
+        return [value is not None for value in await self.mget_batched(keys, batch_size=batch_size)]
+
+    async def smembers_many(self, keys: list[str], batch_size: int = 100) -> list[set[str]]:
+        """
+        分批获取多个集合的成员
+
+        :param keys: 键列表
+        :param batch_size: 每批数量
+        :return:
+        """
+        if batch_size <= 0:
+            raise ValueError('batch_size 必须大于 0')
+        if not keys:
+            return []
+        members: list[set[str]] = []
+        for index in range(0, len(keys), batch_size):
+            batch = keys[index : index + batch_size]
+            async with self.pipeline(transaction=False) as pipe:
+                for key in batch:
+                    pipe.smembers(key)
+                results = await pipe.execute()
+            members.extend(set(result) if result else set() for result in results)
+        return members
+
+    async def delete_batched(self, keys: list[str], batch_size: int = 1000) -> int:
+        """
+        分批删除多个 key
+
+        :param keys: 键列表
+        :param batch_size: 每批数量
+        :return:
+        """
+        if batch_size <= 0:
+            raise ValueError('batch_size 必须大于 0')
+        if not keys:
+            return 0
+        deleted = 0
+        for index in range(0, len(keys), batch_size):
+            batch = keys[index : index + batch_size]
+            deleted += await self.delete(*batch)
+        return deleted
+
 
 # 创建 redis 客户端单例
 redis_client: RedisCli = RedisCli()

backend/plugin/code_generator/README.md

@@ -19,7 +19,7 @@
 CODE_GENERATOR_DOWNLOAD_ZIP_FILENAME = 'fba_generator'
 ```
 
-在 `backend/core/conf.py` 中添加以下内容：
+当前项目的 `backend/core/conf.py` 已包含以下字段：
 
 ```python
 ##################################################
@@ -28,6 +28,10 @@ CODE_GENERATOR_DOWNLOAD_ZIP_FILENAME = 'fba_generator'
 CODE_GENERATOR_DOWNLOAD_ZIP_FILENAME: str
 ```
 
+## 配置项说明
+
+- `CODE_GENERATOR_DOWNLOAD_ZIP_FILENAME`：控制代码生成结果下载压缩包的文件名
+
 ## 使用方式
 
 1. 安装并启用插件后，重启后端服务

backend/plugin/code_generator/api/router.py

@@ -2,11 +2,11 @@ from fastapi import APIRouter
 
 from backend.core.conf import settings
 from backend.plugin.code_generator.api.v1.business import router as business_router
+from backend.plugin.code_generator.api.v1.code_gen import router as code_gen_router
 from backend.plugin.code_generator.api.v1.column import router as column_router
-from backend.plugin.code_generator.api.v1.gen import router as gen_router
 
 v1 = APIRouter(prefix=f'{settings.FASTAPI_API_V1_PATH}/code-generation', tags=['代码生成'])
 
 v1.include_router(business_router, prefix='/businesses')
 v1.include_router(column_router, prefix='/columns')
-v1.include_router(gen_router, prefix='/generations')
+v1.include_router(code_gen_router, prefix='/generations')

backend/plugin/code_generator/api/v1/business.py

@@ -9,20 +9,20 @@ from backend.common.security.permission import RequestPermission
 from backend.common.security.rbac import DependsRBAC
 from backend.database.db import CurrentSession, CurrentSessionTransaction
 from backend.plugin.code_generator.schema.business import (
-    CreateGenBusinessParam,
+    CreateCodeGenBusinessParam,
-    GetGenBusinessDetail,
+    GetCodeGenBusinessDetail,
-    UpdateGenBusinessParam,
+    UpdateCodeGenBusinessParam,
 )
-from backend.plugin.code_generator.schema.column import GetGenColumnDetail
+from backend.plugin.code_generator.schema.column import GetCodeGenColumnDetail
-from backend.plugin.code_generator.service.business_service import gen_business_service
+from backend.plugin.code_generator.service.business_service import code_gen_business_service
-from backend.plugin.code_generator.service.column_service import gen_column_service
+from backend.plugin.code_generator.service.column_service import code_gen_column_service
 
 router = APIRouter()
 
 
 @router.get('/all', summary='获取所有代码生成业务', dependencies=[DependsJwtAuth])
-async def get_all_businesses(db: CurrentSession) -> ResponseSchemaModel[list[GetGenBusinessDetail]]:
+async def get_all_businesses(db: CurrentSession) -> ResponseSchemaModel[list[GetCodeGenBusinessDetail]]:
-    data = await gen_business_service.get_all(db=db)
+    data = await code_gen_business_service.get_all(db=db)
     return response_base.success(data=data)
 
 
@@ -30,8 +30,8 @@ async def get_all_businesses(db: CurrentSession) -> ResponseSchemaModel[list[Get
 async def get_business(
     db: CurrentSession,
     pk: Annotated[int, Path(description='业务 ID')],
-) -> ResponseSchemaModel[GetGenBusinessDetail]:
+) -> ResponseSchemaModel[GetCodeGenBusinessDetail]:
-    data = await gen_business_service.get(db=db, pk=pk)
+    data = await code_gen_business_service.get(db=db, pk=pk)
     return response_base.success(data=data)
 
 
@@ -46,8 +46,8 @@ async def get_business(
 async def get_businesses_paginated(
     db: CurrentSession,
     table_name: Annotated[str | None, Query(description='代码生成业务表名称')] = None,
-) -> ResponseSchemaModel[PageData[GetGenBusinessDetail]]:
+) -> ResponseSchemaModel[PageData[GetCodeGenBusinessDetail]]:
-    page_data = await gen_business_service.get_list(db=db, table_name=table_name)
+    page_data = await code_gen_business_service.get_list(db=db, table_name=table_name)
     return response_base.success(data=page_data)
 
 
@@ -55,8 +55,8 @@ async def get_businesses_paginated(
 async def get_business_all_columns(
     db: CurrentSession,
     pk: Annotated[int, Path(description='业务 ID')],
-) -> ResponseSchemaModel[list[GetGenColumnDetail]]:
+) -> ResponseSchemaModel[list[GetCodeGenColumnDetail]]:
-    data = await gen_column_service.get_columns(db=db, business_id=pk)
+    data = await code_gen_column_service.get_columns(db=db, business_id=pk)
     return response_base.success(data=data)
 
 
@@ -68,8 +68,8 @@ async def get_business_all_columns(
         DependsRBAC,
     ],
 )
-async def create_business(db: CurrentSessionTransaction, obj: CreateGenBusinessParam) -> ResponseModel:
+async def create_business(db: CurrentSessionTransaction, obj: CreateCodeGenBusinessParam) -> ResponseModel:
-    await gen_business_service.create(db=db, obj=obj)
+    await code_gen_business_service.create(db=db, obj=obj)
     return response_base.success()
 
 
@@ -84,9 +84,9 @@ async def create_business(db: CurrentSessionTransaction, obj: CreateGenBusinessP
 async def update_business(
     db: CurrentSessionTransaction,
     pk: Annotated[int, Path(description='业务 ID')],
-    obj: UpdateGenBusinessParam,
+    obj: UpdateCodeGenBusinessParam,
 ) -> ResponseModel:
-    count = await gen_business_service.update(db=db, pk=pk, obj=obj)
+    count = await code_gen_business_service.update(db=db, pk=pk, obj=obj)
     if count > 0:
         return response_base.success()
     return response_base.fail()
@@ -103,7 +103,7 @@ async def update_business(
 async def delete_business(
     db: CurrentSessionTransaction, pk: Annotated[int, Path(description='业务 ID')]
 ) -> ResponseModel:
-    count = await gen_business_service.delete(db=db, pk=pk)
+    count = await code_gen_business_service.delete(db=db, pk=pk)
     if count > 0:
         return response_base.success()
     return response_base.fail()

backend/plugin/code_generator/api/v1/code_gen.py

@@ -9,8 +9,8 @@ from backend.common.security.permission import RequestPermission
 from backend.common.security.rbac import DependsRBAC
 from backend.core.conf import settings
 from backend.database.db import CurrentSession, CurrentSessionTransaction
-from backend.plugin.code_generator.schema.gen import ImportParam
+from backend.plugin.code_generator.schema.code_gen import ImportParam
-from backend.plugin.code_generator.service.gen_service import gen_service
+from backend.plugin.code_generator.service.code_gen_service import code_gen_service
 
 router = APIRouter()
 
@@ -20,7 +20,7 @@ async def get_all_tables(
     db: CurrentSession,
     table_schema: Annotated[str, Query(description='数据库名')] = 'fba',
 ) -> ResponseSchemaModel[list[dict[str, str | None]]]:
-    data = await gen_service.get_tables(db=db, table_schema=table_schema)
+    data = await code_gen_service.get_tables(db=db, table_schema=table_schema)
     return response_base.success(data=data)
 
 
@@ -33,7 +33,7 @@ async def get_all_tables(
     ],
 )
 async def import_table(db: CurrentSessionTransaction, obj: ImportParam) -> ResponseModel:
-    await gen_service.import_business_and_model(db=db, obj=obj)
+    await code_gen_service.import_business_and_model(db=db, obj=obj)
     return response_base.success()
 
 
@@ -41,7 +41,7 @@ async def import_table(db: CurrentSessionTransaction, obj: ImportParam) -> Respo
 async def preview_code(
     db: CurrentSession, pk: Annotated[int, Path(description='业务 ID')]
 ) -> ResponseSchemaModel[dict[str, bytes]]:
-    data = await gen_service.preview(db=db, pk=pk)
+    data = await code_gen_service.preview(db=db, pk=pk)
     return response_base.success(data=data)
 
 
@@ -49,7 +49,7 @@ async def preview_code(
 async def get_generate_paths(
     db: CurrentSession, pk: Annotated[int, Path(description='业务 ID')]
 ) -> ResponseSchemaModel[list[str]]:
-    data = await gen_service.get_generate_path(db=db, pk=pk)
+    data = await code_gen_service.get_generate_path(db=db, pk=pk)
     return response_base.success(data=data)
 
 
@@ -63,13 +63,13 @@ async def get_generate_paths(
     ],
 )
 async def generate_code(db: CurrentSession, pk: Annotated[int, Path(description='业务 ID')]) -> ResponseModel:
-    await gen_service.generate(db=db, pk=pk)
+    await code_gen_service.generate(db=db, pk=pk)
     return response_base.success()
 
 
 @router.get('/{pk}', summary='下载代码', dependencies=[DependsJwtAuth])
 async def download_code(db: CurrentSession, pk: Annotated[int, Path(description='业务 ID')]):  # ruff:ignore[missing-return-type-undocumented-public-function]
-    bio = await gen_service.download(db=db, pk=pk)
+    bio = await code_gen_service.download(db=db, pk=pk)
     return StreamingResponse(
         bio,
         media_type='application/x-zip-compressed',

backend/plugin/code_generator/api/v1/column.py

@@ -8,26 +8,26 @@ from backend.common.security.permission import RequestPermission
 from backend.common.security.rbac import DependsRBAC
 from backend.database.db import CurrentSession, CurrentSessionTransaction
 from backend.plugin.code_generator.schema.column import (
-    CreateGenColumnParam,
+    CreateCodeGenColumnParam,
-    GetGenColumnDetail,
+    GetCodeGenColumnDetail,
-    UpdateGenColumnParam,
+    UpdateCodeGenColumnParam,
 )
-from backend.plugin.code_generator.service.column_service import gen_column_service
+from backend.plugin.code_generator.service.column_service import code_gen_column_service
 
 router = APIRouter()
 
 
 @router.get('/types', summary='获取代码生成模型列类型', dependencies=[DependsJwtAuth])
 async def get_column_types() -> ResponseSchemaModel[list[str]]:
-    column_types = await gen_column_service.get_types()
+    column_types = await code_gen_column_service.get_types()
     return response_base.success(data=column_types)
 
 
 @router.get('/{pk}', summary='获取代码生成模型列详情', dependencies=[DependsJwtAuth])
 async def get_column(
     db: CurrentSession, pk: Annotated[int, Path(description='模型列 ID')]
-) -> ResponseSchemaModel[GetGenColumnDetail]:
+) -> ResponseSchemaModel[GetCodeGenColumnDetail]:
-    data = await gen_column_service.get(db=db, pk=pk)
+    data = await code_gen_column_service.get(db=db, pk=pk)
     return response_base.success(data=data)
 
 
@@ -39,8 +39,8 @@ async def get_column(
         DependsRBAC,
     ],
 )
-async def create_column(db: CurrentSessionTransaction, obj: CreateGenColumnParam) -> ResponseModel:
+async def create_column(db: CurrentSessionTransaction, obj: CreateCodeGenColumnParam) -> ResponseModel:
-    await gen_column_service.create(db=db, obj=obj)
+    await code_gen_column_service.create(db=db, obj=obj)
     return response_base.success()
 
 
@@ -53,9 +53,9 @@ async def create_column(db: CurrentSessionTransaction, obj: CreateGenColumnParam
     ],
 )
 async def update_column(
-    db: CurrentSessionTransaction, pk: Annotated[int, Path(description='模型列 ID')], obj: UpdateGenColumnParam
+    db: CurrentSessionTransaction, pk: Annotated[int, Path(description='模型列 ID')], obj: UpdateCodeGenColumnParam
 ) -> ResponseModel:
-    count = await gen_column_service.update(db=db, pk=pk, obj=obj)
+    count = await code_gen_column_service.update(db=db, pk=pk, obj=obj)
     if count > 0:
         return response_base.success()
     return response_base.fail()
@@ -72,7 +72,7 @@ async def update_column(
 async def delete_column(
     db: CurrentSessionTransaction, pk: Annotated[int, Path(description='模型列 ID')]
 ) -> ResponseModel:
-    count = await gen_column_service.delete(db=db, pk=pk)
+    count = await code_gen_column_service.delete(db=db, pk=pk)
     if count > 0:
         return response_base.success()
     return response_base.fail()

backend/plugin/code_generator/crud/crud_business.py

@@ -4,15 +4,15 @@ from sqlalchemy import Select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy_crud_plus import CRUDPlus
 
-from backend.plugin.code_generator.model import GenBusiness
+from backend.plugin.code_generator.model import CodeGenBusiness
-from backend.plugin.code_generator.schema.business import CreateGenBusinessParam, UpdateGenBusinessParam
+from backend.plugin.code_generator.schema.business import CreateCodeGenBusinessParam, UpdateCodeGenBusinessParam
 from backend.utils.timezone import timezone
 
 
-class CRUDGenBusiness(CRUDPlus[GenBusiness]):
+class CRUDCodeGenBusiness(CRUDPlus[CodeGenBusiness]):
     """代码生成业务 CRUD 类"""
 
-    async def get(self, db: AsyncSession, pk: int) -> GenBusiness | None:
+    async def get(self, db: AsyncSession, pk: int) -> CodeGenBusiness | None:
         """
         获取代码生成业务
 
@@ -22,7 +22,7 @@ class CRUDGenBusiness(CRUDPlus[GenBusiness]):
         """
         return await self.select_model(db, pk, deleted=0)
 
-    async def get_by_name(self, db: AsyncSession, name: str) -> GenBusiness | None:
+    async def get_by_name(self, db: AsyncSession, name: str) -> CodeGenBusiness | None:
         """
         通过 name 获取代码生成业务
 
@@ -32,7 +32,7 @@ class CRUDGenBusiness(CRUDPlus[GenBusiness]):
         """
         return await self.select_model_by_column(db, table_name=name, deleted=0)
 
-    async def get_all(self, db: AsyncSession) -> Sequence[GenBusiness]:
+    async def get_all(self, db: AsyncSession) -> Sequence[CodeGenBusiness]:
         """
         获取所有代码生成业务
 
@@ -55,7 +55,7 @@ class CRUDGenBusiness(CRUDPlus[GenBusiness]):
 
         return await self.select_order('id', 'desc', **filters)
 
-    async def create(self, db: AsyncSession, obj: CreateGenBusinessParam) -> None:
+    async def create(self, db: AsyncSession, obj: CreateCodeGenBusinessParam) -> None:
         """
         创建代码生成业务
 
@@ -65,7 +65,7 @@ class CRUDGenBusiness(CRUDPlus[GenBusiness]):
         """
         await self.create_model(db, obj)
 
-    async def update(self, db: AsyncSession, pk: int, obj: UpdateGenBusinessParam) -> int:
+    async def update(self, db: AsyncSession, pk: int, obj: UpdateCodeGenBusinessParam) -> int:
         """
         更新代码生成业务
 
@@ -96,4 +96,4 @@ class CRUDGenBusiness(CRUDPlus[GenBusiness]):
         )
 
 
-gen_business_dao: CRUDGenBusiness = CRUDGenBusiness(GenBusiness)
+code_gen_business_dao: CRUDCodeGenBusiness = CRUDCodeGenBusiness(CodeGenBusiness)

backend/plugin/code_generator/crud/crud_code_gen.py

@@ -7,7 +7,7 @@ from backend.common.enums import DataBaseType
 from backend.core.conf import settings
 
 
-class CRUDGen:
+class CRUDCodeGen:
     """代码生成 CRUD 类"""
 
     @staticmethod
@@ -194,4 +194,4 @@ class CRUDGen:
         return result.mappings().all()
 
 
-gen_dao: CRUDGen = CRUDGen()
+code_gen_dao: CRUDCodeGen = CRUDCodeGen()

backend/plugin/code_generator/crud/crud_column.py

@@ -3,18 +3,18 @@ from collections.abc import Sequence
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy_crud_plus import CRUDPlus
 
-from backend.plugin.code_generator.model import GenColumn
+from backend.plugin.code_generator.model import CodeGenColumn
 from backend.plugin.code_generator.schema.column import (
-    CreateGenColumnInternalParam,
+    CreateCodeGenColumnInternalParam,
-    CreateGenColumnParam,
+    CreateCodeGenColumnParam,
-    UpdateGenColumnParam,
+    UpdateCodeGenColumnParam,
 )
 
 
-class CRUDGenColumn(CRUDPlus[GenColumn]):
+class CRUDCodeGenColumn(CRUDPlus[CodeGenColumn]):
     """代码生成模型列 CRUD 类"""
 
-    async def get(self, db: AsyncSession, pk: int) -> GenColumn | None:
+    async def get(self, db: AsyncSession, pk: int) -> CodeGenColumn | None:
         """
         获取代码生成模型列
 
@@ -24,7 +24,7 @@ class CRUDGenColumn(CRUDPlus[GenColumn]):
         """
         return await self.select_model(db, pk)
 
-    async def get_all_by_business(self, db: AsyncSession, business_id: int) -> Sequence[GenColumn]:
+    async def get_all_by_business(self, db: AsyncSession, business_id: int) -> Sequence[CodeGenColumn]:
         """
         获取所有代码生成模型列
 
@@ -32,9 +32,9 @@ class CRUDGenColumn(CRUDPlus[GenColumn]):
         :param business_id: 业务 ID
         :return:
         """
-        return await self.select_models_order(db, sort_columns='sort', gen_business_id=business_id)
+        return await self.select_models_order(db, sort_columns='sort', code_gen_business_id=business_id)
 
-    async def create(self, db: AsyncSession, obj: CreateGenColumnParam, pd_type: str | None) -> None:
+    async def create(self, db: AsyncSession, obj: CreateCodeGenColumnParam, pd_type: str | None) -> None:
         """
         创建代码生成模型列
 
@@ -45,7 +45,7 @@ class CRUDGenColumn(CRUDPlus[GenColumn]):
         """
         await self.create_model(db, obj, pd_type=pd_type)
 
-    async def bulk_create(self, db: AsyncSession, objs: list[CreateGenColumnInternalParam]) -> None:
+    async def bulk_create(self, db: AsyncSession, objs: list[CreateCodeGenColumnInternalParam]) -> None:
         """
         批量创建代码生成模型列
 
@@ -55,7 +55,7 @@ class CRUDGenColumn(CRUDPlus[GenColumn]):
         """
         await self.create_models(db, objs)
 
-    async def update(self, db: AsyncSession, pk: int, obj: UpdateGenColumnParam, pd_type: str | None) -> int:
+    async def update(self, db: AsyncSession, pk: int, obj: UpdateCodeGenColumnParam, pd_type: str | None) -> int:
         """
         更新代码生成模型列
 
@@ -78,4 +78,4 @@ class CRUDGenColumn(CRUDPlus[GenColumn]):
         return await self.delete_model(db, pk)
 
 
-gen_column_dao: CRUDGenColumn = CRUDGenColumn(GenColumn)
+code_gen_column_dao: CRUDCodeGenColumn = CRUDCodeGenColumn(CodeGenColumn)

backend/plugin/code_generator/model/__init__.py

@@ -1,2 +1,2 @@
-from backend.plugin.code_generator.model.business import GenBusiness as GenBusiness
+from backend.plugin.code_generator.model.business import CodeGenBusiness as CodeGenBusiness
-from backend.plugin.code_generator.model.column import GenColumn as GenColumn
+from backend.plugin.code_generator.model.column import CodeGenColumn as CodeGenColumn

backend/plugin/code_generator/model/business.py

@@ -5,12 +5,12 @@ from sqlalchemy.orm import Mapped, mapped_column
 from backend.common.model import Base, UniversalText, id_key
 
 
-class GenBusiness(Base):
+class CodeGenBusiness(Base):
     """代码生成业务表"""
 
-    __tablename__ = 'gen_business'
+    __tablename__ = 'code_gen_business'
     __table_args__ = (
-        sa.UniqueConstraint('table_name', 'deleted', name='uk_gen_business_table_name_deleted'),
+        sa.UniqueConstraint('table_name', 'deleted', name='uk_code_gen_business_table_name_deleted'),
         {'comment': '代码生成业务表'},
     )
 

backend/plugin/code_generator/model/column.py

@@ -5,12 +5,12 @@ from sqlalchemy.orm import Mapped, mapped_column
 from backend.common.model import DataClassBase, UniversalText, id_key
 
 
-class GenColumn(DataClassBase):
+class CodeGenColumn(DataClassBase):
     """代码生成模型列表"""
 
-    __tablename__ = 'gen_column'
+    __tablename__ = 'code_gen_column'
     __table_args__ = (
-        sa.UniqueConstraint('gen_business_id', 'name', name='uk_gen_column_business_id_name'),
+        sa.UniqueConstraint('code_gen_business_id', 'name', name='uk_code_gen_column_business_id_name'),
         {'comment': '代码生成模型列表'},
     )
 
@@ -26,4 +26,4 @@ class GenColumn(DataClassBase):
     is_nullable: Mapped[bool] = mapped_column(default=False, comment='是否可为空')
 
     # 逻辑外键
-    gen_business_id: Mapped[int] = mapped_column(sa.BigInteger, default=0, comment='代码生成业务ID')
+    code_gen_business_id: Mapped[int] = mapped_column(sa.BigInteger, default=0, comment='代码生成业务ID')

backend/plugin/code_generator/schema/business.py

@@ -7,7 +7,7 @@ from backend.common.schema import SchemaBase
 from backend.utils.pattern_validate import is_english_identifier
 
 
-class GenBusinessSchemaBase(SchemaBase):
+class CodeGenBusinessSchemaBase(SchemaBase):
     """代码生成业务基础模型"""
 
     app_name: str = Field(description='应用名称（英文）')
@@ -32,15 +32,15 @@ class GenBusinessSchemaBase(SchemaBase):
         return v
 
 
-class CreateGenBusinessParam(GenBusinessSchemaBase):
+class CreateCodeGenBusinessParam(CodeGenBusinessSchemaBase):
     """创建代码生成业务参数"""
 
 
-class UpdateGenBusinessParam(GenBusinessSchemaBase):
+class UpdateCodeGenBusinessParam(CodeGenBusinessSchemaBase):
     """更新代码生成业务参数"""
 
 
-class GetGenBusinessDetail(GenBusinessSchemaBase):
+class GetCodeGenBusinessDetail(CodeGenBusinessSchemaBase):
     """获取代码生成业务详情"""
 
     model_config = ConfigDict(from_attributes=True)

backend/plugin/code_generator/schema/column.py

@@ -4,7 +4,7 @@ from backend.common.schema import SchemaBase
 from backend.plugin.code_generator.utils.type_conversion import sql_type_to_sqlalchemy
 
 
-class GenColumnSchemaBase(SchemaBase):
+class CodeGenColumnSchemaBase(SchemaBase):
     """代码生成模型基础模型"""
 
     name: str = Field(description='列名称')
@@ -15,7 +15,7 @@ class GenColumnSchemaBase(SchemaBase):
     length: int = Field(description='列长度')
     is_pk: bool = Field(False, description='是否主键')
     is_nullable: bool = Field(False, description='是否可为空')
-    gen_business_id: int = Field(description='代码生成业务ID')
+    code_gen_business_id: int = Field(description='代码生成业务ID')
 
     @field_validator('type')
     @classmethod
@@ -24,21 +24,21 @@ class GenColumnSchemaBase(SchemaBase):
         return sql_type_to_sqlalchemy(v)
 
 
-class CreateGenColumnParam(GenColumnSchemaBase):
+class CreateCodeGenColumnParam(CodeGenColumnSchemaBase):
     """创建代码生成模型列参数"""
 
 
-class CreateGenColumnInternalParam(CreateGenColumnParam):
+class CreateCodeGenColumnInternalParam(CreateCodeGenColumnParam):
     """创建代码生成模型列内部参数"""
 
     pd_type: str | None = Field(None, description='列类型对应的 pydantic 类型')
 
 
-class UpdateGenColumnParam(GenColumnSchemaBase):
+class UpdateCodeGenColumnParam(CodeGenColumnSchemaBase):
     """更新代码生成模型列参数"""
 
 
-class GetGenColumnDetail(GenColumnSchemaBase):
+class GetCodeGenColumnDetail(CodeGenColumnSchemaBase):
     """获取代码生成模型列详情"""
 
     model_config = ConfigDict(from_attributes=True)

backend/plugin/code_generator/service/business_service.py

@@ -5,16 +5,16 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.common.exception import errors
 from backend.common.pagination import paging_data
-from backend.plugin.code_generator.crud.crud_business import gen_business_dao
+from backend.plugin.code_generator.crud.crud_business import code_gen_business_dao
-from backend.plugin.code_generator.model import GenBusiness
+from backend.plugin.code_generator.model import CodeGenBusiness
-from backend.plugin.code_generator.schema.business import CreateGenBusinessParam, UpdateGenBusinessParam
+from backend.plugin.code_generator.schema.business import CreateCodeGenBusinessParam, UpdateCodeGenBusinessParam
 
 
-class GenBusinessService:
+class CodeGenBusinessService:
     """代码生成业务服务类"""
 
     @staticmethod
-    async def get(*, db: AsyncSession, pk: int) -> GenBusiness:
+    async def get(*, db: AsyncSession, pk: int) -> CodeGenBusiness:
         """
         获取指定 ID 的业务
 
@@ -23,13 +23,13 @@ class GenBusinessService:
         :return:
         """
 
-        business = await gen_business_dao.get(db, pk)
+        business = await code_gen_business_dao.get(db, pk)
         if not business:
             raise errors.NotFoundError(msg='代码生成业务不存在')
         return business
 
     @staticmethod
-    async def get_all(*, db: AsyncSession) -> Sequence[GenBusiness]:
+    async def get_all(*, db: AsyncSession) -> Sequence[CodeGenBusiness]:
         """
         获取所有业务
 
@@ -37,7 +37,7 @@ class GenBusinessService:
         :return:
         """
 
-        return await gen_business_dao.get_all(db)
+        return await code_gen_business_dao.get_all(db)
 
     @staticmethod
     async def get_list(*, db: AsyncSession, table_name: str) -> dict[str, Any]:
@@ -48,11 +48,11 @@ class GenBusinessService:
         :param table_name: 业务表名
         :return:
         """
-        business_select = await gen_business_dao.get_select(table_name=table_name)
+        business_select = await code_gen_business_dao.get_select(table_name=table_name)
         return await paging_data(db, business_select)
 
     @staticmethod
-    async def create(*, db: AsyncSession, obj: CreateGenBusinessParam) -> None:
+    async def create(*, db: AsyncSession, obj: CreateCodeGenBusinessParam) -> None:
         """
         创建业务
 
@@ -61,13 +61,13 @@ class GenBusinessService:
         :return:
         """
 
-        business = await gen_business_dao.get_by_name(db, obj.table_name)
+        business = await code_gen_business_dao.get_by_name(db, obj.table_name)
         if business:
             raise errors.ConflictError(msg='代码生成业务已存在')
-        await gen_business_dao.create(db, obj)
+        await code_gen_business_dao.create(db, obj)
 
     @staticmethod
-    async def update(*, db: AsyncSession, pk: int, obj: UpdateGenBusinessParam) -> int:
+    async def update(*, db: AsyncSession, pk: int, obj: UpdateCodeGenBusinessParam) -> int:
         """
         更新业务
 
@@ -77,12 +77,12 @@ class GenBusinessService:
         :return:
         """
 
-        business = await gen_business_dao.get(db, pk)
+        business = await code_gen_business_dao.get(db, pk)
         if not business:
             raise errors.NotFoundError(msg='代码生成业务不存在')
-        if business.table_name != obj.table_name and await gen_business_dao.get_by_name(db, obj.table_name):
+        if business.table_name != obj.table_name and await code_gen_business_dao.get_by_name(db, obj.table_name):
             raise errors.ConflictError(msg='代码生成业务已存在')
-        return await gen_business_dao.update(db, pk, obj)
+        return await code_gen_business_dao.update(db, pk, obj)
 
     @staticmethod
     async def delete(*, db: AsyncSession, pk: int) -> int:
@@ -94,10 +94,10 @@ class GenBusinessService:
         :return:
         """
 
-        business = await gen_business_dao.get(db, pk)
+        business = await code_gen_business_dao.get(db, pk)
         if not business:
             raise errors.NotFoundError(msg='代码生成业务不存在')
-        return await gen_business_dao.delete(db, pk)
+        return await code_gen_business_dao.delete(db, pk)
 
 
-gen_business_service: GenBusinessService = GenBusinessService()
+code_gen_business_service: CodeGenBusinessService = CodeGenBusinessService()

backend/plugin/code_generator/service/code_gen_service.py

@@ -17,21 +17,21 @@ from starlette.concurrency import run_in_threadpool
 from backend.common.exception import errors
 from backend.core.conf import settings
 from backend.core.path_conf import BASE_PATH
-from backend.plugin.code_generator.crud.crud_business import gen_business_dao
+from backend.plugin.code_generator.crud.crud_business import code_gen_business_dao
-from backend.plugin.code_generator.crud.crud_column import gen_column_dao
+from backend.plugin.code_generator.crud.crud_code_gen import code_gen_dao
-from backend.plugin.code_generator.crud.crud_gen import gen_dao
+from backend.plugin.code_generator.crud.crud_column import code_gen_column_dao
-from backend.plugin.code_generator.model import GenBusiness
+from backend.plugin.code_generator.model import CodeGenBusiness
-from backend.plugin.code_generator.schema.business import CreateGenBusinessParam
+from backend.plugin.code_generator.schema.business import CreateCodeGenBusinessParam
-from backend.plugin.code_generator.schema.column import CreateGenColumnInternalParam
+from backend.plugin.code_generator.schema.code_gen import ImportParam
-from backend.plugin.code_generator.schema.gen import ImportParam
+from backend.plugin.code_generator.schema.column import CreateCodeGenColumnInternalParam
-from backend.plugin.code_generator.service.column_service import gen_column_service
+from backend.plugin.code_generator.service.column_service import code_gen_column_service
 from backend.plugin.code_generator.utils.format_code import format_python_code
 from backend.plugin.code_generator.utils.gen_template import gen_template
 from backend.plugin.code_generator.utils.type_conversion import sql_type_to_pydantic
 from backend.utils.locks import acquire_distributed_reload_lock
 
 
-class GenService:
+class CodeGenService:
     """代码生成服务类"""
 
     @staticmethod
@@ -43,7 +43,7 @@ class GenService:
         :param table_schema: 数据库 schema 名称
         :return:
         """
-        return await gen_dao.get_all_tables(db, table_schema)
+        return await code_gen_dao.get_all_tables(db, table_schema)
 
     @staticmethod
     async def import_business_and_model(*, db: AsyncSession, obj: ImportParam) -> None:
@@ -57,11 +57,11 @@ class GenService:
         if settings.ENVIRONMENT != 'dev':
             raise errors.ForbiddenError(msg='禁止在非开发环境下导入代码生成业务')
 
-        table_info = await gen_dao.get_table(db, obj.table_schema, obj.table_name)
+        table_info = await code_gen_dao.get_table(db, obj.table_schema, obj.table_name)
         if not table_info:
             raise errors.NotFoundError(msg='数据库表不存在')
 
-        business_info = await gen_business_dao.get_by_name(db, obj.table_name)
+        business_info = await code_gen_business_dao.get_by_name(db, obj.table_name)
         if business_info:
             raise errors.ConflictError(msg='已存在相同数据库表业务')
 
@@ -71,8 +71,8 @@ class GenService:
             if table_info['table_comment'][-1] == '表'
             else table_info['table_comment'] or table_name.split('_')[-1]
         )
-        new_business = GenBusiness(
+        new_business = CodeGenBusiness(
-            **CreateGenBusinessParam(
+            **CreateCodeGenBusinessParam(
                 app_name=obj.app,
                 table_name=table_name,
                 doc_comment=doc_comment,
@@ -86,13 +86,13 @@ class GenService:
         db.add(new_business)
         await db.flush()
 
-        column_info = await gen_dao.get_all_columns(db, obj.table_schema, table_name)
+        column_info = await code_gen_dao.get_all_columns(db, obj.table_schema, table_name)
-        gen_columns = []
+        code_gen_columns = []
         for column in column_info:
             column_type = column['column_type'].split('(')[0].upper()
             pd_type = sql_type_to_pydantic(column_type)
-            gen_columns.append(
+            code_gen_columns.append(
-                CreateGenColumnInternalParam(
+                CreateCodeGenColumnInternalParam(
                     name=column['column_name'],
                     comment=column['column_comment'],
                     type=column_type,
@@ -102,14 +102,14 @@ class GenService:
                     else 0,
                     is_pk=column['is_pk'],
                     is_nullable=column['is_nullable'],
-                    gen_business_id=new_business.id,
+                    code_gen_business_id=new_business.id,
                     pd_type=pd_type,
                 ),
             )
-        await gen_column_dao.bulk_create(db, gen_columns)
+        await code_gen_column_dao.bulk_create(db, code_gen_columns)
 
     @staticmethod
-    async def _render_tpl_code(*, db: AsyncSession, business: GenBusiness) -> dict[str, str]:
+    async def _render_tpl_code(*, db: AsyncSession, business: CodeGenBusiness) -> dict[str, str]:
         """
         渲染模板代码
 
@@ -117,7 +117,7 @@ class GenService:
         :param business: 业务对象
         :return:
         """
-        gen_models = await gen_column_service.get_columns(db=db, business_id=business.id)
+        gen_models = await code_gen_column_service.get_columns(db=db, business_id=business.id)
         if not gen_models:
             raise errors.NotFoundError(msg='代码生成模型表为空')
 
@@ -176,7 +176,7 @@ class GenService:
         :param pk: 业务 ID
         :return:
         """
-        business = await gen_business_dao.get(db, pk)
+        business = await code_gen_business_dao.get(db, pk)
         if not business:
             raise errors.NotFoundError(msg='业务不存在')
 
@@ -206,7 +206,7 @@ class GenService:
         :param pk: 业务 ID
         :return:
         """
-        business = await gen_business_dao.get(db, pk)
+        business = await code_gen_business_dao.get(db, pk)
         if not business:
             raise errors.NotFoundError(msg='业务不存在')
 
@@ -232,7 +232,7 @@ class GenService:
         if settings.ENVIRONMENT != 'dev':
             raise errors.ForbiddenError(msg='禁止在非开发环境下生成代码')
 
-        business = await gen_business_dao.get(db, pk)
+        business = await code_gen_business_dao.get(db, pk)
         if not business:
             raise errors.NotFoundError(msg='业务不存在')
 
@@ -274,7 +274,7 @@ class GenService:
         :param pk: 业务 ID
         :return:
         """
-        business = await gen_business_dao.get(db, pk)
+        business = await code_gen_business_dao.get(db, pk)
         if not business:
             raise errors.NotFoundError(msg='业务不存在')
 
@@ -297,4 +297,4 @@ class GenService:
         return bio
 
 
-gen_service: GenService = GenService()
+code_gen_service: CodeGenService = CodeGenService()

backend/plugin/code_generator/service/column_service.py

@@ -5,19 +5,19 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from backend.common.enums import DataBaseType
 from backend.common.exception import errors
 from backend.core.conf import settings
-from backend.plugin.code_generator.crud.crud_business import gen_business_dao
+from backend.plugin.code_generator.crud.crud_business import code_gen_business_dao
-from backend.plugin.code_generator.crud.crud_column import gen_column_dao
+from backend.plugin.code_generator.crud.crud_column import code_gen_column_dao
 from backend.plugin.code_generator.enums import GenMySQLColumnType, GenPostgreSQLColumnType
-from backend.plugin.code_generator.model import GenColumn
+from backend.plugin.code_generator.model import CodeGenColumn
-from backend.plugin.code_generator.schema.column import CreateGenColumnParam, UpdateGenColumnParam
+from backend.plugin.code_generator.schema.column import CreateCodeGenColumnParam, UpdateCodeGenColumnParam
 from backend.plugin.code_generator.utils.type_conversion import sql_type_to_pydantic
 
 
-class GenColumnService:
+class CodeGenColumnService:
     """代码生成模型列服务类"""
 
     @staticmethod
-    async def get(*, db: AsyncSession, pk: int) -> GenColumn:
+    async def get(*, db: AsyncSession, pk: int) -> CodeGenColumn:
         """
         获取指定 ID 的模型列
 
@@ -26,10 +26,10 @@ class GenColumnService:
         :return:
         """
 
-        column = await gen_column_dao.get(db, pk)
+        column = await code_gen_column_dao.get(db, pk)
         if not column:
             raise errors.NotFoundError(msg='代码生成模型列不存在')
-        if not await gen_business_dao.get(db, column.gen_business_id):
+        if not await code_gen_business_dao.get(db, column.code_gen_business_id):
             raise errors.NotFoundError(msg='代码生成业务不存在')
         return column
 
@@ -44,7 +44,7 @@ class GenColumnService:
         return types
 
     @staticmethod
-    async def get_columns(*, db: AsyncSession, business_id: int) -> Sequence[GenColumn]:
+    async def get_columns(*, db: AsyncSession, business_id: int) -> Sequence[CodeGenColumn]:
         """
         获取指定业务的所有模型列
 
@@ -53,12 +53,12 @@ class GenColumnService:
         :return:
         """
 
-        if not await gen_business_dao.get(db, business_id):
+        if not await code_gen_business_dao.get(db, business_id):
             raise errors.NotFoundError(msg='代码生成业务不存在')
-        return await gen_column_dao.get_all_by_business(db, business_id)
+        return await code_gen_column_dao.get_all_by_business(db, business_id)
 
     @staticmethod
-    async def create(*, db: AsyncSession, obj: CreateGenColumnParam) -> None:
+    async def create(*, db: AsyncSession, obj: CreateCodeGenColumnParam) -> None:
         """
         创建模型列
 
@@ -67,18 +67,18 @@ class GenColumnService:
         :return:
         """
 
-        if not await gen_business_dao.get(db, obj.gen_business_id):
+        if not await code_gen_business_dao.get(db, obj.code_gen_business_id):
             raise errors.NotFoundError(msg='代码生成业务不存在')
 
-        gen_columns = await gen_column_dao.get_all_by_business(db, obj.gen_business_id)
+        code_gen_columns = await code_gen_column_dao.get_all_by_business(db, obj.code_gen_business_id)
-        if obj.name in [gen_column.name for gen_column in gen_columns]:
+        if obj.name in [code_gen_column.name for code_gen_column in code_gen_columns]:
             raise errors.ForbiddenError(msg='模型列已存在')
 
         pd_type = sql_type_to_pydantic(obj.type)
-        await gen_column_dao.create(db, obj, pd_type=pd_type)
+        await code_gen_column_dao.create(db, obj, pd_type=pd_type)
 
     @staticmethod
-    async def update(*, db: AsyncSession, pk: int, obj: UpdateGenColumnParam) -> int:
+    async def update(*, db: AsyncSession, pk: int, obj: UpdateCodeGenColumnParam) -> int:
         """
         更新模型列
 
@@ -88,20 +88,20 @@ class GenColumnService:
         :return:
         """
 
-        column = await gen_column_dao.get(db, pk)
+        column = await code_gen_column_dao.get(db, pk)
         if not column:
             raise errors.NotFoundError(msg='代码生成模型列不存在')
-        if not await gen_business_dao.get(db, column.gen_business_id):
+        if not await code_gen_business_dao.get(db, column.code_gen_business_id):
             raise errors.NotFoundError(msg='代码生成业务不存在')
-        if not await gen_business_dao.get(db, obj.gen_business_id):
+        if not await code_gen_business_dao.get(db, obj.code_gen_business_id):
             raise errors.NotFoundError(msg='代码生成业务不存在')
         if obj.name != column.name:
-            gen_columns = await gen_column_dao.get_all_by_business(db, obj.gen_business_id)
+            code_gen_columns = await code_gen_column_dao.get_all_by_business(db, obj.code_gen_business_id)
-            if obj.name in [gen_column.name for gen_column in gen_columns]:
+            if obj.name in [code_gen_column.name for code_gen_column in code_gen_columns]:
                 raise errors.ConflictError(msg='模型列名已存在')
 
         pd_type = sql_type_to_pydantic(obj.type)
-        return await gen_column_dao.update(db, pk, obj, pd_type=pd_type)
+        return await code_gen_column_dao.update(db, pk, obj, pd_type=pd_type)
 
     @staticmethod
     async def delete(*, db: AsyncSession, pk: int) -> int:
@@ -113,12 +113,12 @@ class GenColumnService:
         :return:
         """
 
-        column = await gen_column_dao.get(db, pk)
+        column = await code_gen_column_dao.get(db, pk)
         if not column:
             raise errors.NotFoundError(msg='代码生成模型列不存在')
-        if not await gen_business_dao.get(db, column.gen_business_id):
+        if not await code_gen_business_dao.get(db, column.code_gen_business_id):
             raise errors.NotFoundError(msg='代码生成业务不存在')
-        return await gen_column_dao.delete(db, pk)
+        return await code_gen_column_dao.delete(db, pk)
 
 
-gen_column_service: GenColumnService = GenColumnService()
+code_gen_column_service: CodeGenColumnService = CodeGenColumnService()

backend/plugin/code_generator/sql/mysql/destroy.sql

@@ -2,5 +2,5 @@ delete from sys_menu where name in ('AddGenCodeBusiness', 'EditGenCodeBusiness',
 
 delete from sys_menu where name = 'PluginCodeGenerator';
 
-drop table if exists gen_column;
+drop table if exists code_gen_column;
-drop table if exists gen_business;
+drop table if exists code_gen_business;

backend/plugin/code_generator/sql/mysql/destroy_snowflake.sql

@@ -2,5 +2,5 @@ delete from sys_menu where name in ('AddGenCodeBusiness', 'EditGenCodeBusiness',
 
 delete from sys_menu where name = 'PluginCodeGenerator';
 
-drop table if exists gen_column;
+drop table if exists code_gen_column;
-drop table if exists gen_business;
+drop table if exists code_gen_business;

backend/plugin/code_generator/sql/mysql/init.sql

@@ -1,5 +1,5 @@
 insert into sys_menu (title, name, path, sort, icon, type, component, perms, status, display, cache, link, remark, parent_id, created_time, updated_time)
-values ('code_generator.menu', 'PluginCodeGenerator', '/plugins/code-generator', 10, 'tabler:code', 1, '/plugins/code_generator/views/index', null, 1, 1, 1, '', null, null, now(), null);
+values ('code-generator.menu', 'PluginCodeGenerator', '/plugins/code-generator', 10, 'tabler:code', 1, '/plugins/code-generator/views/index', null, 1, 1, 1, '', null, null, now(), null);
 
 set @codegen_menu_id = LAST_INSERT_ID();
 
@@ -14,10 +14,10 @@ values
 ('导入', 'ImportGenCode', null, 0, null, 2, null, 'codegen:table:import', 1, 0, 1, '', null, @codegen_menu_id, now(), null),
 ('写入', 'WriteGenCode', null, 0, null, 2, null, 'codegen:local:write', 1, 0, 1, '', null, @codegen_menu_id, now(), null);
 
-insert into gen_business (id, app_name, table_name, doc_comment, table_comment, class_name, schema_name, filename, datetime_mixin, api_version, gen_path, remark, created_time, updated_time)
+insert into code_gen_business (id, app_name, table_name, doc_comment, table_comment, class_name, schema_name, filename, datetime_mixin, api_version, gen_path, remark, created_time, updated_time)
 values (1, 'test', 'sys_opera_log', '操作日志表', '操作日志表', 'SysOperaLog', 'SysOperaLog', 'sys_opera_log', true, 'v1', null, null, '2025-12-15 15:30:33', null);
 
-insert into gen_column (id, name, comment, type, pd_type, `default`, sort, `length`, is_pk, is_nullable, gen_business_id)
+insert into code_gen_column (id, name, comment, type, pd_type, `default`, sort, `length`, is_pk, is_nullable, code_gen_business_id)
 values
 (1, 'trace_id', '请求跟踪 ID', 'String', 'str', null, 2, 32, false, false, 1),
 (2, 'username', '用户名', 'String', 'str', null, 3, 64, false, true, 1),

backend/plugin/code_generator/sql/mysql/init_snowflake.sql

@@ -1,5 +1,5 @@
 insert into sys_menu (id, title, name, path, sort, icon, type, component, perms, status, display, cache, link, remark, parent_id, created_time, updated_time)
-values (2049629108257816580, 'code_generator.menu', 'PluginCodeGenerator', '/plugins/code-generator', 10, 'tabler:code', 1, '/plugins/code_generator/views/index', null, 1, 1, 1, '', null, null, now(), null);
+values (2049629108257816580, 'code-generator.menu', 'PluginCodeGenerator', '/plugins/code-generator', 10, 'tabler:code', 1, '/plugins/code-generator/views/index', null, 1, 1, 1, '', null, null, now(), null);
 
 insert into sys_menu (id, title, name, path, sort, icon, type, component, perms, status, display, cache, link, remark, parent_id, created_time, updated_time)
 values
@@ -12,10 +12,10 @@ values
 (2049629108257816587, '导入', 'ImportGenCode', null, 0, null, 2, null, 'codegen:table:import', 1, 0, 1, '', null, 2049629108257816580, now(), null),
 (2049629108257816588, '写入', 'WriteGenCode', null, 0, null, 2, null, 'codegen:local:write', 1, 0, 1, '', null, 2049629108257816580, now(), null);
 
-insert into gen_business (id, app_name, table_name, doc_comment, table_comment, class_name, schema_name, filename, datetime_mixin, api_version, gen_path, remark, created_time, updated_time)
+insert into code_gen_business (id, app_name, table_name, doc_comment, table_comment, class_name, schema_name, filename, datetime_mixin, api_version, gen_path, remark, created_time, updated_time)
 values (2112248797819043840, 'test', 'sys_opera_log', '操作日志表', '操作日志表', 'SysOperaLog', 'SysOperaLog', 'sys_opera_log', true, 'v1', null, null, '2025-12-15 15:30:33', null);
 
-insert into gen_column (id, name, comment, type, pd_type, `default`, sort, `length`, is_pk, is_nullable, gen_business_id)
+insert into code_gen_column (id, name, comment, type, pd_type, `default`, sort, `length`, is_pk, is_nullable, code_gen_business_id)
 values
 (2112248797881958400, 'trace_id', '请求跟踪 ID', 'String', 'str', null, 2, 32, false, false, 2112248797819043840),
 (2112248797944872960, 'username', '用户名', 'String', 'str', null, 3, 64, false, true, 2112248797819043840),

backend/plugin/code_generator/sql/postgresql/destroy.sql

@@ -2,7 +2,7 @@ delete from sys_menu where name in ('AddGenCodeBusiness', 'EditGenCodeBusiness',
 
 delete from sys_menu where name = 'PluginCodeGenerator';
 
-drop table if exists gen_column;
+drop table if exists code_gen_column;
-drop table if exists gen_business;
+drop table if exists code_gen_business;
 
 select setval(pg_get_serial_sequence('sys_menu', 'id'), coalesce(max(id), 0) + 1, true) from sys_menu;

backend/plugin/code_generator/sql/postgresql/destroy_snowflake.sql

@@ -2,5 +2,5 @@ delete from sys_menu where name in ('AddGenCodeBusiness', 'EditGenCodeBusiness',
 
 delete from sys_menu where name = 'PluginCodeGenerator';
 
-drop table if exists gen_column;
+drop table if exists code_gen_column;
-drop table if exists gen_business;
+drop table if exists code_gen_business;

backend/plugin/code_generator/sql/postgresql/init.sql

@@ -3,7 +3,7 @@ declare
     codegen_menu_id bigint;
 begin
     insert into sys_menu (title, name, path, sort, icon, type, component, perms, status, display, cache, link, remark, parent_id, created_time, updated_time)
-    values ('code_generator.menu', 'PluginCodeGenerator', '/plugins/code-generator', 10, 'tabler:code', 1, '/plugins/code_generator/views/index', null, 1, 1, 1, '', null, null, now(), null)
+    values ('code-generator.menu', 'PluginCodeGenerator', '/plugins/code-generator', 10, 'tabler:code', 1, '/plugins/code-generator/views/index', null, 1, 1, 1, '', null, null, now(), null)
     returning id into codegen_menu_id;
 
     insert into sys_menu (title, name, path, sort, icon, type, component, perms, status, display, cache, link, remark, parent_id, created_time, updated_time)
@@ -20,10 +20,10 @@ end $$;
 
 select setval(pg_get_serial_sequence('sys_menu', 'id'), coalesce(max(id), 0) + 1, true) from sys_menu;
 
-insert into gen_business (id, app_name, table_name, doc_comment, table_comment, class_name, schema_name, filename, datetime_mixin, api_version, gen_path, remark, created_time, updated_time)
+insert into code_gen_business (id, app_name, table_name, doc_comment, table_comment, class_name, schema_name, filename, datetime_mixin, api_version, gen_path, remark, created_time, updated_time)
 values (1, 'test', 'sys_opera_log', '操作日志表', '操作日志表', 'SysOperaLog', 'SysOperaLog', 'sys_opera_log', true, 'v1', null, null, '2025-12-15 15:30:33', null);
 
-insert into gen_column (id, name, comment, type, pd_type, "default", sort, "length", is_pk, is_nullable, gen_business_id)
+insert into code_gen_column (id, name, comment, type, pd_type, "default", sort, "length", is_pk, is_nullable, code_gen_business_id)
 values
 (1, 'trace_id', '请求跟踪 ID', 'String', 'str', null, 2, 32, false, false, 1),
 (2, 'username', '用户名', 'String', 'str', null, 3, 64, false, true, 1),
@@ -45,5 +45,5 @@ values
 (18, 'cost_time', '请求耗时（ms）', 'String', 'str', null, 19, 0, false, false, 1),
 (19, 'opera_time', '操作时间', 'String', 'str', null, 20, 0, false, false, 1);
 
-select setval(pg_get_serial_sequence('gen_business', 'id'),coalesce(max(id), 0) + 1, true) from gen_business;
+select setval(pg_get_serial_sequence('code_gen_business', 'id'),coalesce(max(id), 0) + 1, true) from code_gen_business;
-select setval(pg_get_serial_sequence('gen_column', 'id'),coalesce(max(id), 0) + 1, true) from gen_column;
+select setval(pg_get_serial_sequence('code_gen_column', 'id'),coalesce(max(id), 0) + 1, true) from code_gen_column;

backend/plugin/code_generator/sql/postgresql/init_snowflake.sql

@@ -1,5 +1,5 @@
 insert into sys_menu (id, title, name, path, sort, icon, type, component, perms, status, display, cache, link, remark, parent_id, created_time, updated_time)
-values (2049629108257816580, 'code_generator.menu', 'PluginCodeGenerator', '/plugins/code-generator', 10, 'tabler:code', 1, '/plugins/code_generator/views/index', null, 1, 1, 1, '', null, null, now(), null);
+values (2049629108257816580, 'code-generator.menu', 'PluginCodeGenerator', '/plugins/code-generator', 10, 'tabler:code', 1, '/plugins/code-generator/views/index', null, 1, 1, 1, '', null, null, now(), null);
 
 insert into sys_menu (id, title, name, path, sort, icon, type, component, perms, status, display, cache, link, remark, parent_id, created_time, updated_time)
 values
@@ -12,10 +12,10 @@ values
 (2049629108257816587, '导入', 'ImportGenCode', null, 0, null, 2, null, 'codegen:table:import', 1, 0, 1, '', null, 2049629108257816580, now(), null),
 (2049629108257816588, '写入', 'WriteGenCode', null, 0, null, 2, null, 'codegen:local:write', 1, 0, 1, '', null, 2049629108257816580, now(), null);
 
-insert into gen_business (id, app_name, table_name, doc_comment, table_comment, class_name, schema_name, filename, datetime_mixin, api_version, gen_path, remark, created_time, updated_time)
+insert into code_gen_business (id, app_name, table_name, doc_comment, table_comment, class_name, schema_name, filename, datetime_mixin, api_version, gen_path, remark, created_time, updated_time)
 values (2112248797819043840, 'test', 'sys_opera_log', '操作日志表', '操作日志表', 'SysOperaLog', 'SysOperaLog', 'sys_opera_log', true, 'v1', null, null, '2025-12-15 15:30:33', null);
 
-insert into gen_column (id, name, comment, type, pd_type, "default", sort, "length", is_pk, is_nullable, gen_business_id)
+insert into code_gen_column (id, name, comment, type, pd_type, "default", sort, "length", is_pk, is_nullable, code_gen_business_id)
 values
 (2112248797881958400, 'trace_id', '请求跟踪 ID', 'String', 'str', null, 2, 32, false, false, 2112248797819043840),
 (2112248797944872960, 'username', '用户名', 'String', 'str', null, 3, 64, false, true, 2112248797819043840),

backend/plugin/code_generator/templates/python/schema.jinja

@@ -49,8 +49,8 @@ class Get{{ schema_name }}Detail({{ schema_name }}SchemaBase):
 
     model_config = ConfigDict(from_attributes=True)
 
-    id: int
+    id: int = Field(description='主键 ID')
     {% if datetime_mixin %}
-    created_time: datetime
+    created_time: datetime = Field(description='创建时间')
-    updated_time: datetime | None = None
+    updated_time: datetime | None = Field(None, description='更新时间')
     {% endif %}

backend/plugin/code_generator/templates/python/service.jinja

@@ -10,6 +10,8 @@ from backend.common.pagination import paging_data
 
 
 class {{ class_name }}Service:
+    """{{ doc_comment }}服务类"""
+
     @staticmethod
     async def get(*, db: AsyncSession, pk: int) -> {{ class_name }}:
         """
@@ -25,7 +27,7 @@ class {{ class_name }}Service:
         return {{ table_name }}
 
     @staticmethod
-    async def get_list(db: AsyncSession) -> dict[str, Any]:
+    async def get_list(*, db: AsyncSession) -> dict[str, Any]:
         """
         获取{{ doc_comment }}列表
 

backend/plugin/code_generator/utils/gen_template.py

@@ -5,7 +5,7 @@ from pydantic.alias_generators import to_pascal
 
 from backend.common.enums import PrimaryKeyType
 from backend.core.conf import settings
-from backend.plugin.code_generator.model import GenBusiness, GenColumn
+from backend.plugin.code_generator.model import CodeGenBusiness, CodeGenColumn
 from backend.plugin.code_generator.path_conf import JINJA2_TEMPLATE_DIR
 from backend.plugin.code_generator.utils.type_conversion import sql_type_to_sqlalchemy_name
 from backend.utils.snowflake import snowflake
@@ -31,17 +31,17 @@ class GenTemplate:
         获取 Jinja2 模板对象
 
         :param jinja_file: Jinja2 模板文件路径
-        :return: Template 对象
+        :return:
         """
         return self.env.get_template(jinja_file)
 
     @staticmethod
-    def get_template_path_mapping(business: GenBusiness) -> dict[str, str]:
+    def get_template_path_mapping(business: CodeGenBusiness) -> dict[str, str]:
         """
         获取模板文件到生成文件的路径映射
 
         :param business: 代码生成业务对象
-        :return: {模板路径: 生成文件路径}
+        :return:
         """
         app_name = business.app_name
         filename = business.filename
@@ -59,12 +59,12 @@ class GenTemplate:
             f'sql/postgresql/init{pk_suffix}.jinja': f'{app_name}/sql/postgresql/init{pk_suffix}.sql',
         }
 
-    def get_init_files(self, business: GenBusiness) -> dict[str, str]:
+    def get_init_files(self, business: CodeGenBusiness) -> dict[str, str]:
         """
         获取需要生成的 __init__.py 文件及其内容
 
         :param business: 业务对象
-        :return: {相对路径: 文件内容}
+        :return:
         """
         app_name = business.app_name
         table_name = business.table_name
@@ -84,7 +84,9 @@ class GenTemplate:
         }
 
     @staticmethod
-    def get_vars(business: GenBusiness, models: Sequence[GenColumn]) -> dict[str, str | Sequence[GenColumn]]:
+    def get_vars(
+        business: CodeGenBusiness, models: Sequence[CodeGenColumn]
+    ) -> dict[str, str | Sequence[CodeGenColumn]]:
         """
         获取模板变量
 

backend/plugin/email/README.md

@@ -31,7 +31,7 @@ EMAIL_PORT = 465
 EMAIL_SSL = true
 ```
 
-在 `backend/core/conf.py` 中添加以下内容：
+当前项目的 `backend/core/conf.py` 已包含以下字段：
 
 ```python
 ##################################################
@@ -49,6 +49,14 @@ EMAIL_CAPTCHA_REDIS_PREFIX: str
 EMAIL_CAPTCHA_EXPIRE_SECONDS: int
 ```
 
+## 配置项说明
+
+- `EMAIL_CAPTCHA_EXPIRE_SECONDS`：控制邮箱验证码有效期
+- `EMAIL_CAPTCHA_REDIS_PREFIX`：控制邮箱验证码 Redis 键前缀
+- `EMAIL_HOST`：控制 SMTP 服务器地址
+- `EMAIL_PORT`：控制 SMTP 端口
+- `EMAIL_SSL`：控制是否启用 SSL
+
 ## 使用方式
 
 1. 安装并启用插件后，配置正确的 SMTP 账号与密码

backend/plugin/installer.py

@@ -183,7 +183,8 @@ async def install_git_frontend_plugin(repo_url: str, frontend_project_root: str)
         raise errors.RequestError(msg='未检测到前端插件目录，请确认路径下存在 apps/web-antdv-next/src/plugins')
 
     repo_name = match.group('repo')
-    plugin_name = repo_name.removesuffix('_ui')
+    # 仓库名允许 _ui 或 -ui 后缀，安装目录不保留该后缀
+    plugin_name = repo_name.removesuffix('_ui') if repo_name.endswith('_ui') else repo_name.removesuffix('-ui')
     if not plugin_name:
         raise errors.RequestError(msg='前端插件仓库名称非法')
 

backend/plugin/oauth2/README.md

@@ -34,7 +34,7 @@ OAUTH2_STATE_EXPIRE_SECONDS = 180
 OAUTH2_STATE_REDIS_PREFIX = 'fba:oauth2:state'
 ```
 
-在 `backend/core/conf.py` 中添加以下内容：
+当前项目的 `backend/core/conf.py` 已包含以下字段：
 
 ```python
 ##################################################
@@ -55,6 +55,15 @@ OAUTH2_FRONTEND_LOGIN_REDIRECT_URI: str
 OAUTH2_FRONTEND_BINDING_REDIRECT_URI: str
 ```
 
+## 配置项说明
+
+- `OAUTH2_FRONTEND_BINDING_REDIRECT_URI`：控制第三方账号绑定完成后的前端回跳地址
+- `OAUTH2_FRONTEND_LOGIN_REDIRECT_URI`：控制第三方登录完成后的前端回跳地址
+- `OAUTH2_GITHUB_REDIRECT_URI`：控制 GitHub OAuth 回调地址
+- `OAUTH2_GOOGLE_REDIRECT_URI`：控制 Google OAuth 回调地址
+- `OAUTH2_STATE_EXPIRE_SECONDS`：控制 OAuth state 有效期
+- `OAUTH2_STATE_REDIS_PREFIX`：控制 OAuth state Redis 键前缀
+
 ## 使用方式
 
 1. 安装并启用插件后，在 GitHub、Google 开放平台分别创建 OAuth 应用

backend/plugin/oauth2/service/oauth2_service.py

@@ -14,7 +14,7 @@ from backend.common.context import ctx
 from backend.common.enums import LoginLogStatusType
 from backend.common.exception import errors
 from backend.common.i18n import t
-from backend.common.security import jwt
+from backend.common.security.token import create_access_token, create_refresh_token
 from backend.core.conf import settings
 from backend.database.redis import redis_client
 from backend.plugin.oauth2.crud.crud_user_social import user_social_dao
@@ -94,7 +94,7 @@ class OAuth2Service:
             await user_social_dao.create(db, new_user_social)
 
         # 创建 token
-        access_token_data = await jwt.create_access_token(
+        access_token_data = await create_access_token(
             sys_user.id,
             multi_login=sys_user.is_multi_login,
             # extra info
@@ -106,7 +106,7 @@ class OAuth2Service:
             browser=ctx.browser,
             device=ctx.device,
         )
-        refresh_token_data = await jwt.create_refresh_token(
+        refresh_token_data = await create_refresh_token(
             access_token_data.session_uuid,
             sys_user.id,
             multi_login=sys_user.is_multi_login,

backend/plugin/oauth2/service/user_social_service.py

@@ -12,6 +12,8 @@ from backend.plugin.oauth2.schema.user_social import CreateUserSocialParam
 
 
 class UserSocialService:
+    """用户社交账号服务类"""
+
     @staticmethod
     async def get_bindings(*, db: AsyncSession, user_id: int) -> list[str]:
         """
@@ -19,7 +21,7 @@ class UserSocialService:
 
         :param db: 数据库会话
         :param user_id: 用户 ID
-        :return: 绑定列表，每个元素包含 sid、source 等信息
+        :return:
         """
         bindings = await user_social_dao.get_by_user_id(db, user_id)
         return [binding.source for binding in bindings]

backend/plugin/status.py

@@ -67,7 +67,10 @@ class PluginStatusChecker:
         :param request: FastAPI 请求对象
         :return:
         """
-        plugin_info = cast('str | None', await redis_client.get(f'{settings.PLUGIN_REDIS_PREFIX}:{self.plugin}'))
+        plugin_info = cast(
+            'str | None',
+            await redis_client.get(f'{settings.PLUGIN_REDIS_PREFIX}:{self.plugin}'),
+        )
         if not plugin_info:
             log.warning('插件 {} 状态未初始化或丢失，尝试自动修复', self.plugin)
             try:

backend/plugin/validator.py

@@ -37,7 +37,7 @@ def _validate_settings(v: dict[str, Any]) -> dict[str, Any]:
 class PluginInfoSchema(BaseModel):
     """插件信息模型"""
 
-    icon: str | None = Field(default=None, description='图标路径或链接地址')
+    icon: str | None = Field(None, description='图标路径或链接地址')
     summary: str = Field(..., min_length=1, max_length=100, description='摘要')
     version: str = Field(..., description='版本号')
     description: str = Field(..., min_length=1, max_length=500, description='描述')

backend/utils/limiter.py

@@ -110,7 +110,8 @@ class RedisBucketFactory(BucketFactory):
         :return:
         """
         bucket_key = self._bucket_key(item.name)
-        now = await _redis_time_ms(redis_client)
+        # wrap_item 已取过 Redis 时间，直接复用，省一次往返
+        now = item.timestamp
 
         async with self.lock:
             state = self.buckets.get(bucket_key)
@@ -119,16 +120,29 @@ class RedisBucketFactory(BucketFactory):
                 self.buckets.move_to_end(bucket_key)
                 return state.bucket
 
-            bucket_result = RedisTimeBucket.init(
+        # 锁外做 Redis IO，避免所有限流请求排队等待 script_load
+        bucket = await _maybe_await(
+            RedisTimeBucket.init(
                 rates=self.rates,
                 redis=redis_client,
                 bucket_key=bucket_key,
             )
-            bucket = await _maybe_await(bucket_result)
+        )
+
+        async with self.lock:
+            # 并发初始化同一 bucket 时以先写入者为准
+            state = self.buckets.get(bucket_key)
+            if state is not None:
+                state.last_seen = now
+                self.buckets.move_to_end(bucket_key)
+                return state.bucket
             self.buckets[bucket_key] = RedisBucketState(bucket=bucket, last_seen=now)
             self.schedule_leak(bucket)
-            await self._evict(now)
+            disposed = self._evict(now)
-            return bucket
+
+        for state in disposed:
+            await self._cleanup(state.bucket, now)
+        return bucket
 
     async def get_bucket(self, name: str) -> RedisBucket:
         """
@@ -139,38 +153,38 @@ class RedisBucketFactory(BucketFactory):
         """
         return await self.get(await self.wrap_item(name))
 
-    async def _evict(self, now: int) -> None:
+    def _evict(self, now: int) -> list[RedisBucketState]:
         """
-        淘汰本地 bucket 缓存
+        淘汰本地 bucket 缓存，只改内存状态，不做 Redis IO
 
         :param now: 当前时间戳，单位毫秒
         :return:
         """
+        expired: list[RedisBucketState] = []
         for bucket_key, state in list(self.buckets.items()):
             if now - state.last_seen <= self.cache_ttl:
                 continue
-            await self._dispose(bucket_key, state, now, cleanup=True)
+            self.buckets.pop(bucket_key, None)
+            self.dispose(state.bucket)
+            expired.append(state)
 
         while len(self.buckets) > self.max_cache_size:
-            bucket_key, state = next(iter(self.buckets.items()))
+            _, state = self.buckets.popitem(last=False)
-            await self._dispose(bucket_key, state, now, cleanup=False)
+            self.dispose(state.bucket)
+        return expired
 
-    async def _dispose(self, bucket_key: str, state: RedisBucketState, now: int, *, cleanup: bool) -> None:
+    @staticmethod
+    async def _cleanup(bucket: RedisBucket, now: int) -> None:
         """
-        移除本地 bucket 并按需清理 Redis 过期数据
+        清理已淘汰 bucket 的 Redis 过期数据
 
-        :param bucket_key: Redis bucket key
+        :param bucket: Redis bucket
-        :param state: Redis bucket 缓存状态
         :param now: 当前时间戳，单位毫秒
-        :param cleanup: 是否执行 Redis 过期数据清理
         :return:
         """
-        self.buckets.pop(bucket_key, None)
+        await _maybe_await(bucket.leak(now))
-        self.dispose(state.bucket)
+        if await _maybe_await(bucket.count()) == 0:
-        if cleanup:
+            await _maybe_await(bucket.flush())
-            await _maybe_await(state.bucket.leak(now))
-            if await _maybe_await(state.bucket.count()) == 0:
-                await _maybe_await(state.bucket.flush())
 
     def _bucket_key(self, name: str) -> str:
         """

backend/utils/locks.py

@@ -4,6 +4,7 @@ from typing import Any
 
 import anyio
 
+from backend.common.exception import errors
 from backend.core.path_conf import RELOAD_LOCK_FILE
 from backend.database.redis import redis_client
 
@@ -16,7 +17,8 @@ async def acquire_distributed_reload_lock() -> AsyncGenerator[None, Any]:
         timeout=300,  # 锁持有超时：5 分钟
         blocking_timeout=60,  # 获取锁等待超时：60 秒
     )
-    await lock.acquire()
+    if not await lock.acquire():
+        raise errors.ServerError(msg='获取热重载锁超时，请稍后重试')
 
     # 文件锁（通知文件监控器跳过重载）
     lock_path = anyio.Path(RELOAD_LOCK_FILE)

backend/utils/snowflake.py

@@ -53,7 +53,7 @@ class SnowflakeNodeManager:
     async def acquire_node_id(self) -> tuple[int, int]:
         """从 Redis 获取可用的 datacenter_id 和 worker_id"""
         occupied_nodes = set()
-        async for key in redis_client.scan_iter(match=f'{self.node_redis_prefix}:*'):
+        async for key in redis_client.scan_iter(match=f'{self.node_redis_prefix}:*', count=1000):
             parts = key.split(':')
             if len(parts) >= 5:
                 try:
@@ -118,6 +118,7 @@ class Snowflake:
         self.last_timestamp: int = -1
 
         self._lock = threading.Lock()
+        self._init_lock = asyncio.Lock()
         self._initialized = False
         self._node_manager: SnowflakeNodeManager | None = None
         self._auto_allocated = False  # 标记是否由 Redis 自动分配 ID
@@ -127,7 +128,11 @@ class Snowflake:
         if self._initialized:
             return
 
-        with self._lock:
+        # 初始化涉及 Redis IO，必须用 asyncio.Lock，threading.Lock 会在 await 期间锁死事件循环
+        async with self._init_lock:
+            if self._initialized:
+                return
+
             # 环境变量固定分配
             if settings.SNOWFLAKE_DATACENTER_ID is not None and settings.SNOWFLAKE_WORKER_ID is not None:
                 self.datacenter_id = settings.SNOWFLAKE_DATACENTER_ID