Add response model include data schema

backend/app/admin/api/v1/log/login_log.py

@@ -4,7 +4,6 @@
 
 from fastapi import APIRouter, Depends, Query
 
-from backend.app.admin.schema.login_log import GetLoginLogListDetails
 from backend.app.admin.service.login_log_service import login_log_service
 from backend.common.pagination import DependsPagination, paging_data
 from backend.common.response.response_schema import ResponseModel, response_base
@@ -31,7 +30,7 @@ async def get_pagination_login_logs(
     ip: Annotated[str | None, Query()] = None,
 ) -> ResponseModel:
     log_select = await login_log_service.get_select(username=username, status=status, ip=ip)
-    page_data = await paging_data(db, log_select, GetLoginLogListDetails)
+    page_data = await paging_data(db, log_select)
     return response_base.success(data=page_data)
 
 

backend/app/admin/api/v1/log/opera_log.py

@@ -4,7 +4,6 @@
 
 from fastapi import APIRouter, Depends, Query
 
-from backend.app.admin.schema.opera_log import GetOperaLogListDetails
 from backend.app.admin.service.opera_log_service import opera_log_service
 from backend.common.pagination import DependsPagination, paging_data
 from backend.common.response.response_schema import ResponseModel, response_base
@@ -31,7 +30,7 @@ async def get_pagination_opera_logs(
     ip: Annotated[str | None, Query()] = None,
 ) -> ResponseModel:
     log_select = await opera_log_service.get_select(username=username, status=status, ip=ip)
-    page_data = await paging_data(db, log_select, GetOperaLogListDetails)
+    page_data = await paging_data(db, log_select)
     return response_base.success(data=page_data)
 
 

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

@@ -4,7 +4,7 @@
 
 from fastapi import APIRouter, Depends, Path, Query, Request
 
-from backend.app.admin.schema.api import CreateApiParam, GetApiListDetails, UpdateApiParam
+from backend.app.admin.schema.api import CreateApiParam, UpdateApiParam
 from backend.app.admin.service.api_service import api_service
 from backend.common.pagination import DependsPagination, paging_data
 from backend.common.response.response_schema import ResponseModel, response_base
@@ -44,7 +44,7 @@ async def get_pagination_apis(
     path: Annotated[str | None, Query()] = None,
 ) -> ResponseModel:
     api_select = await api_service.get_select(request=request, name=name, method=method, path=path)
-    page_data = await paging_data(db, api_select, GetApiListDetails)
+    page_data = await paging_data(db, api_select)
     return response_base.success(data=page_data)
 
 

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

@@ -11,7 +11,6 @@
     DeleteAllPoliciesParam,
     DeletePolicyParam,
     DeleteUserRoleParam,
-    GetPolicyListDetails,
     UpdatePoliciesParam,
     UpdatePolicyParam,
 )
@@ -40,7 +39,7 @@ async def get_pagination_casbin(
     sub: Annotated[str | None, Query(description='用户 uuid / 角色')] = None,
 ) -> ResponseModel:
     casbin_select = await casbin_service.get_casbin_list(ptype=ptype, sub=sub)
-    page_data = await paging_data(db, casbin_select, GetPolicyListDetails)
+    page_data = await paging_data(db, casbin_select)
     return response_base.success(data=page_data)
 
 

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

@@ -6,7 +6,6 @@
 
 from backend.app.admin.schema.config import (
     CreateAnyConfigParam,
-    GetAnyConfigListDetails,
     SaveConfigParam,
     UpdateAnyConfigParam,
 )
@@ -98,7 +97,7 @@ async def get_pagination_config(
     type: Annotated[str | None, Query()] = None,
 ) -> ResponseModel:
     config_select = await config_service.get_select(name=name, type=type)
-    page_data = await paging_data(db, config_select, GetAnyConfigListDetails)
+    page_data = await paging_data(db, config_select)
     return response_base.success(data=page_data)
 
 

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

@@ -53,7 +53,7 @@ async def get_data_rule(pk: Annotated[int, Path(...)]) -> ResponseModel:
 )
 async def get_pagination_data_rule(db: CurrentSession, name: Annotated[str | None, Query()] = None) -> ResponseModel:
     data_rule_select = await data_rule_service.get_select(name=name)
-    page_data = await paging_data(db, data_rule_select, GetDataRuleListDetails)
+    page_data = await paging_data(db, data_rule_select)
     return response_base.success(data=page_data)
 
 

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

