52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
"""
|
|
Pagination schemas and utilities for multi-user server deployment.
|
|
Supports cursor-based and offset-based pagination.
|
|
"""
|
|
from typing import TypeVar, Generic, List, Optional
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy.orm import Session, Query
|
|
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class PaginationParams(BaseModel):
|
|
"""Query params for pagination."""
|
|
page: int = Field(default=1, ge=1, description="Page number (1-based)")
|
|
per_page: int = Field(default=25, ge=1, le=100, description="Items per page (max 100)")
|
|
|
|
@property
|
|
def offset(self) -> int:
|
|
return (self.page - 1) * self.per_page
|
|
|
|
|
|
class PaginatedResponse(BaseModel, Generic[T]):
|
|
"""Standard paginated response wrapper."""
|
|
items: List[T]
|
|
total: int
|
|
page: int
|
|
per_page: int
|
|
pages: int
|
|
|
|
@classmethod
|
|
def from_query(cls, query: Query, params: PaginationParams, schema_cls=None):
|
|
total = query.count()
|
|
items = query.offset(params.offset).limit(params.per_page).all()
|
|
if schema_cls:
|
|
items = [schema_cls.model_validate(i) for i in items]
|
|
pages = (total + params.per_page - 1) // params.per_page
|
|
return cls(
|
|
items=items,
|
|
total=total,
|
|
page=params.page,
|
|
per_page=params.per_page,
|
|
pages=pages,
|
|
)
|
|
|
|
|
|
def paginate(db: Session, query, params: PaginationParams):
|
|
"""Apply pagination to a SQLAlchemy query. Returns (items, total)."""
|
|
total = query.count()
|
|
items = query.offset(params.offset).limit(params.per_page).all()
|
|
return items, total
|