分页

约 604 字大约 2 分钟

fba 基于 fastapi-pagination 封装了统一分页能力，核心实现位于 backend/common/pagination.py

普通分页

普通分页适用于后台管理中常见的列表页。接口层添加 DependsPagination，返回类型使用 ResponseSchemaModel[PageData[XxxDetail]]

筛选参数模型

筛选参数统一定义在 Schema 层，并使用 Query 模型绑定

from pydantic import Field

from backend.common.schema import SchemaBase


class UserQueryParam(SchemaBase):
    """用户查询参数"""

    dept: int | None = Field(None, description='部门 ID')
    username: str | None = Field(None, description='用户名')
    phone: str | None = Field(None, description='手机号')
    status: int | None = Field(None, description='状态')

from typing import Annotated

from fastapi import Query

from backend.app.admin.schema.user import UserQueryParam
from backend.common.pagination import DependsPagination, PageData
from backend.common.response.response_schema import ResponseSchemaModel, response_base
from backend.database.db import CurrentSession


@router.get(
    '',
    summary='分页获取用户',
    dependencies=[
        DependsPagination,
    ],
)
async def get_users_paginated(
    db: CurrentSession,
    query: Annotated[UserQueryParam, Query()],
) -> ResponseSchemaModel[PageData[GetUserInfoWithRelationDetail]]:
    page_data = await user_service.get_list(
        db=db,
        dept=query.dept,
        username=query.username,
        phone=query.phone,
        status=query.status,
    )
    return response_base.success(data=page_data)

筛选条件较多时，可以按职责拆分为多个 Query 模型并在路由中并列声明，FastAPI 会分别应用每个模型的默认值、字段校验和模型校验。参数仍保持在请求顶层，不会嵌套为 query 或 filters 对象；Header 和 Cookie 模型也遵循相同规则

page、size 由 DependsPagination 提供，不需要重复定义在业务 Query 模型中

DependsPagination 默认提供以下查询参数：

参数	默认值	说明
page	1	页码，最小为 1
size	20	每页数量，最大 200

返回数据包含：

字段	说明
items	当前页数据列表
total	数据总条数
page	当前页码
size	每页数量
total_pages	总页数
links	分页链接

Service / CRUD

Service 层通常调用 CRUD 层获取 SQLAlchemy Select 对象，再通过 paging_data() 生成分页数据

from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from backend.app.admin.crud.crud_user import user_dao
from backend.common.pagination import paging_data


class UserService:
    """用户服务类"""

    @staticmethod
    async def get_list(
        *,
        db: AsyncSession,
        dept: int | None,
        username: str | None,
        phone: str | None,
        status: int | None,
    ) -> dict[str, Any]:
        user_select = await user_dao.get_select(
            dept=dept,
            username=username,
            phone=phone,
            status=status,
        )
        return await paging_data(db, user_select)

游标分页

游标分页适用于无限滚动、会话列表等需要基于游标继续加载的场景。接口层使用 DependsCursorPagination，返回类型使用 ResponseSchemaModel[CursorPageData[XxxDetail]]

from backend.common.pagination import CursorPageData, DependsCursorPagination
from backend.common.response.response_schema import ResponseSchemaModel, response_base
from backend.database.db import CurrentSession


@router.get(
    '',
    summary='游标分页获取会话',
    dependencies=[
        DependsCursorPagination,
    ],
)
async def get_conversations(
    db: CurrentSession,
) -> ResponseSchemaModel[CursorPageData[GetAIConversationListDetail]]:
    page_data = await conversation_service.get_list(db=db)
    return response_base.success(data=page_data)

游标分页返回数据包含：

字段	说明
items	当前页数据列表
next_cursor	下一页游标
has_more	是否还有更多数据