How to Implement Cursor-Based Pagination with FastCRUD for Large Datasets

FastCRUD provides built-in cursor-based pagination through the CursorPaginatedRequestQuery schema and automatic validation dependencies, enabling efficient infinite-scroll patterns for large SQLAlchemy datasets without offset-based performance degradation.

Cursor-based pagination is essential for high-performance APIs serving large datasets, eliminating the slow OFFSET queries that plague traditional pagination. FastCRUD, an open-source FastAPI CRUD generator, ships with a complete cursor pagination implementation that validates cursors against SQLAlchemy column types and handles integer ranges, datetime strings, and UUIDs automatically.

How Cursor-Based Pagination Works in FastCRUD

FastCRUD’s cursor pagination splits responsibilities across three distinct layers to ensure type safety and performance:

Component	Responsibility	Location
Request Schema	Defines cursor, limit, sort_column, and sort_order with automatic type coercion	fastcrud/core/pagination.py – CursorPaginatedRequestQuery
Validator	Inspects SQLAlchemy column types and raises BadRequestException for malformed or out-of-range cursors	fastcrud/endpoint/endpoint_creator.py – _validate_cursor_value
FastAPI Integration	Exposes validation as a reusable dependency for auto-generated or custom endpoints	fastcrud/endpoint/endpoint_creator.py – _create_cursor_validator

The CursorPaginatedRequestQuery Schema

The foundation of FastCRUD’s pagination is the CursorPaginatedRequestQuery Pydantic model, located in fastcrud/core/pagination.py (lines 128–150). This schema handles input validation and automatically coerces string values to appropriate types:

# fastcrud/core/pagination.py

class CursorPaginatedRequestQuery(BaseModel):
    """
    Pydantic model for cursor‑based pagination query parameters.
    """
    cursor: int | str | None = Field(
        None,
        description="Cursor value for pagination (typically the ID of the last item from previous page). Supports int, datetime (ISO format), or UUID string.",
    )
    limit: int | None = Field(100, description="Maximum number of items to return per page", gt=0, le=1000)
    sort_column: str | None = Field("id", description="Column name to sort by")
    sort_order: str | None = Field(
        "asc",
        description="Sort order: 'asc' for ascending, 'desc' for descending",
        pattern="^(asc|desc)$",
    )
    model_config = {"populate_by_name": True}

    @field_validator("cursor", mode="before")
    @classmethod
    def coerce_cursor(cls, v):
        """Try to coerce string cursor to int if it represents a valid integer."""
        if v is None:
            return None
        if isinstance(v, int):
            return v
        if isinstance(v, str):
            try:
                return int(v)          # integer cursor → keep as int

            except ValueError:
                return v               # keep as string for datetime / UUID

        return v

The coerce_cursor validator ensures that numeric string parameters like "42" are converted to integers, while datetime strings and UUIDs remain as strings for subsequent type-specific validation.

Validating Cursor Values Against Column Types

Before executing queries, FastCRUD validates cursors against the actual SQLAlchemy column types to prevent type mismatches and out-of-range errors. This logic resides in fastcrud/endpoint/endpoint_creator.py.

Type-Specific Validation Logic

The _validate_cursor_value method (lines 460–508) inspects the SQLAlchemy type and Python type to enforce strict validation:

# fastcrud/endpoint/endpoint_creator.py

def _validate_cursor_value(
    self,
    cursor: Any,
    sqlalchemy_type: Any,
    python_type: Any,
    column_name: str,
) -> None:
    # Integer validation (including range checks for SmallInteger, Integer, BigInteger)

    if sqlalchemy_type in (Integer, SmallInteger, BigInteger):
        try:
            cursor_int = int(cursor)
            # Range constants – see lines 462‑466

            INT16_MIN, INT16_MAX = -32768, 32767          # SmallInteger

            INT32_MIN, INT32_MAX = -2147483648, 2147483647  # Integer

            INT64_MIN, INT64_MAX = -9223372036854775808, 9223372036854775807  # BigInteger

            # Choose appropriate range

            if isinstance(sqlalchemy_type, BigInteger):
                min_val, max_val, type_name = INT64_MIN, INT64_MAX, "BIGINT"
            elif isinstance(sqlalchemy_type, SmallInteger):
                min_val, max_val, type_name = INT16_MIN, INT16_MAX, "SMALLINT"
            else:
                min_val, max_val, type_name = INT32_MIN, INT32_MAX, "INTEGER"
            if cursor_int < min_val or cursor_int > max_val:
                raise BadRequestException(
                    detail=f"Cursor value {cursor} exceeds valid {type_name} range ({min_val} to {max_val}) for column '{column_name}'"
                )
        except (ValueError, TypeError):
            raise BadRequestException(
                detail=f"Invalid cursor value for integer column '{column_name}': {cursor}"
            )
        return

    # Datetime validation

    if python_type == datetime:
        if isinstance(cursor, str):
            try:
                datetime.fromisoformat(cursor.replace('Z', '+00:00'))
            except (ValueError, TypeError):
                raise BadRequestException(
                    detail=f"Invalid cursor value for datetime column '{column_name}': {cursor}. Expected ISO 8601 format (e.g., '2024-01-01T12:00:00')."
                )
        return

    # UUID validation

    if python_type == UUID:
        if isinstance(cursor, str):
            try:
                UUID(cursor)
            except (ValueError, TypeError):
                raise BadRequestException(
                    detail=f"Invalid cursor value for UUID column '{column_name}': {cursor}. Expected valid UUID format."
                )
        return

