如何在FastAPI中实现类DRF分页？SqlAlchemy语句执行优化

核心思路 #

和DRF的分页逻辑对齐：先构建未执行的查询对象/语句，再将其传入分页函数统一执行计数和分页查询，把查询构建和执行逻辑解耦，实现灵活复用。

下面分SQLAlchemy两个大版本给出实现代码：

SQLAlchemy v2.x 版本实现 #

import sqlalchemy as sa
from fastapi import FastAPI, Depends
from fastapi_filter import FilterDepends
from fastapi_filter.contrib.sqlalchemy import Filter
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import declarative_base, sessionmaker

metadata = MetaData()
Base = declarative_base(metadata=metadata)

# 初始化异步引擎和会话
engine = create_async_engine("sqlite+aiosqlite:///./test.db", echo=True)
SessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)


class User(Base):
    __tablename__ = "user"

    id = sa.Column(sa.Integer, autoincrement=True, primary_key=True, index=True)
    first_name = sa.Column(sa.String)
    last_name = sa.Column(sa.String)
    username = sa.Column(sa.String, unique=True)
    email = sa.Column(sa.String, unique=True)


class UserFilter(Filter):
    username: str | None = None
    email: str | None = None

    class Constants(Filter.Constants):
        model = User
        ordering_field_name = "ordering"


async def get_session() -> AsyncSession:
    session = SessionLocal()
    try:
        yield session
    finally:
        await session.close()


# 只构建查询语句，不执行
def fetch_users_stmt(filtering=None, limit=None, offset=None):
    stmt = sa.select(User)

    if filtering:
        stmt = filtering.filter(stmt)
        stmt = filtering.sort(stmt)

    if limit:
        stmt = stmt.limit(limit)
    if offset:
        stmt = stmt.offset(offset)
    
    return stmt


# 分页函数负责执行查询
async def paginate(session: AsyncSession, stmt: sa.Select):
    # 构建计数查询（移除limit/offset，只统计总数）
    count_stmt = sa.select(sa.func.count()).select_from(stmt.subquery())
    count_result = await session.execute(count_stmt)
    total_count = count_result.scalar_one()

    # 执行分页查询
    result = await session.execute(stmt)
    items = result.scalars().all()

    return {
        "count": total_count,
        "next": None,  # 可根据需求实现下一页URL生成
        "previous": None,  # 可根据需求实现上一页URL生成
        "results": items
    }


app = FastAPI()


@app.get('/users')
async def get_users(
    filtering: UserFilter = FilterDepends(UserFilter),
    limit: int = 10,
    offset: int = 0,
    session: AsyncSession = Depends(get_session)
):
    # 获取构建好的查询语句
    users_stmt = fetch_users_stmt(filtering, limit, offset)
    # 传入分页函数执行并返回结果
    return await paginate(session, users_stmt)

SQLAlchemy v1.x 版本实现 #

v1.x中异步会话的查询语法和v2差异较大，主要用session.query()构建查询对象：

import sqlalchemy as sa
from fastapi import FastAPI, Depends
from fastapi_filter import FilterDepends
from fastapi_filter.contrib.sqlalchemy import Filter
from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import declarative_base, sessionmaker

metadata = MetaData()
Base = declarative_base(metadata=metadata)

engine = create_async_engine("sqlite+aiosqlite:///./test.db", echo=True)
SessionLocal = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)


class User(Base):
    __tablename__ = "user"

    id = sa.Column(sa.Integer, autoincrement=True, primary_key=True, index=True)
    first_name = sa.Column(sa.String)
    last_name = sa.Column(sa.String)
    username = sa.Column(sa.String, unique=True)
    email = sa.Column(sa.String, unique=True)


class UserFilter(Filter):
    username: str | None = None
    email: str | None = None

    class Constants(Filter.Constants):
        model = User
        ordering_field_name = "ordering"


async def get_session() -> AsyncSession:
    session = SessionLocal()
    try:
        yield session
    finally:
        await session.close()


# 构建查询对象，不执行
def fetch_users_query(filtering=None, limit=None, offset=None):
    query = sa.orm.session.Session.query(User)  # 注意v1的写法

    if filtering:
        query = filtering.filter(query)
        query = filtering.sort(query)

    if limit:
        query = query.limit(limit)
    if offset:
        query = query.offset(offset)
    
    return query


# 分页函数执行查询
async def paginate(session: AsyncSession, query):
    # 复制查询对象，移除limit/offset后计数
    count_query = query.with_entities(sa.func.count(User.id)).order_by(None)
    count_result = await session.execute(count_query)
    total_count = count_result.scalar_one()

    # 执行分页查询
    result = await session.execute(query)
    items = result.scalars().all()

    return {
        "count": total_count,
        "next": None,
        "previous": None,
        "results": items
    }


app = FastAPI()


@app.get('/users')
async def get_users(
    filtering: UserFilter = FilterDepends(UserFilter),
    limit: int = 10,
    offset: int = 0,
    session: AsyncSession = Depends(get_session)
):
    users_query = fetch_users_query(filtering, limit, offset)
    return await paginate(session, users_query)

关键修改说明 #

1. 查询构建与执行解耦：fetch_users_stmt/fetch_users_query只负责根据过滤、分页参数构建查询语句/对象，不执行数据库操作，完全对应DRF中get_queryset()的逻辑。
2. 分页函数统一执行：paginate函数接收会话和查询对象，分别执行计数查询和分页查询，返回DRF风格的分页响应结构。
3. SQLAlchemy版本兼容：
   • v2使用sa.select()构建语句，通过session.execute()执行
   • v1使用session.query()构建查询对象，执行方式也略有不同
4. 计数优化：计数时移除原查询的limit和offset，避免统计错误，同时使用subquery或with_entities保证计数效率。

内容的提问来源于stack exchange，提问作者Альберт Александров