FastaApi Tortoise 分页器

FastaApi目前提供了分页器模块：fastapi_pagination
并且提供了好几种分页形式，网上都能查到，这里不多说。fastapi_pagination缺点在于先查询完所以数据并且序列化完后再分页，分页原理主要是通过查询集切片分页。
下面是

T = TypeVar("T")


def paginate(
    sequence: Sequence[T],
    params: Optional[AbstractParams] = None,
    length_function: Callable[[Sequence[T]], int] = len,
) -> AbstractPage[T]:
    params = resolve_params(params)
    raw_params = params.to_raw_params()

    return create_page(
        items=sequence[raw_params.offset : raw_params.offset + raw_params.limit],
        total=length_function(sequence),
        params=params,
    )


__all__ = ["paginate"]

使用举例：查询文章列表

from fastapi import APIRouter, Depends
from fastapi_pagination import Page, paginate, Params, add_pagination
from tortoise import timezone
from tortoise.expressions import Q
from tortoise.query_utils import Prefetch

from src.faster.database.models import ArticleComment, ForumArticle
from src.faster.routers.cp_test import app_name
from src.faster.routers.forums.pydantics import ForumArticleSchema
from src.my_tools.fastapi_tools import BaseViewSet, Action

cp_test_routers = APIRouter(prefix=f"/{app_name}", tags=["测试"])

class TestViewSet(BaseViewSet):
    model = ForumArticle
    schema = ForumArticleSchema
    pk_type = str
    views = {
    }
    @Action.get("/list", summary="获取或者模糊查询帖子列表", response_model=Page[ForumArticleSchema],
                description="可以根据con_type筛选公告：2，或帖子：1")  #
    async def list(self, keyword: str = None, tag_name: str = None, con_type: int = 1,
                   params: Params = Depends()):
        print(timezone.now())
        query_set = ForumArticle.filter(is_delete=0, con_type=con_type).select_related('user', "anonymous_user") \
            .prefetch_related(Prefetch("article_comments", queryset=ArticleComment.filter(is_delete=False))) \
            .prefetch_related("article_upvotes", "posters", "tags", "at_users")

        if keyword:
            query_set = query_set.filter(Q(content__icontains=keyword) |
                                         (Q(user__name__icontains=keyword) & Q(is_anonymous=False)) |
                                         (Q(anonymous_user__name__icontains=keyword) & Q(
                                             is_anonymous=True)))
        if tag_name:
            query_set = query_set.filter(tag__name__icontains=tag_name)

        print(timezone.now())
        data = await ForumArticleSchema.from_queryset(query_set)

        print(timezone.now())
        return paginate(data)

TestViewSet.register(cp_test_routers)
add_pagination(cp_test_routers)

#print：
#2022-08-10 08:31:02.341401+00:00
#2022-08-10 08:31:02.341401+00:00
#2022-08-10 08:31:07.361233+00:00

经过测试，查询比较复杂数据结构的查询集，900多条数据，序列化时间需要5s左右，明显是没法使用的。数据结构比较简单的使用没问题。所以博主自定义分页器，先查询分页，然后再序列化后返回响应数据。

from __future__ import annotations

import math

from fastapi import Query
from pydantic import BaseModel
from starlette import status
from tortoise.queryset import QuerySet

from src.faster.my_response import MyErrorResponse

from typing import Generic, TypeVar, Sequence

from pydantic.generics import GenericModel



T = TypeVar("T")


class PagePydantic(GenericModel, Generic[T]):
    """分页模型"""
    data: Sequence[T]
    total: int
    page: int
    size: int
    total_pages: int
    next: str
    previous: str


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")
    order_by: str = Query(None, max_length=32, description="Sort key")  # 默认值None表示选传


async def pagination(pydantic_model, query_set: QuerySet, params: Params, callback=None):
    """分页响应"""
    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时
        return MyErrorResponse(msg="页数输入有误", status_code=status.HTTP_400_BAD_REQUEST)

    # 排序后分页
    if order_by:
        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)

    # 生成下一页参数（如果没有下一页则为null）
    next = f"?page={page + 1}&size={size}" if (page + 1) <= total_pages else "null"
    # 生成上一页参数（如果没有上一页则为null）
    previous = f"?page={page - 1}&size={size}" if (page - 1) >= 1 else "null"
    # query_set = await query_set
    data = await pydantic_model.from_queryset(query_set)
    return PagePydantic(**{
        "data": data,  # todo 此处排序之后序列化比较耗时，如何优化
        "total": total,
        "page": page,
        "size": size,
        "total_pages": total_pages,
        "next": next,
        "previous": previous,
    })

主要通过对QuerySet进行分页排序等进行操作，到最后在序列化时再执行对应sql命令，数据量就会小很多。
接口使用事例：

from src.faster.routers.pagination import Params, pagination, PagePydantic

class ForumArticlesViewSet(BaseViewSet):
    model = ForumArticle
    schema = ForumArticleSchema
    pk_type = int
    views = {
        "create": ForumArticleCreateSchema,
        "get": ForumArticleSchema
    }

    Action.get("/list", summary="获取或者模糊查询帖子列表", response_model=PagePydantic[ForumArticleSchema],
                description="可以根据con_type筛选公告：2，或帖子：1")  #
    async def list(self, is_mine: bool = None, keyword: str = None, tag_name: str = None, con_type: int = 1,
                   params: Params = Depends(), user: User = Depends(get_login_user)):
        query_set = ForumArticle.filter(is_delete=0, con_type=con_type).order_by("-is_topping", "-create_date")\
            .select_related('user', "anonymous_user") \
            .prefetch_related(Prefetch("article_comments", queryset=ArticleComment.filter(is_delete=False)))\
            .prefetch_related("tags")

        if is_mine:
            query_set = query_set.filter(user=user)
        if keyword:
            query_set = query_set.filter(Q(content__icontains=keyword) |
                                         (Q(user__name__icontains=keyword) & Q(is_anonymous=False)) |
                                         (Q(anonymous_user__name__icontains=keyword) & Q(
                                             is_anonymous=True)))

        # query_set = await query_set
        if tag_name:
            query_set = query_set.filter(tags__name__icontains=tag_name)
        serializer_data = await pagination(pydantic_model=ForumArticleSchema, query_set=query_set, params=params)

        return serializer_data

项目源码：https://github.com/NotBeBarnon/fastapi-tortoise-pagination，支持pip安装直接使用