添加schema响应模式 (#48)

Author: wu-clan

.pre-commit-config.yaml

@@ -8,7 +8,7 @@ repos:
       - id: check-toml
 
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: v0.8.2
+    rev: v0.9.5
     hooks:
       - id: ruff
         args:
@@ -19,7 +19,7 @@ repos:
       - id: ruff-format
 
   - repo: https://github.com/pdm-project/pdm
-    rev: 2.21.0
+    rev: 2.22.3
     hooks:
       - id: pdm-export
         args:

.ruff.toml

@@ -15,7 +15,7 @@ select = [
     "RUF100",
     "I002",
     "F404",
-    "TCH",
+    "TC",
     "UP007"
 ]
 preview = true
@@ -25,8 +25,8 @@ lines-between-types = 1
 order-by-type = true
 
 [lint.per-file-ignores]
-"**/api/v1/*.py" = ["TCH"]
-"**/model/*.py" = ["TCH003"]
+"**/api/v1/*.py" = ["TC"]
+"**/model/*.py" = ["TC003"]
 "**/model/__init__.py" = ["F401"]
 
 [format]

backend/app/admin/api/v1/auth/auth.py

@@ -5,8 +5,8 @@
 
 from backend.app.admin.service.auth_service import auth_service
 from backend.common.security.jwt import DependsJwtAuth
-from backend.common.response.response_schema import response_base, ResponseModel
-from backend.app.admin.schema.token import GetSwaggerToken
+from backend.common.response.response_schema import response_base, ResponseModel, ResponseSchemaModel
+from backend.app.admin.schema.token import GetSwaggerToken, GetLoginToken
 from backend.app.admin.schema.user import Auth2
 
 router = APIRouter()
@@ -19,7 +19,7 @@ async def swagger_login(form_data: OAuth2PasswordRequestForm = Depends()) -> Get
 
 
 @router.post('/login', summary='验证码登录')
-async def user_login(request: Request, obj: Auth2) -> ResponseModel:
+async def user_login(request: Request, obj: Auth2) -> ResponseSchemaModel[GetLoginToken]:
     data = await auth_service.login(request=request, obj=obj)
     return response_base.success(data=data)
 

backend/app/admin/api/v1/auth/captcha.py

@@ -5,7 +5,8 @@
 from fastapi_limiter.depends import RateLimiter
 from starlette.concurrency import run_in_threadpool
 
-from backend.common.response.response_schema import ResponseModel, response_base
+from backend.app.admin.schema.captcha import GetCaptchaDetail
+from backend.common.response.response_schema import ResponseSchemaModel, response_base
 from backend.core.conf import settings
 from backend.database.db import uuid4_str
 from backend.database.redis import redis_client
@@ -18,18 +19,18 @@
     summary='获取登录验证码',
     dependencies=[Depends(RateLimiter(times=5, seconds=10))],
 )
-async def get_captcha(request: Request) -> ResponseModel:
+async def get_captcha(request: Request) -> ResponseSchemaModel[GetCaptchaDetail]:
     """
     此接口可能存在性能损耗，尽管是异步接口，但是验证码生成是IO密集型任务，使用线程池尽量减少性能损耗
     """
     img_type: str = 'base64'
     img, code = await run_in_threadpool(img_captcha, img_byte=img_type)
     uuid = uuid4_str()
     request.app.state.captcha_uuid = uuid
