diff --git a/README.md b/README.md index 884836d..7d2a6d5 100644 --- a/README.md +++ b/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 ... diff --git a/docs/pagination.md b/docs/pagination.md index a0da588..c096e3d 100644 --- a/docs/pagination.md +++ b/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 diff --git a/plugin/fastsqla/skills/fastsqla-pagination/SKILL.md b/plugin/fastsqla/skills/fastsqla-pagination/SKILL.md index ad90ed0..d24a8fb 100644 --- a/plugin/fastsqla/skills/fastsqla-pagination/SKILL.md +++ b/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 diff --git a/src/fastsqla.py b/src/fastsqla.py index d9fa3ab..5b4c562 100644 --- a/src/fastsqla.py +++ b/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,6 +42,10 @@ __all__ = [ "Base", "Collection", + "CursorMeta", + "CursorPage", + "CursorPaginate", + "CursorPaginateType", "Item", "MissingConfigurationError", "Page", @@ -39,6 +53,7 @@ "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.""" diff --git a/tests/integration/test_cursor_pagination.py b/tests/integration/test_cursor_pagination.py new file mode 100644 index 0000000..a1a8688 --- /dev/null +++ b/tests/integration/test_cursor_pagination.py @@ -0,0 +1,245 @@ +import base64 +import json +from datetime import UTC, date, datetime +from decimal import Decimal +from typing import Any +from uuid import UUID + +from fastapi import FastAPI, HTTPException +from httpx import AsyncClient +from pytest import fixture, mark, param, raises +from sqlalchemy import Numeric, asc, desc, event, func, select +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + + +@fixture +async def item(engine: AsyncEngine, session: AsyncSession) -> type[Any]: + class Base(DeclarativeBase): + pass + + class Item(Base): + __tablename__ = "cursor_item" + cohort: Mapped[int] = mapped_column(primary_key=True) + id: Mapped[int] = mapped_column(primary_key=True) + name: Mapped[str] + optional: Mapped[str | None] + flag: Mapped[bool] = mapped_column(default=False) + day: Mapped[date] + timestamp: Mapped[datetime] + token: Mapped[UUID] + amount: Mapped[Decimal] = mapped_column(Numeric(10, 2)) + + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + session.add_all( + Item( + cohort=cohort, id=id_, name=f"{cohort}{id_}", day=date(2026, 1, n), + timestamp=datetime(2026, 1, n, tzinfo=UTC), token=UUID(int=n), + amount=Decimal(n) / 10 + ) + for n, (cohort, id_) in enumerate([(1, 1), (1, 2), (2, 1), (2, 2), (3, 1)], 1) + ) + await session.commit() + return Item + + +@fixture +def statements(engine: AsyncEngine) -> list[str]: + statements: list[str] = [] + + def capture(_conn: Any, _cursor: Any, statement: str, *_args: Any): + statements.append(statement) + + event.listen(engine.sync_engine, "before_cursor_execute", capture) + return statements + + +async def page( + session: AsyncSession, stmt: Any, limit: int = 2, + cursor: str | None = None, mapper: Any = lambda row: row[0], +) -> Any: + from fastsqla import new_cursor_pagination + + dependency = new_cursor_pagination(row_mapper=mapper) + paginate = dependency(session=session, cursor=cursor, limit=limit) + return await paginate(stmt) + + +@mark.parametrize( + ("keys", "expected"), + [ + param("cohort id", "11 12 21 22 31", id="ascending"), + param("-cohort -id", "31 22 21 12 11", id="descending"), + param("cohort -id", "12 11 22 21 31", id="mixed"), + param("-cohort id", "31 21 22 11 12", id="reverse-mixed"), + *(param(f"{key} cohort id", "11 12 21 22 31", id=key) + for key in ("day", "timestamp", "token", "amount")), + ] +) +async def test_traverses_ties_and_typed_keys_with_changing_limit( + item: type[Any], session: AsyncSession, keys: str, expected: str +): + ordering = [ + (desc if key.startswith("-") else asc)(getattr(item, key.lstrip("-"))) + for key in keys.split() + ] + stmt = select(item).order_by(*ordering) + first = await page(session, stmt) + second = await page(session, stmt, limit=1, cursor=first.meta.next_cursor) + third = await page(session, stmt, cursor=second.meta.next_cursor) + assert [row.name for row in first.data + second.data + third.data] == expected.split() + assert first.meta.next_cursor is not None + assert second.meta.next_cursor is not None + assert third.meta.model_dump() == {"next_cursor": None} + + +@mark.parametrize( + ("cohort", "expected"), [(99, []), (1, ["11", "12"])], ids=["empty", "full-page"] +) +async def test_terminal_pages( + item: type[Any], session: AsyncSession, cohort: int, expected: list[str] +): + result = await page( + session, select(item).where(item.cohort == cohort).order_by(item.cohort, item.id) + ) + assert [row.name for row in result.data] == expected + assert result.meta.next_cursor is None + + +async def test_retains_filters_after_boundary_deletion_and_insertion_ahead( + item: type[Any], session: AsyncSession +): + stmt = select(item).where(item.cohort <= 2).order_by(item.cohort, item.id) + first = await page(session, stmt) + await session.delete(first.data[-1]) + session.add( + item( + cohort=0, id=0, name="00", day=date(2026, 1, 1), + timestamp=datetime(2026, 1, 1, tzinfo=UTC), token=UUID(int=0), amount=Decimal(0) + ) + ) + await session.commit() + second = await page(session, stmt, cursor=first.meta.next_cursor) + assert [row.name for row in second.data] == ["21", "22"] + assert second.meta.next_cursor is None + + +async def test_projection_hides_cursor_columns_and_runs_one_query( + item: type[Any], session: AsyncSession, statements: list[str] +): + stmt = select(item.name.label("label")).order_by(item.cohort, item.id) + result = await page(session, stmt, mapper=lambda row: dict(row._mapping)) + assert result.data == [{"label": "11"}, {"label": "12"}] + assert result.meta.next_cursor is not None + assert len(statements) == 1 + assert "count(" not in statements[0].lower() + + +@mark.parametrize( + "cursor", ["", "not-a-cursor", "e30", "a+b"], + ids=["empty", "malformed", "invalid-payload", "invalid-alphabet"] +) +async def test_rejects_bad_cursors_before_sql( + item: type[Any], session: AsyncSession, statements: list[str], cursor: str +): + with raises(HTTPException) as error: + await page(session, select(item).order_by(item.cohort, item.id), cursor=cursor) + assert error.value.status_code == 422 + assert statements == [] + + +@mark.parametrize( + ("key", "dialect", "changes"), + [ + *(("cohort", "sqlite", changes) for changes in [ + {"v": True}, {"v": 2}, {"order": []}, {"values": [True, 1]}, + {"values": ["1", 1]}, {"values": [2**100, 1]}, {"values": [1]}, {"extra": 1}, + ]), + ("cohort", "postgresql", {"values": [2**100, 1]}), + ("name", "postgresql", {"values": ["\0", 1]}), + ("timestamp", "postgresql", {"values": ["2026-01-01T00:00:00Z", 1]}), + ("amount", "postgresql", {"values": ["1e999999", 1]}), + ("amount", "postgresql", {"values": ["1e-20000", 1]}), + ] +) +async def test_rejects_invalid_payload( + item: type[Any], session: AsyncSession, key: str, dialect: str, changes: dict +): + from fastsqla import _cursor_order, _decode_cursor + + stmt = select(item).order_by(getattr(item, key), item.id) + first = await page(session, stmt) + token = first.meta.next_cursor + payload = json.loads(base64.urlsafe_b64decode(token + "=" * (-len(token) % 4))) + encoded = json.dumps(payload | changes).encode() + token = base64.urlsafe_b64encode(encoded).decode().rstrip("=") + with raises(HTTPException) as error: + _decode_cursor(token, _cursor_order(stmt), dialect) + assert error.value.status_code == 422 + + +@mark.parametrize( + "kind", + ["unordered", "nullable", "expression", "limit", "offset", "distinct", "grouped", + "fetch", "projection", "outer-join", "type"] +) +async def test_rejects_unsupported_queries_before_sql( + item: type[Any], session: AsyncSession, statements: list[str], kind: str +): + stmt = select(item).order_by(item.cohort, item.id) + statements_by_kind = { + "unordered": select(item), + "nullable": select(item).order_by(item.optional), + "expression": select(item).order_by(func.lower(item.name)), + "limit": stmt.limit(1), + "offset": stmt.offset(1), + "distinct": stmt.distinct(), + "grouped": stmt.group_by(item.cohort, item.id), + "fetch": stmt.fetch(1), + "projection": select(func.count()).order_by(item.cohort, item.id), + "outer-join": stmt.outerjoin(item.__table__.alias(), item.id == 0), + "type": select(item).order_by(item.flag), + } + with raises(ValueError): + await page(session, statements_by_kind[kind]) + assert statements == [] + + +@mark.parametrize( + ("padding_size", "minimum_cursor_length"), [(0, 1), (5000, 4097)], ids=["short", "long"] +) +async def test_http_continuation( + app: FastAPI, client: AsyncClient, item: type[Any], session: AsyncSession, + padding_size: int, minimum_cursor_length: int, +): + from fastsqla import CursorPage, CursorPaginate + + rows = (await session.scalars(select(item).order_by(item.cohort, item.id))).all() + expected = [row.name + "x" * padding_size for row in rows] + for row, name in zip(rows, expected, strict=True): + row.name = name + await session.commit() + + @app.get("/cursor") + async def endpoint(paginate: CursorPaginate[str]) -> CursorPage[str]: + return await paginate(select(item.name).order_by(item.name, item.cohort, item.id)) + + first = await client.get("/cursor", params={"limit": 2}) + assert first.status_code == 200 + assert first.json()["data"] == expected[:2] + cursor = first.json()["meta"]["next_cursor"] + assert len(cursor) >= minimum_cursor_length + second = await client.get("/cursor", params={"limit": 3, "cursor": cursor}) + assert second.status_code == 200 + assert second.json() == {"data": expected[2:], "meta": {"next_cursor": None}} + invalid = await client.get("/cursor", params={"cursor": "invalid"}) + assert invalid.status_code == 422 + + +@mark.parametrize("default,maximum", [(0, 10), (11, 10), (1, 0), (True, 10)]) +def test_invalid_factory_bounds(default: int, maximum: int): + from fastsqla import new_cursor_pagination + + with raises(ValueError): + new_cursor_pagination(default, maximum)