22# -*- coding: utf-8 -*-
33from __future__ import annotations
44
5- import math
6-
5+ from math import ceil
76from typing import TYPE_CHECKING , Generic , Sequence , TypeVar
87
98from fastapi import Depends , Query
109from fastapi_pagination import pagination_ctx
1110from fastapi_pagination .bases import AbstractPage , AbstractParams , RawParams
1211from fastapi_pagination .ext .sqlalchemy import paginate
1312from fastapi_pagination .links .bases import create_links
14- from pydantic import BaseModel
13+ from pydantic import BaseModel , Field
1514
1615if TYPE_CHECKING :
1716 from sqlalchemy import Select
1817 from sqlalchemy .ext .asyncio import AsyncSession
1918
2019T = TypeVar ('T' )
21- DataT = TypeVar ('DataT' )
2220SchemaT = TypeVar ('SchemaT' )
2321
2422
25- class _Params (BaseModel , AbstractParams ):
23+ class _CustomPageParams (BaseModel , AbstractParams ):
2624 page : int = Query (1 , ge = 1 , description = 'Page number' )
2725 size : int = Query (20 , gt = 0 , le = 100 , description = 'Page size' ) # 默认 20 条记录
2826
@@ -33,53 +31,90 @@ def to_raw_params(self) -> RawParams:
3331 )
3432
3533
36- class _Page (AbstractPage [T ], Generic [T ]):
37- items : Sequence [T ] # 数据
38- total : int # 总数据数
39- page : int # 第n页
40- size : int # 每页数量
41- total_pages : int # 总页数
42- links : dict [str , str | None ] # 跳转链接
34+ class _Links (BaseModel ):
35+ first : str = Field (..., description = '首页链接' )
36+ last : str = Field (..., description = '尾页链接' )
37+ self : str = Field (..., description = '当前页链接' )
38+ next : str | None = Field (None , description = '下一页链接' )
39+ prev : str | None = Field (None , description = '上一页链接' )
40+
41+
42+ class _PageDetails (BaseModel ):
43+ items : list = Field ([], description = '当前页数据' )
44+ total : int = Field (..., description = '总条数' )
45+ page : int = Field (..., description = '当前页' )
46+ size : int = Field (..., description = '每页数量' )
47+ total_pages : int = Field (..., description = '总页数' )
48+ links : _Links
49+
4350
44- __params_type__ = _Params # 使用自定义的Params
51+ class _CustomPage (_PageDetails , AbstractPage [T ], Generic [T ]):
52+ __params_type__ = _CustomPageParams
4553
4654 @classmethod
4755 def create (
4856 cls ,
49- items : Sequence [ T ] ,
57+ items : list ,
5058 total : int ,
51- params : _Params ,
52- ) -> _Page [T ]:
59+ params : _CustomPageParams ,
60+ ) -> _CustomPage [T ]:
5361 page = params .page
5462 size = params .size
55- total_pages = math .ceil (total / params .size )
56- links = create_links (** {
57- 'first' : {'page' : 1 , 'size' : f'{ size } },
58- 'last' : {'page' : f'{ math .ceil (total / params .size )} , 'size' : f'{ size } } if total > 0 else None ,
59- 'next' : {'page' : f'{ page + 1 } , 'size' : f'{ size } } if (page + 1 ) <= total_pages else None ,
60- 'prev' : {'page' : f'{ page - 1 } , 'size' : f'{ size } } if (page - 1 ) >= 1 else None ,
61- }).model_dump ()
63+ total_pages = ceil (total / params .size )
64+ links = create_links (
65+ first = {'page' : 1 , 'size' : size },
66+ last = {'page' : f'{ ceil (total / params .size )} , 'size' : size } if total > 0 else {'page' : 1 , 'size' : size },
67+ next = {'page' : f'{ page + 1 } , 'size' : size } if (page + 1 ) <= total_pages else None ,
68+ prev = {'page' : f'{ page - 1 } , 'size' : size } if (page - 1 ) >= 1 else None ,
69+ ).model_dump ()
70+
71+ return cls (
72+ items = items ,
73+ total = total ,
74+ page = params .page ,
75+ size = params .size ,
76+ total_pages = total_pages ,
77+ links = links , # type: ignore
78+ )
79+
6280
63- return cls (items = items , total = total , page = params .page , size = params .size , total_pages = total_pages , links = links )
81+ class PageData (_PageDetails , Generic [SchemaT ]):
82+ """
83+ 包含 data schema 的统一返回模型，适用于分页接口
6484
85+ E.g. ::
86+
87+ @router.get('/test', response_model=ResponseSchemaModel[PageData[GetApiDetail]])
88+ def test():
89+ return ResponseSchemaModel[PageData[GetApiDetail]](data=GetApiDetail(...))
90+
91+
92+ @router.get('/test')
93+ def test() -> ResponseSchemaModel[PageData[GetApiDetail]]:
94+ return ResponseSchemaModel[PageData[GetApiDetail]](data=GetApiDetail(...))
95+
96+
97+ @router.get('/test')
98+ def test() -> ResponseSchemaModel[PageData[GetApiDetail]]:
99+ res = CustomResponseCode.HTTP_200
100+ return ResponseSchemaModel[PageData[GetApiDetail]](code=res.code, msg=res.msg, data=GetApiDetail(...))
101+ """
65102
66- class _PageData (BaseModel , Generic [DataT ]):
67- page_data : DataT | None = None
103+ items : Sequence [SchemaT ]
68104
69105
70- async def paging_data (db : AsyncSession , select : Select , page_data_schema : SchemaT ) -> dict :
106+ async def paging_data (db : AsyncSession , select : Select ) -> dict :
71107 """
72108 基于 SQLAlchemy 创建分页数据
73109
74110 :param db:
75111 :param select:
76- :param page_data_schema:
77112 :return:
78113 """
79- _paginate = await paginate (db , select )
80- page_data = _PageData [ _Page [ page_data_schema ]]( page_data = _paginate ) .model_dump ()[ 'page_data' ]
114+ paginated_data : _CustomPage = await paginate (db , select )
115+ page_data = paginated_data .model_dump ()
81116 return page_data
82117
83118
84119# 分页依赖注入
85- DependsPagination = Depends (pagination_ctx (_Page ))
120+ DependsPagination = Depends (pagination_ctx (_CustomPage ))
0 commit comments