-    await redis_client.set(f'{settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{uuid}', code, ex=settings.CAPTCHA_EXPIRATION_TIME)
-    return response_base.success(
-        data={
-            'image_type': img_type,
-            'image': img,
-        }
+    await redis_client.set(
+        f'{settings.CAPTCHA_LOGIN_REDIS_PREFIX}:{uuid}',
+        code,
+        ex=settings.CAPTCHA_LOGIN_EXPIRE_SECONDS,
     )
+    data = GetCaptchaDetail(image_type=img_type, image=img)
+    return response_base.success(data=data)

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

@@ -5,8 +5,8 @@
 from fastapi import APIRouter, Query
 
 from backend.common.security.jwt import CurrentUser, DependsJwtAuth
-from backend.common.pagination import paging_data, DependsPagination
-from backend.common.response.response_schema import response_base, ResponseModel
+from backend.common.pagination import paging_data, DependsPagination, PageData
+from backend.common.response.response_schema import response_base, ResponseModel, ResponseSchemaModel
 from backend.database.db import CurrentSession
 from backend.app.admin.schema.user import CreateUser, GetUserInfo, ResetPassword, UpdateUser, Avatar
 from backend.app.admin.service.user_service import UserService
@@ -30,7 +30,7 @@ async def password_reset(obj: ResetPassword) -> ResponseModel:
 
 
 @router.get('/{username}', summary='查看用户信息', dependencies=[DependsJwtAuth])
-async def get_user(username: str) -> ResponseModel:
+async def get_user(username: str) -> ResponseSchemaModel[GetUserInfo]:
     current_user = await UserService.get_userinfo(username=username)
     data = GetUserInfo(**select_as_dict(current_user))
     return response_base.success(data=data)
@@ -52,15 +52,22 @@ async def update_avatar(username: str, avatar: Avatar) -> ResponseModel:
     return response_base.fail()
 
 
-@router.get('', summary='（模糊条件）分页获取所有用户', dependencies=[DependsJwtAuth, DependsPagination])
+@router.get(
+    '',
+    summary='（模糊条件）分页获取所有用户',
+    dependencies=[
+        DependsJwtAuth,
+        DependsPagination,
+    ],
+)
 async def get_all_users(
     db: CurrentSession,
     username: Annotated[str | None, Query()] = None,
     phone: Annotated[str | None, Query()] = None,
     status: Annotated[int | None, Query()] = None,
-) -> ResponseModel:
+) -> ResponseSchemaModel[PageData[GetUserInfo]]:
     user_select = await UserService.get_select(username=username, phone=phone, status=status)
-    page_data = await paging_data(db, user_select, GetUserInfo)
+    page_data = await paging_data(db, user_select)
     return response_base.success(data=page_data)
 
 

backend/app/admin/schema/captcha.py

@@ -0,0 +1,10 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from pydantic import Field
+
+from backend.common.schema import SchemaBase
+
+
+class GetCaptchaDetail(SchemaBase):
+    image_type: str = Field(description='图片类型')
+    image: str = Field(description='图片内容')

backend/common/pagination.py

@@ -2,27 +2,25 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
-import math
-
+from math import ceil
 from typing import TYPE_CHECKING, Generic, Sequence, TypeVar
 
 from fastapi import Depends, Query
 from fastapi_pagination import pagination_ctx
 from fastapi_pagination.bases import AbstractPage, AbstractParams, RawParams
 from fastapi_pagination.ext.sqlalchemy import paginate
 from fastapi_pagination.links.bases import create_links
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 if TYPE_CHECKING:
     from sqlalchemy import Select
     from sqlalchemy.ext.asyncio import AsyncSession
 
 T = TypeVar('T')
-DataT = TypeVar('DataT')
 SchemaT = TypeVar('SchemaT')
 
 
-class _Params(BaseModel, AbstractParams):
+class _CustomPageParams(BaseModel, AbstractParams):
     page: int = Query(1, ge=1, description='Page number')
     size: int = Query(20, gt=0, le=100, description='Page size')  # 默认 20 条记录
 
@@ -33,53 +31,90 @@ def to_raw_params(self) -> RawParams:
         )
 
 
-class _Page(AbstractPage[T], Generic[T]):
-    items: Sequence[T]  # 数据
-    total: int  # 总数据数
-    page: int  # 第n页
-    size: int  # 每页数量
-    total_pages: int  # 总页数
-    links: dict[str, str | None]  # 跳转链接
+class _Links(BaseModel):
+    first: str = Field(..., description='首页链接')
+    last: str = Field(..., description='尾页链接')
+    self: str = Field(..., description='当前页链接')
+    next: str | None = Field(None, description='下一页链接')
+    prev: str | None = Field(None, description='上一页链接')
+
+
+class _PageDetails(BaseModel):
+    items: list = Field([], description='当前页数据')
+    total: int = Field(..., description='总条数')
+    page: int = Field(..., description='当前页')
+    size: int = Field(..., description='每页数量')
+    total_pages: int = Field(..., description='总页数')
+    links: _Links
+
 
