"""Generic pagination utilities for large datasets. Provides: - approximate_count: Fast count via pg_class.reltuples (no Seq Scan) - paginated_list: Generic keyset/offset pagination for any model """ from __future__ import annotations import uuid from typing import Any, TypeVar from sqlalchemy import func, select, text from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.sql import Select T = TypeVar("T") async def approximate_count(db: AsyncSession, table_name: str) -> int: """Get approximate row count via pg_class.reltuples. This is ~5000x faster than SELECT count(*) on 1M rows because it reads pre-computed statistics instead of scanning the table. Accuracy: ~95-99% (updated by ANALYZE/VACUUM). """ result = await db.execute( text("SELECT reltuples::bigint FROM pg_class WHERE relname = :name"), {"name": table_name}, ) count = result.scalar() return int(count) if count is not None else 0 async def exact_count(db: AsyncSession, base_query: Select) -> int: """Get exact count via SELECT count(*). Use this for small tables or when exact count is required. For large tables (>100k rows), use approximate_count instead. """ count_q = select(func.count()).select_from(base_query.subquery()) result = await db.execute(count_q) return result.scalar() or 0 async def paginated_list( db: AsyncSession, base_query: Select, model: Any, page: int = 1, page_size: int = 20, cursor: str | None = None, sort_by: str = "id", sort_order: str = "asc", use_approximate_count: bool = False, table_name: str | None = None, serializer=None, ) -> dict[str, Any]: """Generic paginated list with keyset and offset support. Args: db: Async session base_query: Base SELECT query (with filters applied, without pagination) model: SQLAlchemy model class (for sort column access) page: Page number (for offset pagination) page_size: Items per page cursor: Keyset cursor (model UUID) — if provided with sort_by='id', uses keyset sort_by: Column name to sort by sort_order: 'asc' or 'desc' use_approximate_count: If True, use pg_class.reltuples instead of count(*) table_name: Table name for approximate count (required if use_approximate_count=True) serializer: Function to serialize each item Returns: dict with items, total, page, page_size, next_cursor """ # Determine pagination mode use_keyset = cursor is not None and sort_by == "id" and sort_order == "asc" # Apply keyset filter if cursor provided query = base_query if use_keyset and cursor is not None: query = query.where(model.id > uuid.UUID(cursor)) # Count if use_approximate_count and table_name: total = await approximate_count(db, table_name) else: count_q = select(func.count()).select_from(query.subquery()) total = (await db.execute(count_q)).scalar() or 0 # Sort sort_col = getattr(model, sort_by, getattr(model, "id", None)) if sort_col is None: sort_col = model.id if sort_order == "desc": sort_col = sort_col.desc() query = query.order_by(sort_col) # Paginate if use_keyset: query = query.limit(page_size) else: offset = (page - 1) * page_size query = query.offset(offset).limit(page_size) result = await db.execute(query) items = result.scalars().all() # Next cursor for keyset pagination next_cursor = None if use_keyset and len(items) == page_size and items: next_cursor = str(items[-1].id) # Serialize serialized = [serializer(item) for item in items] if serializer else items return { "items": serialized, "total": total, "page": page, "page_size": page_size, "next_cursor": next_cursor, }