FastAPI分页功能实现

分页功能实现

一、站在巨人的肩膀上（fastapi-pagination）

GitHub：https://github.com/uriyyo/fastapi-pagination

文档：https://uriyyo-fastapi-pagination.netlify.app/

二、安装fastapi-pagination

pip install fastapi-pagination

三、直接使用

from fastapi_pagination import Page, paginate, Params,add_pagination

• Page：用于在路径中(response_model)声明返回模型
• Params：用于提供分页参数
• paginate：用于将数据进行分页

1. 使用依赖项

from fastapi import FastAPI, Depends
from fastapi_pagination import Page, paginate, add_pagination, Params
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    name: str
    surname: str

users = [
    User(name='name', surname='name'),
    User(name='name2', surname='name'),
    User(name='name3', surname='name'),
    User(name='name4', surname='name'),
    User(name='name5', surname='name'),
    User(name='name6', surname='name'),
    User(name='name7', surname='name'),
    User(name='name8', surname='name'),
    User(name='name9', surname='name'),
    # ...
]

# 【response_model=Page[User]】在路径参数中声明响应数据类型
# 【params: Params = Depends()】把Params当做依赖项进行引入
@app.get('/users', response_model=Page[User])
async def get_users(params: Params = Depends()):
    # 将需要分页的数据进行分页，把params参数传入
    return paginate(users, params)

2. 省略依赖项

from fastapi import FastAPI, Depends
from fastapi_pagination import Page, paginate, add_pagination, Params
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    name: str
    surname: str

users = [
    User(name='name', surname='name'),
    User(name='name2', surname='name'),
    User(name='name3', surname='name'),
    User(name='name4', surname='name'),
    User(name='name5', surname='name'),
    User(name='name6', surname='name'),
    User(name='name7', surname='name'),
    User(name='name8', surname='name'),
    User(name='name9', surname='name'),
    # ...
]

# 【response_model=Page[User]】在路径参数中声明响应数据类型
@app.get('/users', response_model=Page[User])
async def get_users():
    # 将需要分页的数据进行分页
    return paginate(users)
# 在接口函数之后，使用add_pagination进行默认添加分页依赖项
add_pagination(app)

四、Limit-Offset

from fastapi import FastAPI
from fastapi_pagination import LimitOffsetPage, add_pagination, paginate
from pydantic import BaseModel

app = FastAPI()

class User(BaseModel):
    name: str
    surname: str

users = [
    User(name='name', surname='name'),
    User(name='name2', surname='name'),
    User(name='name3', surname='name'),
    User(name='name4', surname='name'),
    User(name='name5', surname='name'),
    User(name='name6', surname='name'),
    User(name='name7', surname='name'),
    User(name='name8', surname='name'),
    User(name='name9', surname='name'),
    # ...
]

@app.get('/users', response_model=LimitOffsetPage[User])
async def get_users():
    return paginate(users)

add_pagination(app)

五、自定义参数信息

1.自定义Params(把page参数起始值改为1)

from fastapi import Query
from fastapi_pagination.bases import AbstractParams, RawParams
from pydantic import BaseModel

class Params(BaseModel, AbstractParams):
    # 设置默认值为1，不能够小于1
    page: int = Query(1, ge=1, description="Page number")
    # 设置默认值为20，最大为100
    size: int = Query(20, gt=0, le=100, description="Page size")

    def to_raw_params(self) -> RawParams:
        return RawParams(
            limit=self.size,
            # 更爱page参数起始值从1开始
            offset=self.size * (self.page - 1),
        )

2.自定义Page模型

from __future__ import annotations

import math
from typing import TypeVar, Generic, Sequence
from fastapi_pagination.bases import AbstractPage

T = TypeVar("T")

