Optimize schemas with model relationships

Author: wu-clan

backend/app/admin/api/v1/sys/dict_data.py

@@ -4,7 +4,12 @@
 
 from fastapi import APIRouter, Depends, Path, Query
 
-from backend.app.admin.schema.dict_data import CreateDictDataParam, GetDictDataDetail, UpdateDictDataParam
+from backend.app.admin.schema.dict_data import (
+    CreateDictDataParam,
+    GetDictDataDetail,
+    GetDictDataWithRelation,
+    UpdateDictDataParam,
+)
 from backend.app.admin.service.dict_data_service import dict_data_service
 from backend.common.pagination import DependsPagination, PageData, paging_data
 from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
@@ -17,7 +22,7 @@
 
 
 @router.get('/{pk}', summary='获取字典详情', dependencies=[DependsJwtAuth])
-async def get_dict_data(pk: Annotated[int, Path(...)]) -> ResponseSchemaModel[GetDictDataDetail]:
+async def get_dict_data(pk: Annotated[int, Path(...)]) -> ResponseSchemaModel[GetDictDataWithRelation]:
     data = await dict_data_service.get(pk=pk)
     return response_base.success(data=data)
 

backend/app/admin/api/v1/sys/role.py

@@ -7,6 +7,7 @@
 from backend.app.admin.schema.role import (
     CreateRoleParam,
     GetRoleDetail,
+    GetRoleWithRelationDetail,
     UpdateRoleMenuParam,
     UpdateRoleParam,
     UpdateRoleRuleParam,
@@ -49,7 +50,7 @@ async def get_role_all_rules(pk: Annotated[int, Path(...)]) -> ResponseSchemaMod
 
 
 @router.get('/{pk}', summary='获取角色详情', dependencies=[DependsJwtAuth])
-async def get_role(pk: Annotated[int, Path(...)]) -> ResponseSchemaModel[GetRoleDetail]:
+async def get_role(pk: Annotated[int, Path(...)]) -> ResponseSchemaModel[GetRoleWithRelationDetail]:
     data = await role_service.get(pk=pk)
     return response_base.success(data=data)
 

backend/app/admin/api/v1/sys/user.py

@@ -7,8 +7,8 @@
 from backend.app.admin.schema.user import (
     AddUserParam,
     AvatarParam,
-    GetCurrentUserInfoDetail,
-    GetUserInfoDetail,
+    GetCurrentUserInfoWithRelationDetail,
+    GetUserInfoWithRelationDetail,
     RegisterUserParam,
     ResetPasswordParam,
     UpdateUserParam,
@@ -32,7 +32,7 @@ async def register_user(obj: RegisterUserParam) -> ResponseModel:
 
 
 @router.post('/add', summary='添加用户', dependencies=[DependsRBAC])
-async def add_user(request: Request, obj: AddUserParam) -> ResponseSchemaModel[GetUserInfoDetail]:
+async def add_user(request: Request, obj: AddUserParam) -> ResponseSchemaModel[GetUserInfoWithRelationDetail]:
     await user_service.add(request=request, obj=obj)
     data = await user_service.get_userinfo(username=obj.username)
     return response_base.success(data=data)
@@ -47,13 +47,13 @@ async def password_reset(request: Request, obj: ResetPasswordParam) -> ResponseM
 
 
 @router.get('/me', summary='获取当前用户信息', dependencies=[DependsJwtAuth], response_model_exclude={'password'})
-async def get_current_user(request: Request) -> ResponseSchemaModel[GetCurrentUserInfoDetail]:
-    data = GetCurrentUserInfoDetail(**request.user.model_dump())
+async def get_current_user(request: Request) -> ResponseSchemaModel[GetCurrentUserInfoWithRelationDetail]:
+    data = request.user.model_dump()
     return response_base.success(data=data)
 
 
 @router.get('/{username}', summary='查看用户信息', dependencies=[DependsJwtAuth])
-async def get_user(username: Annotated[str, Path(...)]) -> ResponseSchemaModel[GetUserInfoDetail]:
+async def get_user(username: Annotated[str, Path(...)]) -> ResponseSchemaModel[GetUserInfoWithRelationDetail]:
     data = await user_service.get_userinfo(username=username)
     return response_base.success(data=data)
 
@@ -103,7 +103,7 @@ async def get_pagination_users(
     username: Annotated[str | None, Query()] = None,
     phone: Annotated[str | None, Query()] = None,
     status: Annotated[int | None, Query()] = None,
-) -> ResponseSchemaModel[PageData[GetUserInfoDetail]]:
+) -> ResponseSchemaModel[PageData[GetUserInfoWithRelationDetail]]:
     user_select = await user_service.get_select(dept=dept, username=username, phone=phone, status=status)
     page_data = await paging_data(db, user_select)
     return response_base.success(data=page_data)

backend/app/admin/schema/dict_data.py

@@ -30,6 +30,9 @@ class GetDictDataDetail(DictDataSchemaBase):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
-    type: GetDictTypeDetail | None = None
     created_time: datetime
     updated_time: datetime | None = None
+
+
+class GetDictDataWithRelation(DictDataSchemaBase):
+    type: GetDictTypeDetail | None = None

backend/app/admin/schema/role.py

@@ -38,5 +38,8 @@ class GetRoleDetail(RoleSchemaBase):
     id: int
     created_time: datetime
     updated_time: datetime | None = None
+
+
+class GetRoleWithRelationDetail(GetRoleDetail):
     menus: list[GetMenuDetail | None] = []
     rules: list[GetDataRuleDetail | None] = []

backend/app/admin/schema/token.py

@@ -2,15 +2,15 @@
 # -*- coding: utf-8 -*-
 from datetime import datetime
 
-from backend.app.admin.schema.user import GetUserInfoNoRelationDetail
+from backend.app.admin.schema.user import GetUserInfoDetail
 from backend.common.enums import StatusType
 from backend.common.schema import SchemaBase
 
 
 class GetSwaggerToken(SchemaBase):
     access_token: str
     token_type: str = 'Bearer'
-    user: GetUserInfoNoRelationDetail
+    user: GetUserInfoDetail
 
 
 class AccessTokenBase(SchemaBase):
@@ -24,7 +24,7 @@ class GetNewToken(AccessTokenBase):
 
 
 class GetLoginToken(AccessTokenBase):
-    user: GetUserInfoNoRelationDetail
+    user: GetUserInfoDetail
 
 
 class KickOutToken(SchemaBase):

backend/app/admin/schema/user.py

@@ -7,7 +7,7 @@
 from typing_extensions import Self
 
 from backend.app.admin.schema.dept import GetDeptDetail
-from backend.app.admin.schema.role import GetRoleDetail
+from backend.app.admin.schema.role import GetRoleWithRelationDetail
 from backend.common.enums import StatusType
 from backend.common.schema import CustomPhoneNumber, SchemaBase
 
@@ -33,6 +33,12 @@ class AddUserParam(AuthSchemaBase):
     email: EmailStr = Field(examples=['[email protected]'])
 
 
+class ResetPasswordParam(SchemaBase):
+    old_password: str
+    new_password: str
+    confirm_password: str
+
+
 class UserInfoSchemaBase(SchemaBase):
     dept_id: int | None = None
     username: str
@@ -53,7 +59,7 @@ class AvatarParam(SchemaBase):
     url: HttpUrl = Field(description='头像 http 地址')
 
 
-class GetUserInfoNoRelationDetail(UserInfoSchemaBase):
+class GetUserInfoDetail(UserInfoSchemaBase):
     model_config = ConfigDict(from_attributes=True)
 
     dept_id: int | None = None
@@ -68,14 +74,14 @@ class GetUserInfoNoRelationDetail(UserInfoSchemaBase):
     last_login_time: datetime | None = None
 
 
-class GetUserInfoDetail(GetUserInfoNoRelationDetail):
+class GetUserInfoWithRelationDetail(GetUserInfoDetail):
     model_config = ConfigDict(from_attributes=True)
 
     dept: GetDeptDetail | None = None
-    roles: list[GetRoleDetail]
+    roles: list[GetRoleWithRelationDetail]
 
 
-class GetCurrentUserInfoDetail(GetUserInfoDetail):
+class GetCurrentUserInfoWithRelationDetail(GetUserInfoWithRelationDetail):
     model_config = ConfigDict(from_attributes=True)
 
     dept: str | None = None
@@ -92,13 +98,3 @@ def handel(cls, data: Any) -> Self:
         if roles:
             data['roles'] = [role['name'] for role in roles]
         return data
-
-
-class CurrentUserIns(GetUserInfoDetail):
-    model_config = ConfigDict(from_attributes=True)
-
-
-class ResetPasswordParam(SchemaBase):
-    old_password: str
-    new_password: str
-    confirm_password: str

backend/common/security/jwt.py

@@ -15,7 +15,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.app.admin.model import User
-from backend.app.admin.schema.user import CurrentUserIns
+from backend.app.admin.schema.user import GetUserInfoWithRelationDetail
 from backend.common.dataclasses import AccessToken, NewToken, RefreshToken, TokenPayload
 from backend.common.exception.errors import AuthorizationError, TokenError
 from backend.core.conf import settings
@@ -211,7 +211,7 @@ def superuser_verify(request: Request) -> bool:
     return superuser
 
 
-async def jwt_authentication(token: str) -> CurrentUserIns:
+async def jwt_authentication(token: str) -> GetUserInfoWithRelationDetail:
     """
     JWT authentication
 
@@ -227,7 +227,7 @@ async def jwt_authentication(token: str) -> CurrentUserIns:
     if not cache_user:
         async with async_db_session() as db:
             current_user = await get_current_user(db, user_id)
-            user = CurrentUserIns(**select_as_dict(current_user))
+            user = GetUserInfoWithRelationDetail(**select_as_dict(current_user))
             await redis_client.setex(
                 f'{settings.JWT_USER_REDIS_PREFIX}:{user_id}',
                 settings.JWT_USER_REDIS_EXPIRE_SECONDS,
@@ -236,5 +236,5 @@ async def jwt_authentication(token: str) -> CurrentUserIns:
     else:
         # TODO: 在恰当的时机，应替换为使用 model_validate_json
         # https://docs.pydantic.dev/latest/concepts/json/#partial-json-parsing
-        user = CurrentUserIns.model_validate(from_json(cache_user, allow_partial=True))
+        user = GetUserInfoWithRelationDetail.model_validate(from_json(cache_user, allow_partial=True))
     return user

backend/middleware/jwt_auth_middleware.py

@@ -7,7 +7,7 @@
 from starlette.authentication import AuthCredentials, AuthenticationBackend, AuthenticationError
 from starlette.requests import HTTPConnection
 
-from backend.app.admin.schema.user import CurrentUserIns
+from backend.app.admin.schema.user import GetUserInfoWithRelationDetail
 from backend.common.exception.errors import TokenError
 from backend.common.log import log
 from backend.common.security.jwt import jwt_authentication
@@ -32,7 +32,7 @@ def auth_exception_handler(conn: HTTPConnection, exc: _AuthenticationError) -> R
         """覆盖内部认证错误处理"""
         return MsgSpecJSONResponse(content={'code': exc.code, 'msg': exc.msg, 'data': None}, status_code=exc.code)
 
-    async def authenticate(self, request: Request) -> tuple[AuthCredentials, CurrentUserIns] | None:
+    async def authenticate(self, request: Request) -> tuple[AuthCredentials, GetUserInfoWithRelationDetail] | None:
         token = request.headers.get('Authorization')
         if not token:
             return