-    __params_type__ = _Params  # 使用自定义的Params
+class _CustomPage(_PageDetails, AbstractPage[T], Generic[T]):
+    __params_type__ = _CustomPageParams
 
     @classmethod
     def create(
         cls,
-        items: Sequence[T],
+        items: list,
         total: int,
-        params: _Params,
-    ) -> _Page[T]:
+        params: _CustomPageParams,
+    ) -> _CustomPage[T]:
         page = params.page
         size = params.size
-        total_pages = math.ceil(total / params.size)
-        links = create_links(**{
-            'first': {'page': 1, 'size': f'{size}'},
-            'last': {'page': f'{math.ceil(total / params.size)}', 'size': f'{size}'} if total > 0 else None,
-            'next': {'page': f'{page + 1}', 'size': f'{size}'} if (page + 1) <= total_pages else None,
-            'prev': {'page': f'{page - 1}', 'size': f'{size}'} if (page - 1) >= 1 else None,
-        }).model_dump()
+        total_pages = ceil(total / params.size)
+        links = create_links(
+            first={'page': 1, 'size': size},
+            last={'page': f'{ceil(total / params.size)}', 'size': size} if total > 0 else {'page': 1, 'size': size},
+            next={'page': f'{page + 1}', 'size': size} if (page + 1) <= total_pages else None,
+            prev={'page': f'{page - 1}', 'size': size} if (page - 1) >= 1 else None,
+        ).model_dump()
+
+        return cls(
+            items=items,
+            total=total,
+            page=params.page,
+            size=params.size,
+            total_pages=total_pages,
+            links=links,  # type: ignore
+        )
+
 
-        return cls(items=items, total=total, page=params.page, size=params.size, total_pages=total_pages, links=links)
+class PageData(_PageDetails, Generic[SchemaT]):
+    """
+    包含 data schema 的统一返回模型，适用于分页接口
 
+    E.g. ::
+
+        @router.get('/test', response_model=ResponseSchemaModel[PageData[GetApiDetail]])
+        def test():
+            return ResponseSchemaModel[PageData[GetApiDetail]](data=GetApiDetail(...))
+
+
+        @router.get('/test')
+        def test() -> ResponseSchemaModel[PageData[GetApiDetail]]:
+            return ResponseSchemaModel[PageData[GetApiDetail]](data=GetApiDetail(...))
+
+
+        @router.get('/test')
+        def test() -> ResponseSchemaModel[PageData[GetApiDetail]]:
+            res = CustomResponseCode.HTTP_200
+            return ResponseSchemaModel[PageData[GetApiDetail]](code=res.code, msg=res.msg, data=GetApiDetail(...))
+    """
 
-class _PageData(BaseModel, Generic[DataT]):
-    page_data: DataT | None = None
+    items: Sequence[SchemaT]
 
 
-async def paging_data(db: AsyncSession, select: Select, page_data_schema: SchemaT) -> dict:
+async def paging_data(db: AsyncSession, select: Select) -> dict:
     """
     基于 SQLAlchemy 创建分页数据
 
     :param db:
     :param select:
-    :param page_data_schema:
     :return:
     """
-    _paginate = await paginate(db, select)
-    page_data = _PageData[_Page[page_data_schema]](page_data=_paginate).model_dump()['page_data']
+    paginated_data: _CustomPage = await paginate(db, select)
+    page_data = paginated_data.model_dump()
     return page_data
 
 
 # 分页依赖注入
-DependsPagination = Depends(pagination_ctx(_Page))
+DependsPagination = Depends(pagination_ctx(_CustomPage))

backend/common/response/response_schema.py

@@ -1,23 +1,19 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
-from datetime import datetime
-from typing import Any
+from typing import Any, Generic, TypeVar
 
 from fastapi import Response
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel
 
 from backend.common.response.response_code import CustomResponse, CustomResponseCode
-from backend.core.conf import settings
 from backend.utils.serializers import MsgSpecJSONResponse
 
-_ExcludeData = set[int | str] | dict[int | str, Any]
-
-__all__ = ['ResponseModel', 'response_base']
+SchemaT = TypeVar('SchemaT')
 
 
 class ResponseModel(BaseModel):
     """
-    统一返回模型
+    通用型统一返回模型，不包含 data schema
 
     E.g. ::
 
@@ -37,33 +33,26 @@ def test() -> ResponseModel:
             return ResponseModel(code=res.code, msg=res.msg, data={'test': 'test'})
     """
 