This implementation guards against integer overflow (checking SmallInteger (-32,768 to 32,767), Integer (-2,147,483,648 to 2,147,483,647), and BigInteger ranges), malformed datetime strings, and invalid UUID formats, raising BadRequestException before any database query executes.

Creating the Validator Dependency

The _create_cursor_validator factory (lines 522–554) generates FastAPI dependencies that automatically validate cursors against model column types:

# fastcrud/endpoint/endpoint_creator.py

def _create_cursor_validator(self) -> Callable:
    """
    Returns a FastAPI dependency that validates cursor arguments.
    """
    def validate_pagination_args(
        query: CursorPaginatedRequestQuery = Depends(),
    ) -> CursorPaginatedRequestQuery:
        # If a cursor is supplied, look up the column types for the requested sort column

        if query.cursor is not None and query.sort_column:
            sqlalchemy_type = self.sqlalchemy_column_types.get(query.sort_column)
            python_type = self.column_types.get(query.sort_column)
            self._validate_cursor_value(query.cursor, sqlalchemy_type, python_type, query.sort_column)
        return query
    return validate_pagination_args

This dependency can be injected into any endpoint, ensuring that query.cursor is validated against the actual database column type before the CRUD operation proceeds.

Implementing Cursor Pagination in FastAPI Endpoints

FastCRUD offers two integration paths: automatic generation via the FastCrud builder or manual validation in custom endpoints.

Enabling Cursor Pagination in Auto-Generated Routers

The simplest approach uses the read_pagination parameter when constructing a FastCrud instance:

from fastcrud import FastCrud
from fastapi import FastAPI
from models import Item
from schemas import ItemCreate, ItemRead

app = FastAPI()

router = FastCrud(
    model=Item,
    create_schema=ItemCreate,
    read_schema=ItemRead,
    read_pagination="cursor",   # Enable cursor mode globally

).router

app.include_router(router, prefix="/items")

Setting read_pagination="cursor" automatically attaches the cursor validator to the generated GET /items endpoint. The endpoint accepts cursor, limit, sort_column, and sort_order query parameters with full validation.

Manual Validation in Custom Endpoints

For endpoints requiring additional filters or custom logic, reuse the validator directly:

from fastapi import APIRouter, Depends, Query
from fastcrud import CursorPaginatedRequestQuery, FastCrud
from models import Item
from schemas import ItemRead

router = APIRouter()
crud = FastCrud(model=Item, read_schema=ItemRead)
validator = crud._create_cursor_validator()

@router.get("/custom-items")
async def custom_items(
    query: CursorPaginatedRequestQuery = Depends(validator),
    custom_filter: str | None = Query(None),
):
    """
    Custom endpoint with cursor pagination validation.
    """
    return await crud.crud.get_multi(
        db=await crud.session(),
        limit=query.limit,
        cursor=query.cursor,
        sort_column=query.sort_column,
        sort_order=query.sort_order,
        extra_filter=custom_filter,
    )

The validator dependency raises BadRequestException before the endpoint function executes if the cursor format does not match the sort_column type.

Handling Pagination Responses

FastCRUD returns a standardized dictionary containing:

• data: List of items for the current page
• has_more: Boolean indicating if additional rows exist
• total_count: Total matching rows (if requested)

To implement infinite scroll, extract the next_cursor from the last item:

result = await crud.crud.get_multi(...)
next_cursor = None
if result["has_more"]:
    last_item = result["data"][-1]
    next_cursor = getattr(last_item, sort_column)

return {
    "items": result["data"],
    "next_cursor": next_cursor,
    "has_more": result["has_more"],
}

Complete Working Example

The following implementation demonstrates a production-ready infinite-scroll endpoint:

# demo_app.py

from fastapi import FastAPI, Depends, APIRouter
from fastcrud import FastCrud, CursorPaginatedRequestQuery
from models import Article
from schemas import ArticleRead, ArticleCreate

