Architecture: Standardize backend pagination query schemas and list responses

[Shevilll]

Description

As the InPactAI platform continues to scale, our data models will naturally accumulate larger volumes of records. Currently, many of our list-retrieval endpoints in the FastAPI backend (such as /users/, /posts/, /sponsorships/, /sponsorship-applications/, /collaborations/, etc.) perform unrestricted .select("*") queries against Supabase without any form of pagination.

This introduces several architectural risks:

1. Performance Degradation: Fetching all records from a table in a single request degrades database response times and increases payload sizes.
2. Memory Overheads: Unrestricted data fetches increase memory consumption on both the FastAPI server and the client frontend.
3. Inefficient Network I/O: Returning unnecessary data wastes network bandwidth.

To address these concerns, we should establish a standardized, unified pagination framework for all backend list endpoints. This will ensure consistency, keep our API controllers clean and DRY (Don't Repeat Yourself), and prepare our API for scaling.

---

Proposed Architectural Design

We propose implementing two distinct pagination strategies, each suited to different types of endpoints:

1. Offset-Based Pagination (Limit / Offset)

Best for static tables, administration dashboards, or lists where jumping to specific pages is required (e.g., /users/, /sponsorship-payments/, /sponsorship-applications/).

Unified Query Parameters (Dependency)

We can leverage FastAPI's dependency injection system to define reusable query parameters:

from fastapi import Query
from pydantic import BaseModel

class PaginationParams(BaseModel):
    limit: int = 20
    offset: int = 0

async def get_pagination_params(
    limit: int = Query(default=20, ge=1, le=100, description="Maximum number of items to return"),
    offset: int = Query(default=0, ge=0, description="Number of items to skip")
) -> PaginationParams:
    return PaginationParams(limit=limit, offset=offset)

Standardized Response Model

A generic Pydantic envelope ensuring all list responses include consistent metadata:

from typing import Generic, TypeVar, List
from pydantic import BaseModel

T = TypeVar("T")

class PaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    limit: int
    offset: int
    has_more: bool

Controller Helper (Supabase Integration)

To keep API controllers DRY, we can introduce a utility that translates these parameters into Supabase query builders:

from supabase import Client

async def paginate_supabase_query(
    supabase_client: Client,
    table_name: str,
    params: PaginationParams,
    select_query: str = "*"
) -> dict:
    # Use count='exact' to get the total number of matching items
    response = supabase_client.table(table_name) \
        .select(select_query, count="exact") \
        .range(params.offset, params.offset + params.limit - 1) \
        .execute()
        
    total = response.count if response.count is not None else 0
    items = response.data
    has_more = params.offset + params.limit < total
    
    return {
        "items": items,
        "total": total,
        "limit": params.limit,
        "offset": params.offset,
        "has_more": has_more
    }

---

2. Cursor-Based Pagination (Token-based)

Best for high-frequency or rapidly changing lists, such as social feeds, messages, or chat histories (e.g., /posts/, /collaborations/, /chat/). Cursor-based pagination avoids duplicate items or skipped items when new records are created while paginating.

Unified Query Parameters (Dependency)

from typing import Optional
from fastapi import Query
from pydantic import BaseModel

class CursorParams(BaseModel):
    limit: int = 20
    cursor: Optional[str] = None

async def get_cursor_params(
    limit: int = Query(default=20, ge=1, le=100, description="Maximum number of items to return"),
    cursor: Optional[str] = Query(default=None, description="Base64 encoded cursor indicating page offset / timestamp")
) -> CursorParams:
    return CursorParams(limit=limit, cursor=cursor)

Standardized Response Model

class CursorPaginatedResponse(BaseModel, Generic[T]):
    items: List[T]
    limit: int
    next_cursor: Optional[str]
    has_more: bool

---

Implementation Steps

1. Define Schemas: Create a new file Backend/app/schemas/pagination.py containing the core query and response schemas.
2. Build Utilities: Create a backend utility module Backend/app/services/pagination.py or similar helper functions to automate the range selection logic in Supabase queries.
3. Refactor Endpoints: Update existing list endpoints under Backend/app/routes/ (e.g., /posts/, /users/, /sponsorships/) to accept pagination dependencies and return standardized envelopes.
4. Update Frontend Client: Update the frontend query utilities to correctly handle and parse the new structured pagination response format.

Establishing this standardized contract early in development will greatly simplify frontend integration and guarantee API performance in the long run!