class Page(AbstractPage[T], Generic[T]):
    # 修改Page模型
    data: Sequence[T] # 数据
    total: int # 总数据数
    page: int # 第n页
    size: int # 每页数量
    next: str # 下一页参数
    previous: str # 上一页参数
    total_pages: int # 总页数

    # 使用自定义的Params
    __params_type__ = Params  # Set params related to Page

    @classmethod
    def create(
            cls,
            items: data,
            total: int,
            params: Params,
    ) -> Page[T]:
        # 从params获取page和size
        page = params.page
        size = params.size
        # 通过总数和每页数量计算出总页数
        total_pages = math.ceil(total / params.size)
        # 生成下一页参数（如果没有下一页则为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"

        # 根据定义的模型参数进行返回
        return cls(data=items, total=total, page=params.page,
                   size=params.size,
                   next=next,
                   previous=previous,
                   total_pages=total_pages)

3.重写整合（pagination.py）

from __future__ import annotations
import math
from typing import TypeVar, Generic, Sequence

from fastapi import Query
from fastapi_pagination.bases import AbstractPage, AbstractParams, RawParams
from pydantic import BaseModel

T = TypeVar("T")

class Params(BaseModel, AbstractParams):
    page: int = Query(1, ge=1, description="Page number")
    size: int = Query(20, gt=0, le=100, description="Page size")

    def to_raw_params(self) -> RawParams:
        return RawParams(
            limit=self.size,
            offset=self.size * (self.page - 1),
        )

class Page(AbstractPage[T], Generic[T]):
    results: Sequence[T]
    total: int
    page: int
    size: int
    next: str
    previous: str
    total_pages: int

    __params_type__ = Params  # Set params related to Page

    @classmethod
    def create(
            cls,
            results: results,
            total: int,
            params: Params,
    ) -> Page[T]:
        page = params.page
        size = params.size
        total_pages = math.ceil(total / params.size)
        next = f"?page={page + 1}&size={size}" if (page + 1) <= total_pages else "null"
        previous = f"?page={page - 1}&size={size}" if (page - 1) >= 1 else "null"

        return cls(results=results, total=total, page=params.page,
                   size=params.size,
                   next=next,
                   previous=previous,
                   total_pages=total_pages)

4.使用自定义后的分页功能

# 把Page从自定义的包中导入即可
from pagination import Page

@app.get('/users', response_model=Page[User])
async def get_users():
    return paginate(users)

六、集成SQLAlchemy使用

1、简单使用

1） 更改【paginate】包的路径为【fastapi_pagination.ext.sqlalchemy】

from fastapi_pagination.ext.sqlalchemy import paginate

2） 创建数据返回模型

class User(BaseModel):
    email: str
    password: str
    is_active: bool

    # 这个配置必须得要
    class Config:
        orm_mode = True

3） 使用【sqlalchemy.paginate】

# 【db: Session = Depends(get_db)】：数据库依赖项
@app.get('/users', response_model=Page[User])
async def get_users(db: Session = Depends(get_db)):
    # 此处的参数为【db.query(models.User)】
    # 【sqlalchemy.paginate】的参数为【sqlalchemy.orm.query.Query】类型
    return paginate(db.query(models.User))

2、使用异步

1）安装支持异步的aiomysql

pip install aiomysql

2）sqlalchemy引擎需要异步创建

from sqlalchemy.ext.asyncio import create_async_engine

# 这里要使用支持异步的aiomysql
SQLALCHEMY_DATABASE_URL = "mysql+aiomysql://root:123456@127.0.0.1:3306/users?charset=utf8"
# 注意sqlite不支持异步
engine = create_async_engine(SQLALCHEMY_DATABASE_URL)

3）创建异步的数据库会话

from sqlalchemy.ext.asyncio import AsyncSession

SessionLocal = sessionmaker(bind=engine, class_=AsyncSession)

4）创建异步的数据库依赖项

# 依赖项要定义为异步函数
async def get_db():
    async with SessionLocal() as session:
        yield session

5）paginate使用async_sqlalchemy模块下的

from fastapi_pagination.ext.async_sqlalchemy import paginate
from sqlalchemy.future import select

@app.get('/users', response_model=Page[User])
async def get_users(db: Session = Depends(get_db)):
    # 第一个参数为数据库连接
    # 第二个参数为Select类型的参数
    # 要添加await关键字
    return await paginate(db, select(models.User))

七、演示