28 lines
874 B
Python
28 lines
874 B
Python
from sqlalchemy import select, func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from schemas.pagination import PageResult
|
|
|
|
|
|
def paginate_query(db: Session, query, current_page: int = 1, page_size: int = 10) -> PageResult:
|
|
"""通用分页查询函数"""
|
|
# 计算总记录数
|
|
total = db.execute(select(func.count()).select_from(query.subquery())).scalar_one_or_none() or 0
|
|
|
|
# 计算总页数
|
|
total_pages = (total + page_size - 1) // page_size if page_size != 0 else 0
|
|
|
|
# 执行分页查询
|
|
results = db.execute(query.offset((current_page - 1) * page_size).limit(page_size)).all()
|
|
|
|
# 转换为字典列表
|
|
records = [row._asdict() if hasattr(row, "_asdict") else dict(row) for row in results]
|
|
|
|
return PageResult(
|
|
current=current_page,
|
|
size=page_size,
|
|
total=total,
|
|
pages=total_pages,
|
|
records=records
|
|
)
|