-    # TODO: json_encoders 配置失效: https://github.com/tiangolo/fastapi/discussions/10252
-    model_config = ConfigDict(json_encoders={datetime: lambda x: x.strftime(settings.DATETIME_FORMAT)})
-
     code: int = CustomResponseCode.HTTP_200.code
     msg: str = CustomResponseCode.HTTP_200.msg
     data: Any | None = None
 
 
-class ResponseBase:
-    """
-    统一返回方法
-
-    .. tip::
+class ResponseSchemaModel(ResponseModel, Generic[SchemaT]):
+    """包含 data schema 的统一返回模型，适用于非分页接口"""
 
-        此类中的方法将返回 ResponseModel 模型，作为一种编码风格而存在；
+    data: SchemaT
 
-    E.g. ::
 
-        @router.get('/test')
-        def test() -> ResponseModel:
-            return response_base.success(data={'test': 'test'})
-    """
+class ResponseBase:
+    """统一返回方法"""
 
     @staticmethod
-    def __response(*, res: CustomResponseCode | CustomResponse = None, data: Any | None = None) -> ResponseModel:
+    def __response(
+        *, res: CustomResponseCode | CustomResponse = None, data: Any | None = None
+    ) -> ResponseModel | ResponseSchemaModel:
         """
-        请求成功返回通用方法
+        请求返回通用方法
 
         :param res: 返回信息
         :param data: 返回数据
@@ -76,15 +65,15 @@ def success(
         *,
         res: CustomResponseCode | CustomResponse = CustomResponseCode.HTTP_200,
         data: Any | None = None,
-    ) -> ResponseModel:
+    ) -> ResponseModel | ResponseSchemaModel:
         return self.__response(res=res, data=data)
 
     def fail(
         self,
         *,
         res: CustomResponseCode | CustomResponse = CustomResponseCode.HTTP_400,
         data: Any = None,
-    ) -> ResponseModel:
+    ) -> ResponseModel | ResponseSchemaModel:
         return self.__response(res=res, data=data)
 
     @staticmethod
@@ -94,11 +83,11 @@ def fast_success(
         data: Any | None = None,
     ) -> Response:
         """
-        此方法是为了提高接口响应速度而创建的，如果返回数据无需进行 pydantic 解析和验证，则推荐使用，相反，请不要使用！
+        此方法是为了提高接口响应速度而创建的，在解析较大 json 时有显著性能提升，但将丢失 pydantic 解析和验证
 
         .. warning::
 
-            使用此返回方法时，不要指定接口参数 response_model，也不要在接口函数后添加箭头返回类型
+            使用此返回方法时，不能指定接口参数 response_model 和箭头返回类型
 
         :param res:
         :param data:

backend/common/schema.py

@@ -1,8 +1,12 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+from datetime import datetime
+
 from pydantic import BaseModel, ConfigDict, EmailStr, validate_email
 from pydantic_extra_types.phone_numbers import PhoneNumber
 
+from backend.core.conf import settings
+
 # 自定义验证错误信息不包含验证预期内容（也就是输入内容），受支持的预期内容字段参考以下链接
 # https://github.com/pydantic/pydantic-core/blob/a5cb7382643415b716b1a7a5392914e50f726528/tests/test_errors.py#L266
 # 替换预期内容字段方式，参考以下链接
@@ -149,4 +153,7 @@ def _validate(cls, __input_value: str) -> str:
 
 
 class SchemaBase(BaseModel):
-    model_config = ConfigDict(use_enum_values=True)
+    model_config = ConfigDict(
+        use_enum_values=True,
+        json_encoders={datetime: lambda x: x.strftime(settings.DATETIME_FORMAT)},
+    )

backend/core/conf.py

@@ -58,7 +58,7 @@ def validator_api_url(cls, values):
 
     # Captcha
     CAPTCHA_LOGIN_REDIS_PREFIX: str = 'fba:login:captcha'
-    CAPTCHA_EXPIRATION_TIME: int = 60 * 5  # 过期时间，单位：秒
+    CAPTCHA_LOGIN_EXPIRE_SECONDS: int = 60 * 5  # 过期时间，单位：秒
 
     # Token
     TOKEN_ALGORITHM: str = 'HS256'  # 算法