app = FastAPI()
router = APIRouter()

article_crud = FastCrud(
    model=Article,
    create_schema=ArticleCreate,
    read_schema=ArticleRead,
    read_pagination="cursor",
)

cursor_validator = article_crud._create_cursor_validator()

@router.get("/articles")
async def list_articles(
    query: CursorPaginatedRequestQuery = Depends(cursor_validator),
):
    """
    Returns paginated articles for infinite-scroll UIs.
    """
    data = await article_crud.crud.get_multi(
        db=await article_crud.session(),
        limit=query.limit,
        cursor=query.cursor,
        sort_column=query.sort_column,
        sort_order=query.sort_order,
    )
    
    next_cursor = None
    if data["has_more"]:
        last_item = data["data"][-1]
        next_cursor = getattr(last_item, query.sort_column)
    
    return {
        "items": data["data"],
        "next_cursor": next_cursor,
        "has_more": data["has_more"],
    }

app.include_router(router, prefix="/api")

Requests to GET /api/articles?limit=50&sort_column=created_at&sort_order=desc return the first page, while subsequent requests include cursor=2024-01-15T08:23:00Z to fetch the next slice with proper validation.

Key Source Files and References

File	Purpose	Key Lines
fastcrud/core/pagination.py	Request schema definition	CursorPaginatedRequestQuery at lines 128–150
fastcrud/endpoint/endpoint_creator.py	Core validation logic	_validate_cursor_value at lines 460–508; integer range constants at lines 462–466
fastcrud/endpoint/endpoint_creator.py	Dependency factory	_create_cursor_validator at lines 522–554
fastcrud/core/__init__.py	Public exports	Re-exports CursorPaginatedRequestQuery
tests/sqlalchemy/endpoint/test_cursor_validation.py	Validation tests	Edge cases for integer overflow and type mismatches
tests/sqlalchemy/endpoint/test_cursor_paginated_request_query_reusability.py	Schema tests	Reusability and coercion logic

Summary

• FastCRUD implements cursor-based pagination via the CursorPaginatedRequestQuery schema and automatic validation dependencies, eliminating the performance penalties of offset-based queries.
• The CursorPaginatedRequestQuery class in fastcrud/core/pagination.py handles input coercion, converting numeric strings to integers while preserving datetime and UUID strings.
• Type-safe validation occurs in _validate_cursor_value (lines 460–508 of endpoint_creator.py), which enforces integer range limits for SmallInteger, Integer, and BigInteger, and validates ISO 8601 datetime and UUID formats.
• The _create_cursor_validator factory (lines 522–554) generates FastAPI dependencies that automatically validate cursors against model column types before endpoint execution.
• Enable automatic pagination by setting read_pagination="cursor" in the FastCrud constructor, or manually inject the validator into custom endpoints using Depends(validator).

Frequently Asked Questions

What is cursor-based pagination and why use it with FastCRUD?

Cursor-based pagination (also called "keyset" pagination) uses a unique value from the last item of a page to fetch the next set of results, rather than using OFFSET and LIMIT. This approach maintains consistent performance regardless of dataset size, avoiding the slowdown that occurs when querying deep pages in large tables. FastCRUD implements this pattern with automatic type validation, making it ideal for infinite-scroll UIs and high-throughput APIs.

How does FastCRUD validate cursor values?

FastCRUD validates cursors through the _validate_cursor_value method in fastcrud/endpoint/endpoint_creator.py (lines 460–508). The validator inspects the SQLAlchemy column type and applies specific checks: it enforces range limits for SmallInteger (-32,768 to 32,767), Integer (-2,147,483,648 to 2,147,483,647), and BigInteger; validates ISO 8601 datetime strings; and verifies UUID format. If validation fails, it raises BadRequestException before any database query executes.

Can I use cursor pagination with custom SQLAlchemy models?

Yes, cursor pagination works with any SQLAlchemy model that uses supported column types for the sort column (integer variants, datetime, or UUID). To use it in auto-generated endpoints, instantiate FastCrud with read_pagination="cursor". For custom endpoints, call _create_cursor_validator() on your FastCrud instance to obtain a dependency that validates the CursorPaginatedRequestQuery against your model's column types, then inject it using Depends().

What is the maximum page size limit in FastCRUD cursor pagination?

The default page size is 100 items, and the maximum allowed limit is 1000 items per request. These constraints are enforced by the CursorPaginatedRequestQuery schema in fastcrud/core/pagination.py, which uses Pydantic's Field constraints (gt=0, le=1000) to validate the limit parameter. This protects your database from excessive load while still allowing large page sizes for bulk data exports when necessary.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}