Spaces:
Running
Running
File size: 1,200 Bytes
f01ec67 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 | from fastapi import HTTPException
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSession
from typing import TypeVar, Any, Tuple, List
T = TypeVar("T")
async def paginate(db: AsyncSession, query: Any, page: int, limit: int) -> Tuple[List[T], int]:
"""Generic pagination. Returns (items, total_count)"""
count_query = select(func.count()).select_from(query.subquery())
total_count = await db.scalar(count_query)
offset = (page - 1) * limit
paginated_query = query.offset(offset).limit(limit)
result = await db.execute(paginated_query)
items = result.scalars().all()
return items, total_count
async def get_or_404(db: AsyncSession, model: type[T], id: str) -> T:
"""Get by ID or raise HTTPException 404"""
obj = await db.get(model, id)
if not obj:
raise HTTPException(status_code=404, detail=f"{model.__name__} not found")
return obj
async def exists(db: AsyncSession, model: type[T], **filters) -> bool:
"""Check if record exists matching filters"""
query = select(model).filter_by(**filters)
result = await db.execute(query.limit(1))
return result.scalar_one_or_none() is not None
|