Filtering and Pagination¶

List endpoints in APIs routinely combine pagination, search, date-range bounds, collection membership, and multi-column sorting. SQLSpec provides composable StatementFilter value objects in sqlspec.core along with offset and keyset cursor pagination helpers.

Built-In `StatementFilter` Classes¶

Pass any combination of StatementFilter instances as positional arguments to session.select(...), session.select_with_total(...), or service.paginate(...):

Filter Class	Appended SQL Semantics	Key Arguments
`LimitOffsetFilter`	`LIMIT :limit OFFSET :offset` (or `FETCH FIRST` on Oracle/Db2/MSSQL)	`limit`, `offset`
`OrderByFilter`	`ORDER BY <field> ASC\\|DESC [NULLS FIRST\\|LAST]`	`field_name`, `sort_order="asc"`, `nulls=None`
`SearchFilter`	Case-sensitive or case-insensitive `LIKE` / `ILIKE` across one or more columns	`field_name`, `value`, `ignore_case=False`
`BeforeAfterFilter`	`< :before` and/or `> :after` timestamp range bounds (`OnBeforeAfterFilter` for inclusive `<=`/`>=`)	`field_name`, `before=None`, `after=None`
`InCollectionFilter`	`WHERE <field> IN (:p1, ...)` (`NotInCollectionFilter` for `NOT IN`)	`field_name`, `values`
`NullFilter` / `NotNullFilter`	`WHERE <field> IS NULL` / `IS NOT NULL`	`field_name`

Because StatementFilter objects bind named parameters (:limit, :offset, etc.), always bind any additional base-query parameters as named keyword arguments (status="active"), never positional tuples.

Combining Multiple Filters on One Query¶

Filters compose in any order without mutating the underlying SQL statement:

from datetime import UTC, datetime

from sqlspec import SQL, SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.core import BeforeAfterFilter, InCollectionFilter, LimitOffsetFilter, OrderByFilter, SearchFilter

spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

with spec.provide_session(config) as session:
    session.execute("CREATE TABLE articles (id INTEGER PRIMARY KEY, title TEXT, status TEXT, published_at TEXT)")
    session.execute_many(
        "INSERT INTO articles (title, status, published_at) VALUES (:title, :status, :published_at)",
        [
            {"title": "SQLite WAL Tuning", "status": "published", "published_at": "2026-03-01T10:00:00Z"},
            {"title": "Postgres Poolers", "status": "published", "published_at": "2026-04-15T12:00:00Z"},
            {"title": "Draft Notes", "status": "draft", "published_at": "2026-04-20T09:00:00Z"},
        ],
    )

    base_query = SQL("SELECT id, title, status FROM articles")
    rows = session.select(
        base_query,
        InCollectionFilter("status", ["published"]),
        SearchFilter("title", "Pool", ignore_case=True),
        BeforeAfterFilter("published_at", after=datetime(2026, 1, 1, tzinfo=UTC).isoformat()),
        OrderByFilter("published_at", sort_order="desc"),
        LimitOffsetFilter(limit=10, offset=0),
    )
    assert [row["title"] for row in rows] == ["Postgres Poolers"]

Offset Pagination and Total Counts¶

For classic page-number navigation, chain .paginate(page=..., page_size=...) on a SQL statement and call session.select_with_total(...) (or service.paginate_limit_offset(...) to receive an OffsetPagination envelope):

docs/examples/patterns/pagination.py

"""Paginate queries with SQL.paginate(), select_with_total(), and fluent builder modifiers."""

from typing import Any

from sqlspec import SQLSpec, sql
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.core import SQL

__all__ = ("main",)


