75 lines
2.0 KiB
Python
75 lines
2.0 KiB
Python
|
from typing import Generic, Optional, Sequence, TypeVar
|
|||
|
|
|
|||
|
|
import math
|
|||
|
|
from fastapi import Query
|
|||
|
|
from pydantic import BaseModel
|
|||
|
|
from tortoise.queryset import QuerySet
|
|||
|
|
|
|||
|
|
T = TypeVar("T")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class PaginationPydantic(BaseModel, Generic[T]):
|
|||
|
|
"""分页模型"""
|
|||
|
|
status_code: int = 200
|
|||
|
|
message: str = "Success"
|
|||
|
|
total: int
|
|||
|
|
page: int
|
|||
|
|
size: int
|
|||
|
|
total_pages: int
|
|||
|
|
data: Sequence[T]
|
|||
|
|
|
|||
|
|
|
|||
|
|
class Params(BaseModel):
|
|||
|
|
"""传参"""
|
|||
|
|
# 设置默认值为1，不能够小于1
|
|||
|
|
page: int = Query(1, ge=1, description="Page number")
|
|||
|
|
|
|||
|
|
# 设置默认值为10，最大为100
|
|||
|
|
size: int = Query(10, gt=0, le=200, description="Page size")
|
|||
|
|
|
|||
|
|
# 默认值None表示选传
|
|||
|
|
order_by: Optional[str] = Query(None, max_length=32, description="Sort key")
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def pagination(pydantic_model, query_set: QuerySet, params: Params, callback=None):
|
|||
|
|
"""分页响应"""
|
|||
|
|
"""
|
|||
|
|
pydantic_model: Pydantic model
|
|||
|
|
query_set: QuerySet
|
|||
|
|
params: Params
|
|||
|
|
callback: if you want to do something for query_set,it will be useful.
|
|||
|
|
"""
|
|||
|
|
page: int = params.page
|
|||
|
|
size: int = params.size
|
|||
|
|
order_by: str = params.order_by
|
|||
|
|
total = await query_set.count()
|
|||
|
|
|
|||
|
|
# 通过总数和每页数量计算出总页数
|
|||
|
|
total_pages = math.ceil(total / size)
|
|||
|
|
|
|||
|
|
if page > total_pages and total: # 排除查询集为空时报错，即total=0时
|
|||
|
|
raise ValueError("页数输入有误")
|
|||
|
|
|
|||
|
|
# 排序后分页
|
|||
|
|
if order_by:
|
|||
|
|
order_by = order_by.split(',')
|
|||
|
|
query_set = query_set.order_by(*order_by)
|
|||
|
|
|
|||
|
|
# 分页
|
|||
|
|
query_set = query_set.offset((page - 1) * size) # 页数 * 页面大小=偏移量
|
|||
|
|
query_set = query_set.limit(size)
|
|||
|
|
|
|||
|
|
if callback:
|
|||
|
|
"""对查询集操作"""
|
|||
|
|
query_set = await callback(query_set)
|
|||
|
|
|
|||
|
|
# query_set = await query_set
|
|||
|
|
data = await pydantic_model.from_queryset(query_set)
|
|||
|
|
return PaginationPydantic(**{
|
|||
|
|
"total": total,
|
|||
|
|
"page": page,
|
|||
|
|
"size": size,
|
|||
|
|
"total_pages": total_pages,
|
|||
|
|
"data": data,
|
|||
|
|
})
|