47 lines
2.2 KiB
Python
47 lines
2.2 KiB
Python
"""Pagination + sorting helpers."""
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import InstrumentedAttribute
|
|
|
|
from app.config import settings
|
|
from app.utils.errors import ValidationError
|
|
|
|
|
|
async def paginate(query, page: int, page_size: int, db: AsyncSession):
|
|
"""Apply offset/limit to `query` and return (items, meta dict).
|
|
|
|
`page_size` is clamped to [1, MAX_PAGE_SIZE] after validation — routers
|
|
enforce the contract bounds via Query(ge=1, le=100) before calling this.
|
|
"""
|
|
total = await db.scalar(select(func.count()).select_from(query.subquery()))
|
|
items = (await db.execute(query.offset((page - 1) * page_size).limit(page_size))).scalars().all()
|
|
return items, {"total": total or 0, "page": page, "page_size": page_size}
|
|
|
|
|
|
def sort_query(stmt, model, sort: str | None, default: str = "-created_at"):
|
|
"""Apply `?sort=-created_at` (leading `-` = descending) against a model.
|
|
|
|
Only real columns of the model are accepted — anything else is a
|
|
VALIDATION_ERROR (400), never a raw SQL fragment.
|
|
"""
|
|
raw = (sort or default).strip()
|
|
descending = raw.startswith("-")
|
|
name = raw[1:] if descending else raw
|
|
column: InstrumentedAttribute | None = getattr(model, name, None)
|
|
if column is None or not isinstance(column, InstrumentedAttribute):
|
|
raise ValidationError(f"فیلد مرتبسازی نامعتبر است: {name}", field="sort")
|
|
return stmt.order_by(column.desc() if descending else column.asc())
|
|
|
|
|
|
def default_pagination(page: int | None, page_size: int | None) -> tuple[int, int]:
|
|
"""Contract defaults: page=1, page_size=20 (max 100) with 400 on out-of-bounds."""
|
|
p = page if page is not None else 1
|
|
ps = page_size if page_size is not None else settings.DEFAULT_PAGE_SIZE
|
|
if p < 1:
|
|
raise ValidationError("شماره صفحه باید بزرگتر از صفر باشد", field="page")
|
|
if ps < 1:
|
|
raise ValidationError("اندازه صفحه باید بزرگتر از صفر باشد", field="page_size")
|
|
if ps > settings.MAX_PAGE_SIZE:
|
|
raise ValidationError(f"اندازه صفحه نمیتواند از {settings.MAX_PAGE_SIZE} بیشتر باشد", field="page_size")
|
|
return p, ps
|