def main() -> tuple[list[dict[str, Any]], int]:
    """Run offset pagination via SQL.paginate, select_with_total, and the fluent builder."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

    with spec.provide_session(config) as session:
        session.execute("CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT)")
        session.execute_many("INSERT INTO items (name) VALUES (?)", [(f"Item {i}",) for i in range(1, 51)])

        page_two = session.select(SQL("SELECT id, name FROM items ORDER BY id").paginate(page=2, page_size=10))
        page_one, total = session.select_with_total(
            SQL("SELECT id, name FROM items ORDER BY id").paginate(page=1, page_size=10)
        )
        builder_rows = session.execute(
            sql.select("id", "name").from_("items").where_like("name", "Item%").order_by("id").limit(10).offset(0)
        ).all()

    assert len(page_two) == 10
    assert page_two[0]["name"] == "Item 11"
    assert len(page_one) == 10
    assert len(builder_rows) == 10
    assert total == 50
    return page_two, total


if __name__ == "__main__":
    rows, total_rows = main()
    print(f"Page 2 first={rows[0]['name']}, total={total_rows}")

When paginating complex queries containing JOIN, GROUP BY, or CTEs with named parameters, running an explicit scalar SELECT COUNT(*) query alongside session.select(query, LimitOffsetFilter(...), **named_params) avoids wrapping large projections inside a derived count subquery.

Keyset Cursor Pagination¶

Large tables degrade under deep OFFSET 100000 scans because the database must read and discard skipped rows. CursorFilter and SQLSpecSyncService.paginate_cursor / SQLSpecAsyncService.paginate_cursor implement indexed keyset seek pagination with opaque forward (next_cursor) and backward (previous_cursor) tokens:

docs/examples/patterns/cursor_pagination.py

"""Paginate ordered query results forward and backward using keyset cursors."""

from typing import Any

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.core import CursorFilter, CursorKeys, LimitOffsetFilter
from sqlspec.service import SQLSpecSyncService

__all__ = ("main",)


def main() -> tuple[int, list[dict[str, Any]]]:
    """Compare limit-offset pagination with bidirectional keyset cursor pagination."""
    spec = SQLSpec()
    config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))
    keys: CursorKeys = [("created_at", "desc"), ("id", "desc")]
    query = "SELECT id, name, created_at FROM items"

    try:
        with spec.provide_session(config) as session:
            session.execute("CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT, created_at TEXT)")
            session.execute_many(
                "INSERT INTO items (id, name, created_at) VALUES (?, ?, ?)",
                [(i, f"Item {i}", f"2026-01-{1 + i // 5:02d}") for i in range(1, 26)],
            )
            service = SQLSpecSyncService(session)
            offset_page = service.paginate_limit_offset(query, LimitOffsetFilter(limit=10, offset=0))
            first_page = service.paginate_cursor(query, CursorFilter(keys, limit=10))
            second_page = service.paginate_cursor(query, CursorFilter(keys, limit=10, cursor=first_page.next_cursor))
            previous_page = service.paginate(query, CursorFilter(keys, limit=10, cursor=second_page.previous_cursor))

        assert offset_page.total == 25
        assert len(offset_page.items) == 10
        assert len(first_page.items) == 10
        assert len(second_page.items) == 10
        assert {row["id"] for row in first_page.items}.isdisjoint(row["id"] for row in second_page.items)
        assert previous_page.items == first_page.items
        return offset_page.total, first_page.items
    finally:
        config.close_pool()


if __name__ == "__main__":
    total_count, first_items = main()
    print(f"Total: {total_count}, first page count: {len(first_items)}")

Always include a unique tie-breaker column (such as ("id", "desc")) as the final key in CursorKeys so rows with identical timestamps paginate deterministically without duplicates or gaps.

Declarative HTTP Filter Providers (Litestar)¶

In Litestar services, create_filter_dependencies generates typed query-parameter dependency providers (id_filter, created_at, updated_at, search, sort_field, and limit_offset) from a single FilterConfig dictionary:

docs/examples/patterns/filter_dependencies.py

"""Generate Litestar query-parameter filter dependencies from a declarative FilterConfig."""

from typing import Any

from sqlspec.extensions.litestar.providers import FilterConfig, create_filter_dependencies

__all__ = ("main",)


def main() -> dict[str, Any]:
    """Build filter, sort, search, and pagination dependency providers for a users endpoint."""
    user_filters: FilterConfig = {
        "id_filter": str,
        "id_field": "id",
        "sort_field": ["created_at", "uploaded_collections", "name"],
        "sort_field_aliases": {"lastUpload": "uploaded_collections"},
        "sort_order": "desc",
        "pagination_type": "limit_offset",
        "pagination_size": 20,
        "search": "name,email",
        "search_ignore_case": True,
        "created_at": True,
        "updated_at": True,
    }
    deps = create_filter_dependencies(user_filters)
    assert deps
    return deps


if __name__ == "__main__":
    print(f"Generated {len(main())} filter dependencies")

Always pass an explicit sort_field allowlist (and optional sort_field_aliases) so untrusted query strings can only sort by indexed, permitted columns.

Recap¶

Compose SearchFilter, BeforeAfterFilter, InCollectionFilter, OrderByFilter, and LimitOffsetFilter onto base SQL or builder queries instead of hand-rolling conditional SQL strings.
Keep all base-query parameters named (**params) whenever attaching StatementFilter instances.
Prefer CursorFilter keyset pagination with a unique tie-breaker key for infinite-scroll feeds and high-cardinality tables.