webapp/app/utils/pagination.py

from typing import Any, Dict
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql import Select


async def paginate(
    db: AsyncSession,
    stmt: Select,
    page: int = 1,
    size: int = 20,
) -> Dict[str, Any]:
    if page < 1:
        page = 1
    if size < 1:
        size = 20

    # ---------- 1️⃣ 查询总数 ----------
    count_stmt = select(func.count()).select_from(
        stmt.subquery()
    )
    total = await db.scalar(count_stmt) or 0

    # ---------- 2️⃣ 查询分页数据 ----------
    result = await db.execute(
        stmt
        .offset((page - 1) * size)
        .limit(size)
    )
    items = result.scalars().all()

    return {
        "items": items,
        "total": total,
        "page": page,
        "size": size,
    }