feat: add native forward cursor pagination

README.md

@@ -63,7 +63,8 @@ following [`SQLAlchemy`'s best practices](https://docs.sqlalchemy.org/en/20/orm/
             ...
     ```
 
-* Built-in pagination:
+* Built-in pagination with offset/limit and
+    [forward cursors](https://hadrien.github.io/FastSQLA/pagination/#forward-only-cursor-pagination):
 
     ```python
     ...

docs/pagination.md

@@ -64,6 +64,51 @@ async def list_heros(paginate: Paginate, age:int | None = None):
     return await paginate(stmt)
 ```
 
+## Forward-only cursor pagination
+
+- `CursorPaginate[T]`: `cursor` and `limit` query parameters.
+- `CursorPage[T]`: `data` and `meta.next_cursor`; `null` marks the end.
+- For JSON input, use `new_cursor_pagination()` as below (`Hero` and `HeroModel` from above).
+
+```python
+from typing import Literal
+from fastsqla import CursorPage, Session, new_cursor_pagination
+from pydantic import BaseModel, ConfigDict, Field
+
+cursor_dependency = new_cursor_pagination(default_page_size=10, max_page_size=100)
+
+class HeroSearch(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    cursor: str | None = Field(None, min_length=1)
+    limit: int = Field(10, ge=1, le=100)
+    min_age: int | None = Field(None, ge=0)
+    order_by: Literal["age", "name"] = "age"
+
+@app.post("/heroes/search")
+async def search_heroes(body: HeroSearch, session: Session) -> CursorPage[HeroModel]:
+    column = {"age": Hero.age, "name": Hero.name}[body.order_by]
+    stmt = select(Hero).order_by(column, Hero.id)
+    if body.min_age is not None:
+        stmt = stmt.where(Hero.age >= body.min_age)
+    paginate = cursor_dependency(session=session, cursor=body.cursor, limit=body.limit)
+    return await paginate(stmt)
+```
+
+- POST body: `{"min_age": 18, "order_by": "name", "limit": 10, "cursor": null}`.
+- Omit `cursor` for page one; send `meta.next_cursor` to continue. Keep filters and ordering
+  fixed; reapply authorization each request.
+- Direct calls require explicit `session`, `cursor`, and `limit`. Validate body values and
+  keep page-size limits aligned with the factory.
+- Order by non-null columns with a unique tie-breaker, including across joins. Ascending,
+  descending, and mixed directions are supported.
+- Unsupported: expressions, nullable ordering, outer joins, grouping/distinct/unions,
+  existing limits/offsets, and deduplication.
+- Default mapping: `row[0]`. For projections, use `row_mapper=lambda row: row._mapping`.
+  Map each SQL row to one item; ordering columns need not appear in the response.
+- Invalid cursors return HTTP 422. FastSQLA imposes no cursor-length cap.
+- Cursors expose ordering values. Changing those values during traversal can skip or
+  repeat items; prefer immutable columns and matching indexes.
+
 ## `SQLModel` example
 
 ```python

plugin/fastsqla/skills/fastsqla-pagination/SKILL.md

@@ -2,7 +2,7 @@
 name: fastsqla-pagination
 description: >
   Paginate SQLAlchemy select queries in FastAPI endpoints using FastSQLA.
-  Covers the built-in Paginate dependency (offset/limit query params),
+  Covers Paginate (offset/limit) and CursorPaginate (forward-only cursor/limit),
   Page/Item/Collection response models, and the new_pagination() factory
   for custom page sizes, count queries, and result processing.
 ---
@@ -13,7 +13,15 @@ FastSQLA provides a `Paginate` dependency that adds `offset` and `limit` query p
 
 ## Response Models
 
-FastSQLA exports three generic response wrappers:
+FastSQLA exports these generic response wrappers:
+
+### `CursorPage[T]` — forward-only cursor pagination
+
+Use `CursorPaginate[T]` with non-null column ordering and a unique tie-breaker.
+Pass `meta.next_cursor` as `cursor` until it is null; metadata contains no other fields.
+`new_cursor_pagination()` configures page sizes and a `row_mapper` that preserves row count.
+Keep filters fixed and reapply authorization. Cursors expose values and read live data.
+See the [cursor guide](https://hadrien.github.io/FastSQLA/pagination/#forward-only-cursor-pagination) for supported queries.
 
 ### `Page[T]` — paginated list with metadata
 

src/fastsqla.py

@@ -1,14 +1,21 @@
+import base64
 import functools
+import json
 import math
 import os
+import re
 import warnings
 from collections.abc import AsyncGenerator, Awaitable, Callable, Iterable
 from contextlib import _AsyncGeneratorContextManager, asynccontextmanager
+from datetime import date, datetime
+from decimal import Decimal
 from typing import Annotated, TypedDict, TypeVar
+from uuid import UUID
 
+import sqlalchemy as sa
 from fastapi import Depends as BaseDepends
-from fastapi import FastAPI, Query
-from pydantic import BaseModel, Field
+from fastapi import FastAPI, HTTPException, Query
+from pydantic import BaseModel, Field, TypeAdapter
 from sqlalchemy import Result, Select, func, select
 from sqlalchemy.ext.asyncio import (
     AsyncEngine,
@@ -18,6 +25,9 @@
 )
 from sqlalchemy.ext.declarative import DeferredReflection
 from sqlalchemy.orm import DeclarativeBase
+from sqlalchemy.sql import operators, visitors
+from sqlalchemy.sql.elements import Label, UnaryExpression
+from sqlalchemy.sql.selectable import Join
 from structlog import get_logger
 
 logger = get_logger(__name__)
@@ -32,13 +42,18 @@
 __all__ = [
     "Base",
     "Collection",
+    "CursorMeta",
+    "CursorPage",
+    "CursorPaginate",
+    "CursorPaginateType",
     "Item",
     "MissingConfigurationError",
     "Page",
     "Paginate",
     "PaginateType",
     "Session",
     "lifespan",
+    "new_cursor_pagination",
     "new_pagination",
     "open_session",
 ]
@@ -491,3 +506,198 @@ async def paginate(stmt: Select) -> Page:
 It adds **`offset`** and **`limit`** query parameters to the endpoint, which are used to
 paginate. The model returned by the endpoint is a [`Page`][fastsqla.Page] model.
 """
+
+
+class CursorMeta(BaseModel):
+    next_cursor: str | None = Field(description="Next cursor, or null at the end.")
+
+
+class CursorPage[T](Collection[T]):
+    """Forward page with a continuation cursor or null at the end."""
+
+    meta: CursorMeta
+
+
+type CursorPaginateType[T] = Callable[[Select], Awaitable[CursorPage[T]]]
+type _CursorOrder = list[tuple[sa.Column, bool]]
+type _CursorValue = int | str | UUID | datetime | date | Decimal
+
+
+def _cursor_order(stmt: Select) -> _CursorOrder:
+    # SQLAlchemy statement introspection is isolated here for compatibility testing.
+    if not isinstance(stmt, Select) or any(
+        getattr(stmt, name) is not None
+        for name in ("_limit_clause", "_offset_clause", "_fetch_clause")
+    ):
+        raise ValueError("Cursor pagination requires an unlimited Select")
+    if stmt._distinct or stmt._group_by_clauses or stmt._having_criteria:
+        raise ValueError("Cursor pagination does not support DISTINCT or aggregation")
+    if any(
+        not isinstance(col.element if isinstance(col, Label) else col, sa.Column)
+        for col in stmt.selected_columns
+    ):
+        raise ValueError("Cursor pagination requires entity or column selections")
+    sources = stmt.get_final_froms()
+    if any(
+        isinstance(node, Join) and (node.isouter or node.full)
+        for source in sources
+        for node in visitors.iterate(source)
+    ):
+        raise ValueError("Cursor pagination does not support outer joins")
+    order = []
+    for expression in stmt._order_by_clauses:
+        descending = False
+        if isinstance(expression, UnaryExpression) and expression.modifier in (
+            operators.asc_op, operators.desc_op
+        ):
+            descending = expression.modifier is operators.desc_op
+            expression = expression.element
+        if (
+            not isinstance(expression, sa.Column)
+            or not isinstance(expression.table, sa.Table)
+            or expression.nullable
+            or not any(source.is_derived_from(expression.table) for source in sources)
+        ):
+            raise ValueError("Cursor ordering requires non-null columns from the query")
+        if not isinstance(
+            expression.type,
+            (sa.Integer, sa.String, sa.Uuid, sa.DateTime, sa.Date, sa.Numeric),
+        ) or expression.type.python_type not in (int, str, UUID, datetime, date, Decimal):
+            raise ValueError("Unsupported cursor column type")
+        order.append((expression, descending))
+    if not order:
+        raise ValueError("Cursor pagination requires an explicit unique ordering")
+    return order
+
+
+def _cursor_schema(order: _CursorOrder) -> list[str]:
+    return [
+        json.dumps([col.table.fullname, col.name, desc, col.type.python_type.__name__])
+        for col, desc in order
+    ]
+
+
+def _encode_cursor(order: _CursorOrder, values: tuple[_CursorValue, ...]) -> str:
+    key_types = tuple(column.type.python_type for column, _ in order)
+    adapter = TypeAdapter(tuple[key_types])
+    encoded = adapter.dump_python(adapter.validate_python(values, strict=True), mode="json")
+    payload = {"v": 1, "order": _cursor_schema(order), "values": encoded}
+    return base64.urlsafe_b64encode(json.dumps(payload).encode()).decode().rstrip("=")
+
+
+def _decode_cursor(cursor: str, order: _CursorOrder, dialect: str) -> list[_CursorValue]:
+    try:
+        if not re.fullmatch(r"[A-Za-z0-9_-]+", cursor):
+            raise ValueError("Invalid encoding")
+        padded = cursor + "=" * (-len(cursor) % 4)
+        payload = json.loads(base64.b64decode(padded, altchars=b"-_", validate=True))
+        if (
+            not isinstance(payload, dict)
+            or set(payload) != {"v", "order", "values"}
+            or type(payload["v"]) is not int
+            or payload["v"] != 1
+            or payload["order"] != _cursor_schema(order)
+            or not isinstance(payload["values"], list)
+            or len(payload["values"]) != len(order)
+        ):
+            raise ValueError("Invalid payload")
+        values = []
+        for (column, _), value in zip(order, payload["values"], strict=True):
+            value_type = column.type.python_type
+            if type(value) is not (int if value_type is int else str):
+                raise ValueError("Invalid key type")
+            if value_type is int:
+                bits = 16 if isinstance(column.type, sa.SmallInteger) else 32
+                if dialect == "sqlite" or isinstance(column.type, sa.BigInteger):
+                    bits = 64
+                if not -(2 ** (bits - 1)) <= value < 2 ** (bits - 1):
+                    raise ValueError("Integer key out of range")
+            parsed = TypeAdapter(value_type).validate_json(json.dumps(value), strict=True)
+            if value_type is Decimal and (
+                parsed.as_tuple().exponent < -16383
+                or parsed.adjusted()
+                >= (column.type.precision or 131072) - (column.type.scale or 0)
+            ):
+                raise ValueError("Decimal key out of range")
+            if dialect == "postgresql" and (
+                (value_type is str and "\0" in parsed)
+                or (
+                    value_type is datetime
+                    and (parsed.utcoffset() is not None) != column.type.timezone
+                )
+            ):
+                raise ValueError("Key cannot be represented by the database column")
+            values.append(parsed)
+        return values
+    except (ValueError, TypeError, RecursionError) as error:
+        raise HTTPException(status_code=422, detail="Invalid cursor") from error
+
+
+def new_cursor_pagination[T](
+    default_page_size: int = 10,
+    max_page_size: int = 100,
+    *,
+    row_mapper: Callable[[sa.Row], T] = lambda row: row[0],
+) -> Callable[..., CursorPaginateType[T]]:
+    """Create a forward cursor dependency with a one-row-to-one-item mapper.
+
+    Args:
+        default_page_size: Default limit when the client omits it.
+        max_page_size: Maximum accepted limit.
+        row_mapper: Maps each original result row to exactly one response item.
+
+    Raises:
+        ValueError: Page-size bounds or the supplied Select are unsupported.
+    """
+    if (
+        type(default_page_size) is not int
+        or type(max_page_size) is not int
+        or not 1 <= default_page_size <= max_page_size
+    ):
+        raise ValueError("Require 1 <= default_page_size <= max_page_size")
+
+    def dependency(
+        session: Session,
+        cursor: str | None = Query(None, min_length=1),
+        limit: int = Query(default_page_size, ge=1, le=max_page_size),
+    ) -> CursorPaginateType[T]:
+        async def paginate(stmt: Select) -> CursorPage[T]:
+            order = _cursor_order(stmt)
+            columns = [column for column, _ in order]
+            if cursor is not None:
+                dialect = session.get_bind(clause=stmt).dialect.name
+                values = tuple(_decode_cursor(cursor, order, dialect))
+                same_direction = len({desc for _, desc in order}) == 1
+                if same_direction and dialect in ("postgresql", "sqlite", "mysql"):
+                    keys = sa.tuple_(*columns)
+                    condition = keys < values if order[0][1] else keys > values
+                else:
+                    terms, prefix = [], []
+                    for (column, desc), value in zip(order, values, strict=True):
+                        comparison = column < value if desc else column > value
+                        terms.append(sa.and_(*prefix, comparison))
+                        prefix.append(column == value)
+                    bound = columns[0] <= values[0] if order[0][1] else columns[0] >= values[0]
+                    condition = sa.and_(bound, sa.or_(*terms))
+                stmt = stmt.where(condition)
+            stmt = stmt.add_columns(*(col.label(None) for col in columns)).limit(limit + 1)
+            result = await session.execute(stmt)
+            width = len(result.keys()) - len(columns)
+            frozen = result.freeze()
+            rows = frozen().all()
+            next_cursor = (
+                _encode_cursor(order, rows[limit - 1][-len(columns) :])
+                if len(rows) > limit
+                else None
+            )
+            original = frozen().columns(*range(width)).all()
+            data = [row_mapper(row) for row in original[:limit]]
+            return CursorPage(data=data, meta=CursorMeta(next_cursor=next_cursor))
+
+        return paginate
+
+    return dependency
+
+
+CursorPaginate = Annotated[CursorPaginateType[T], Depends(new_cursor_pagination())]
+"""Inject a forward paginator accepting cursor and limit query parameters."""