Filtering & Pagination¶

Pagination with SQL Objects¶

Use SQL.paginate() to add LIMIT/OFFSET to any query, and select_with_total to get both the page data and the total matching count.

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

db_path = tmp_path / "pagination.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

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)])

    # Approach 1: Use SQL.paginate() for simple offset pagination
    query = SQL("select id, name from items order by id").paginate(page=2, page_size=10)
    page_data = session.select(query)
    print(f"Page 2: {len(page_data)} items, first={page_data[0]['name']}")

    # Approach 2: Use select_with_total for pagination with total count
    paged_query = SQL("select id, name from items order by id").paginate(page=1, page_size=10)
    data, total = session.select_with_total(paged_query)
    print(f"Page 1: {len(data)} items, total: {total}")

    # Approach 3: Use the builder for dynamic pagination
    builder_query = (
        sql.select("id", "name").from_("items").where_like("name", "Item%").order_by("id").limit(10).offset(0)
    )
    result = session.execute(builder_query)
    print(f"Builder: {len(result.all())} items")

Core Filter Types¶

SQLSpec defines filter types in sqlspec.core that can be used independently or with framework integrations:

• LimitOffsetFilter(limit, offset) -- pagination

• OrderByFilter(field_name, sort_order) -- sorting (supports expression mode)

• SearchFilter(field_name, value, ignore_case) -- text search

• BeforeAfterFilter(field_name, before, after) -- date range

• InCollectionFilter(field_name, values) -- set membership

• NotInCollectionFilter(field_name, values) -- set exclusion

• NullFilter(field_name) -- IS NULL check

• NotNullFilter(field_name) -- IS NOT NULL check

# Disambiguate columns in a JOIN
query = sql.select("p.name", "c.name").from_("parent p").join("child c", "p.id = c.parent_id")
filter_obj = SearchFilter(field_name="p.name", value="alice")
# Results in: WHERE p.name LIKE :p_name_search

from sqlglot import exp

# Sort by COALESCE(lines, 0)
expr = exp.func("COALESCE", exp.column("lines"), exp.Literal.number(0))
filter_obj = OrderByFilter(field_name=expr, sort_order="desc")

Framework Filter Dependencies¶

When using the Litestar extension, create_filter_dependencies() auto-generates Litestar dependency providers from a declarative configuration. FastAPI provides the same filter contract through SQLSpecPlugin.provide_filters() for use with Depends(). These providers parse query parameters from incoming requests and produce filter objects.

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

# Define filter configuration for a "users" endpoint
user_filters: FilterConfig = {
    "id_filter": str,  # Filter by user IDs
    "id_field": "id",  # Column name for ID filter
    "sort_field": ["created_at", "uploaded_collections", "name"],  # Allowed sort columns
    # orderBy accepts camel aliases like uploadedCollections by default
    "sort_field_aliases": {"lastUpload": "uploaded_collections"},  # Optional non-mechanical aliases
    "sort_order": "desc",  # Default sort direction
    "pagination_type": "limit_offset",  # Enable pagination
    "pagination_size": 20,  # Default page size
    "search": "name,email",  # Searchable fields
    "search_ignore_case": True,  # Case-insensitive search
    "created_at": True,  # Enable created_at range filter
    "updated_at": True,  # Enable updated_at range filter
}

# Generate Litestar dependency providers
deps = create_filter_dependencies(user_filters)
print(f"Generated {len(deps)} filter dependencies")

Using filters in a Litestar handler:

from litestar import get
from sqlspec.core import FilterTypes
from sqlspec.extensions.litestar.providers import create_filter_dependencies

user_filter_deps = create_filter_dependencies({
    "pagination_type": "limit_offset",
    "pagination_size": 20,
    "sort_field": ["created_at", "uploaded_collections", "name"],
    "sort_order": "desc",
    "search": "name,email",
})

@get("/users", dependencies=user_filter_deps)
async def list_users(
    db_session: AsyncpgDriver,
    filters: list[FilterTypes],
) -> dict:
    query = sql.select("*").from_("users")
    data, total = await db_session.select_with_total(query, *filters)
    return {"data": data, "total": total}

The generated dependencies automatically handle query parameters for configured fields like ?currentPage=2&pageSize=10&searchString=alice&orderBy=uploadedCollections&sortOrder=asc. Camelized orderBy values are accepted by default for every configured sort_field value, so orderBy=uploadedCollections is normalized to the SQL-facing field uploaded_collections before the OrderByFilter is created. Raw configured values such as orderBy=uploaded_collections also remain accepted for compatibility.

Sort aliases are closed over the configured sort_field allowlist. Use sort_field_aliases when the public API name is not a mechanical camel-case conversion, or set sort_field_camelize=False to require raw configured orderBy values only:

user_filter_deps = create_filter_dependencies({
    "sort_field": ["created_at", "uploaded_collections"],
    "sort_field_aliases": {"lastUpload": "uploaded_collections"},
})

snake_case_only_filter_deps = create_filter_dependencies({
    "sort_field": ["created_at", "uploaded_collections"],
    "sort_field_camelize": False,
})

orderBy=lastUpload is accepted, but aliases that target fields outside sort_field are rejected when the provider is created. Unknown orderBy values still fail validation before reaching SQL construction.

For FastAPI, use the same configuration with Depends():

filters = Depends(
    db_ext.provide_filters({
        "sort_field": ["created_at", "uploaded_collections"],
    })
)

SQLSpec does not ship generated filter providers for Flask, Starlette, or Sanic; their integrations do not have a runtime orderBy alias surface.

Service Layer¶

For common database operations and pagination in application services, SQLSpec provides base classes SQLSpecAsyncService and SQLSpecSyncService in sqlspec.service.

from sqlspec.service import SQLSpecAsyncService

class UserService(SQLSpecAsyncService):
    async def list_users(self, filters: list[StatementFilter]) -> OffsetPagination[User]:
        query = sql.select("*").from_("users")
        return await self.paginate(query, *filters, schema_type=User)

async def some_handler(db_session: AsyncDriver, filters: list[StatementFilter]):
    service = UserService(db_session)
    page = await service.list_users(filters)
    return page  # Returns OffsetPagination container

Related Guides¶

• Drivers and Querying for select_with_total and query methods.

• Query Builder for building queries with .where() clauses.