@@ -39,7 +39,7 @@ async def get_pagination_dict_datas(
     status: Annotated[int | None, Query()] = None,
 ) -> ResponseModel:
     dict_data_select = await dict_data_service.get_select(label=label, value=value, status=status)
-    page_data = await paging_data(db, dict_data_select, GetDictDataListDetails)
+    page_data = await paging_data(db, dict_data_select)
     return response_base.success(data=page_data)
 
 

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

@@ -4,7 +4,7 @@
 
 from fastapi import APIRouter, Depends, Path, Query
 
-from backend.app.admin.schema.dict_type import CreateDictTypeParam, GetDictTypeListDetails, UpdateDictTypeParam
+from backend.app.admin.schema.dict_type import CreateDictTypeParam, UpdateDictTypeParam
 from backend.app.admin.service.dict_type_service import dict_type_service
 from backend.common.pagination import DependsPagination, paging_data
 from backend.common.response.response_schema import ResponseModel, response_base
@@ -31,7 +31,7 @@ async def get_pagination_dict_types(
     status: Annotated[int | None, Query()] = None,
 ) -> ResponseModel:
     dict_type_select = await dict_type_service.get_select(name=name, code=code, status=status)
-    page_data = await paging_data(db, dict_type_select, GetDictTypeListDetails)
+    page_data = await paging_data(db, dict_type_select)
     return response_base.success(data=page_data)
 
 

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

@@ -4,7 +4,7 @@
 
 from fastapi import APIRouter, Depends, Path, Query
 
-from backend.app.admin.schema.notice import CreateNoticeParam, GetNoticeListDetails, UpdateNoticeParam
+from backend.app.admin.schema.notice import CreateNoticeParam, UpdateNoticeParam
 from backend.app.admin.service.notice_service import notice_service
 from backend.common.pagination import DependsPagination, paging_data
 from backend.common.response.response_schema import ResponseModel, response_base
@@ -32,7 +32,7 @@ async def get_notice(pk: Annotated[int, Path(...)]) -> ResponseModel:
 )
 async def get_pagination_notice(db: CurrentSession) -> ResponseModel:
     notice_select = await notice_service.get_select()
-    page_data = await paging_data(db, notice_select, GetNoticeListDetails)
+    page_data = await paging_data(db, notice_select)
     return response_base.success(data=page_data)
 
 

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

@@ -72,7 +72,7 @@ async def get_pagination_roles(
     status: Annotated[int | None, Query()] = None,
 ) -> ResponseModel:
     role_select = await role_service.get_select(name=name, status=status)
-    page_data = await paging_data(db, role_select, GetRoleListDetails)
+    page_data = await paging_data(db, role_select)
     return response_base.success(data=page_data)
 
 

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

@@ -108,7 +108,7 @@ async def get_pagination_users(
     status: Annotated[int | None, Query()] = None,
 ):
     user_select = await user_service.get_select(dept=dept, username=username, phone=phone, status=status)
-    page_data = await paging_data(db, user_select, GetUserInfoListDetails)
+    page_data = await paging_data(db, user_select)
     return response_base.success(data=page_data)
 
 

backend/common/pagination.py

@@ -2,27 +2,25 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
-import math
-
-from typing import TYPE_CHECKING, Dict, Generic, Sequence, TypeVar
+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,92 @@ 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: Sequence[SchemaT],
         total: int,
-        params: _Params,
-    ) -> _Page[T]:
+        params: _CustomPageParams,
+    ) -> _CustomPage[T]:
         page = params.page
         size = params.size
-        total_pages = math.ceil(total / params.size)
+        total_pages = 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,
+            'last': {'page': f'{ceil(total / params.size)}', 'size': f'{size}'} if total > 0 else 1,
             '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()
 
-        return cls(items=items, total=total, page=params.page, size=params.size, total_pages=total_pages, links=links)
+        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[GetApiListDetails]])
+        def test():
+            return ResponseSchemaModel[PageData[GetApiListDetails]](data=GetApiListDetails(...))
+
 
+        @router.get('/test')
+        def test() -> ResponseSchemaModel[PageData[GetApiListDetails]]:
+            return ResponseSchemaModel[PageData[GetApiListDetails]](data=GetApiListDetails(...))
+
+
+        @router.get('/test')
+        def test() -> ResponseSchemaModel[PageData[GetApiListDetails]]:
+            res = CustomResponseCode.HTTP_200
+            return ResponseSchemaModel[PageData[GetApiListDetails]](
+                code=res.code, msg=res.msg, data=GetApiListDetails(...)
+            )
+    """
 
-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))