diff --git a/backend/pagination.py b/backend/pagination.py new file mode 100644 index 0000000..28cb412 --- /dev/null +++ b/backend/pagination.py @@ -0,0 +1,46 @@ +"""Pagination utilities for API responses.""" + +from typing import TypeVar + +from pydantic import BaseModel + +from schemas import PaginatedResponse + +RecordT = TypeVar("RecordT", bound=BaseModel) + + +def calculate_total_pages(total: int, per_page: int) -> int: + """Calculate total number of pages.""" + return (total + per_page - 1) // per_page if total > 0 else 1 + + +def calculate_offset(page: int, per_page: int) -> int: + """Calculate the offset for a given page.""" + return (page - 1) * per_page + + +def create_paginated_response( + records: list[RecordT], + total: int, + page: int, + per_page: int, +) -> PaginatedResponse[RecordT]: + """ + Create a paginated response with calculated metadata. + + Args: + records: List of records for the current page + total: Total count of all records (before pagination) + page: Current page number (1-indexed) + per_page: Number of records per page + + Returns: + A PaginatedResponse with all pagination metadata + """ + return PaginatedResponse[RecordT]( + records=records, + total=total, + page=page, + per_page=per_page, + total_pages=calculate_total_pages(total, per_page), + ) diff --git a/backend/routes/audit.py b/backend/routes/audit.py index 83cf60d..388d216 100644 --- a/backend/routes/audit.py +++ b/backend/routes/audit.py @@ -11,6 +11,11 @@ from sqlalchemy.ext.asyncio import AsyncSession from auth import require_permission from database import get_db from models import CounterRecord, Permission, RandomNumberOutcome, SumRecord, User +from pagination import ( + calculate_offset, + calculate_total_pages, + create_paginated_response, +) from schemas import ( CounterRecordResponse, PaginatedCounterRecords, @@ -39,10 +44,9 @@ async def paginate_with_user_email( # Get total count count_result = await db.execute(select(func.count(model.id))) total = count_result.scalar() or 0 - total_pages = (total + per_page - 1) // per_page if total > 0 else 1 # Get paginated records with user email - offset = (page - 1) * per_page + offset = calculate_offset(page, per_page) query = ( select(model, User.email) .join(User, model.user_id == User.id) @@ -54,7 +58,7 @@ async def paginate_with_user_email( rows = result.all() records: list[R] = [row_mapper(record, email) for record, email in rows] - return records, total, total_pages + return records, total, calculate_total_pages(total, per_page) def _map_counter_record(record: CounterRecord, email: str) -> CounterRecordResponse: @@ -86,16 +90,10 @@ async def get_counter_records( _current_user: User = Depends(require_permission(Permission.VIEW_AUDIT)), ) -> PaginatedCounterRecords: """Get paginated counter action records.""" - records, total, total_pages = await paginate_with_user_email( + records, total, _ = await paginate_with_user_email( db, CounterRecord, page, per_page, _map_counter_record ) - return PaginatedCounterRecords( - records=records, - total=total, - page=page, - per_page=per_page, - total_pages=total_pages, - ) + return create_paginated_response(records, total, page, per_page) @router.get("/sum", response_model=PaginatedSumRecords) @@ -106,16 +104,10 @@ async def get_sum_records( _current_user: User = Depends(require_permission(Permission.VIEW_AUDIT)), ) -> PaginatedSumRecords: """Get paginated sum action records.""" - records, total, total_pages = await paginate_with_user_email( + records, total, _ = await paginate_with_user_email( db, SumRecord, page, per_page, _map_sum_record ) - return PaginatedSumRecords( - records=records, - total=total, - page=page, - per_page=per_page, - total_pages=total_pages, - ) + return create_paginated_response(records, total, page, per_page) @router.get("/random-jobs", response_model=list[RandomNumberOutcomeResponse]) diff --git a/backend/routes/booking.py b/backend/routes/booking.py index b7cce58..b91178d 100644 --- a/backend/routes/booking.py +++ b/backend/routes/booking.py @@ -12,6 +12,7 @@ from sqlalchemy.orm import joinedload from auth import require_permission from database import get_db from models import Appointment, AppointmentStatus, Availability, Permission, User +from pagination import calculate_offset, create_paginated_response from schemas import ( AppointmentResponse, AvailableSlotsResponse, @@ -325,10 +326,9 @@ async def get_all_appointments( # Get total count count_result = await db.execute(select(func.count(Appointment.id))) total = count_result.scalar() or 0 - total_pages = (total + per_page - 1) // per_page if total > 0 else 1 # Get paginated appointments with explicit eager loading of user relationship - offset = (page - 1) * per_page + offset = calculate_offset(page, per_page) result = await db.execute( select(Appointment) .options(joinedload(Appointment.user)) @@ -344,13 +344,7 @@ async def get_all_appointments( for apt in appointments ] - return PaginatedAppointments( - records=records, - total=total, - page=page, - per_page=per_page, - total_pages=total_pages, - ) + return create_paginated_response(records, total, page, per_page) @admin_appointments_router.post( diff --git a/backend/routes/invites.py b/backend/routes/invites.py index 741c280..5a765e9 100644 --- a/backend/routes/invites.py +++ b/backend/routes/invites.py @@ -15,6 +15,7 @@ from invite_utils import ( normalize_identifier, ) from models import Invite, InviteStatus, Permission, User +from pagination import calculate_offset, create_paginated_response from schemas import ( AdminUserResponse, InviteCheckResponse, @@ -188,10 +189,9 @@ async def list_all_invites( # Get total count count_result = await db.execute(count_query) total = count_result.scalar() or 0 - total_pages = (total + per_page - 1) // per_page if total > 0 else 1 # Get paginated invites (relationships loaded via selectin) - offset = (page - 1) * per_page + offset = calculate_offset(page, per_page) query = query.order_by(desc(Invite.created_at)).offset(offset).limit(per_page) result = await db.execute(query) invites = result.scalars().all() @@ -199,13 +199,7 @@ async def list_all_invites( # Build responses using preloaded relationships records = [build_invite_response(invite) for invite in invites] - return PaginatedInviteRecords( - records=records, - total=total, - page=page, - per_page=per_page, - total_pages=total_pages, - ) + return create_paginated_response(records, total, page, per_page) @admin_router.post("/invites/{invite_id}/revoke", response_model=InviteResponse)