Filters

Composable SQL statement filters for building WHERE clauses, pagination, ordering, and search conditions. Filters can be applied to any SQL statement via apply_filter().

Filter Functions

sqlspec.core.filters.apply_filter(statement, filter_obj)

Apply a statement filter to a SQL query object.

Parameters:

• statement (SQL) -- The SQL query object to modify.

• filter_obj (StatementFilter) -- The filter to apply.

Return type:

SQL

Returns:

The modified query object.

sqlspec.core.filters.canonicalize_filters(filters)

Deduplicate and sort filters by type and field name for consistent hashing.

Parameters:

filters (Sequence[StatementFilter]) -- Sequence of StatementFilter objects

Return type:

tuple[StatementFilter, ...]

Returns:

Canonically sorted tuple of unique filters

sqlspec.core.filters.find_filter(filter_type, filters)

Get the filter specified by filter type from the filters.

Parameters:

• filter_type (type[TypeVar(FilterTypeT, bound= StatementFilter)]) -- The type of filter to find.

• filters (Sequence[StatementFilter | Any]) -- Filters to search through.

Return type:

Optional[TypeVar(FilterTypeT, bound= StatementFilter)]

Returns:

The match filter instance or None.

Base

class sqlspec.core.filters.StatementFilter

Bases: object

Abstract base class for filters that can be appended to a statement.

__reduce__()

Reconstruct via cls(*ctor_args) — works on mypyc native classes.

Return type:

tuple[typing.Any, ...]

__eq__(other)

Compare filters by their cache key for round-trip and deduplication checks.

Return type:

bool

__hash__()

Hash via the canonical filter key so unhashable members are normalized.

Return type:

int

abstractmethod append_to_statement(statement)

Append the filter to the statement.

This method modifies the SQL expression and adds parameters via add_named_parameter() on the returned statement.

Return type:

SQL

extract_parameters()

Return the parameters this filter would contribute.

This is an introspection helper and is not called during the normal filter-application path (append_to_statement handles both expression modification and parameter addition).

Returns:

• positional_parameters: List of positional parameter values

• named_parameters: Dict of parameter name to value

Return type:

Tuple of (positional_parameters, named_parameters) where

abstractmethod get_cache_key()

Return a cache key for this filter's configuration.

Return type:

tuple[typing.Any, ...]

Returns:

Tuple of hashable values representing the filter's configuration

class sqlspec.core.filters.PaginationFilter

Bases: StatementFilter

Base class for pagination-related filters.

abstractmethod append_to_statement(statement)

Append the filter to the statement.

This method modifies the SQL expression and adds parameters via add_named_parameter() on the returned statement.

Return type:

SQL

Value and Choice Filters

class sqlspec.core.filters.BooleanFilter

Bases: StatementFilter

Filter for boolean comparison queries.

Constructs WHERE field_name = :param clauses.

__init__(field_name, value)

get_param_name()

Get parameter name without storing it.

Return type:

str

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

append_to_statement(statement)

Apply boolean filter to SQL expression.

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

class sqlspec.core.filters.ChoicesFilter

Bases: InCollectionFilter[T]

Filter for field matching a defined set of choice values.

Constructs WHERE ... IN (...) clauses.

Date Filters

class sqlspec.core.filters.BeforeAfterFilter

Bases: _DatetimeBoundFilter

Filter for datetime range queries.

Applies WHERE clauses for before/after datetime filtering.

__init__(field_name, before=None, after=None)

__eq__(other)

Compare filters by their cache key for round-trip and deduplication checks.

Return type:

bool

__hash__()

Hash via the canonical filter key so unhashable members are normalized.

Return type:

int

__reduce__()

Reconstruct via cls(*ctor_args) — works on mypyc native classes.

Return type:

tuple[typing.Any, ...]

append_to_statement(statement)

Apply filter to SQL expression only.

Return type:

SQL

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

get_param_names()

Get parameter names without storing them.

Return type:

list[str]

class sqlspec.core.filters.OnBeforeAfterFilter

Bases: _DatetimeBoundFilter

Filter for inclusive datetime range queries.

Applies WHERE clauses for on-or-before/on-or-after datetime filtering.

__init__(field_name, on_or_before=None, on_or_after=None)

__eq__(other)

Compare filters by their cache key for round-trip and deduplication checks.

Return type:

bool

__hash__()

Hash via the canonical filter key so unhashable members are normalized.

Return type:

int

__reduce__()

Reconstruct via cls(*ctor_args) — works on mypyc native classes.

Return type:

tuple[typing.Any, ...]

append_to_statement(statement)

Apply filter to SQL expression only.

Return type:

SQL

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

get_param_names()

Get parameter names without storing them.

Return type:

list[str]

Collection Filters

class sqlspec.core.filters.InAnyFilter

Bases: StatementFilter, Generic[T]

Base class for collection-based filters that support ANY operations.

__init__(field_name, values=None)

get_param_names()

Get parameter names without storing them.

Return type:

list[str]

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

append_to_statement(statement)

Append the filter to the statement.

This method modifies the SQL expression and adds parameters via add_named_parameter() on the returned statement.

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

class sqlspec.core.filters.InCollectionFilter

Bases: InAnyFilter[T]

Filter for IN clause queries.

Constructs WHERE ... IN (...) clauses.

class sqlspec.core.filters.NotInCollectionFilter

Bases: InAnyFilter[T]

Filter for NOT IN clause queries.

Constructs WHERE ... NOT IN (...) clauses.

class sqlspec.core.filters.AnyCollectionFilter

Bases: InAnyFilter[T]

Filter for PostgreSQL-style ANY clause queries.

Constructs WHERE column_name = ANY (array_expression) clauses.

class sqlspec.core.filters.NotAnyCollectionFilter

Bases: InAnyFilter[T]

Filter for PostgreSQL-style NOT ANY clause queries.

Constructs WHERE NOT (column_name = ANY (array_expression)) clauses.

Null Filters

class sqlspec.core.filters.NullFilter

Bases: StatementFilter

Filter for IS NULL queries.

Constructs WHERE field_name IS NULL clauses.

__init__(field_name)

extract_parameters()

Extract filter parameters.

Returns empty parameters since IS NULL requires no values.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

append_to_statement(statement)

Apply IS NULL filter to SQL expression.

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

class sqlspec.core.filters.NotNullFilter

Bases: StatementFilter

Filter for IS NOT NULL queries.

Constructs WHERE field_name IS NOT NULL clauses.

__init__(field_name)

extract_parameters()

Extract filter parameters.

Returns empty parameters since IS NOT NULL requires no values.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

append_to_statement(statement)

Apply IS NOT NULL filter to SQL expression.

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

Pagination

class sqlspec.core.filters.LimitOffsetFilter

Bases: PaginationFilter

Filter for LIMIT and OFFSET clauses.

Adds pagination support through LIMIT/OFFSET SQL clauses.

__init__(limit, offset)

get_param_names()

Get parameter names without storing them.

Return type:

list[str]

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

append_to_statement(statement)

Append the filter to the statement.

This method modifies the SQL expression and adds parameters via add_named_parameter() on the returned statement.

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

class sqlspec.core.filters.OffsetPagination

Bases: Generic[T]

Container for data returned using limit/offset pagination.

Parameters:

• items (Sequence[TypeVar(T)]) -- List of data being sent as part of the response.

• limit (int) -- Maximal number of items to send.

• offset (int) -- Offset from the beginning of the query. Identical to an index.

• total (int) -- Total number of items.

__init__(items, limit, offset, total)

class sqlspec.core.filters.CursorFilter

Bases: PaginationFilter

Paginate by ordered keys whose last key uniquely identifies each row.

Replaces ORDER BY, LIMIT, and OFFSET and fetches limit + 1 rows. Assemble results with build_page() or the service's paginate(). Apply this filter last so other filters constrain the inner query when grouping or set operations require wrapping. paginate() enforces that order automatically.

Parameters:

• keys (str | Sequence[CursorKey | str | tuple[str, Literal['asc', 'desc']]]) -- A column name or sequence of names, (name, direction) tuples, or CursorKey objects, including a unique final tiebreaker. Bare names sort ascending.

• limit (int) -- Maximum page size.

• cursor (str | None) -- Optional token identifying the next or previous page.

• secret (str | bytes | None) -- Optional nonempty signing key.

Raises:

• ValueError -- Keys, limit, or secret are invalid.

• InvalidCursorError -- The token does not match the requested ordering.

__init__(keys, limit, cursor=None, secret=None)

append_to_statement(statement)

Apply cursor predicates, ordering and sentinel over-fetch.

Parameters:

statement (SQL) -- Base query with any earlier filters applied.

Return type:

SQL

Returns:

A query with cursor values bound as named parameters.

encode(row, *, backward)

Mint a token from a selected row.

Parameters:

• row (Mapping[str, Any]) -- Result row containing all declared key columns.

• backward (bool) -- Whether the token traverses backward.

Return type:

str

Returns:

A token bound to this filter's ordering and signing key.

Raises:

ImproperConfigurationError -- A key is missing or unexpectedly NULL.

build_page(rows)

Trim the sentinel and mint adjacent-page cursors.

Parameters:

rows (list[dict[str, Any]]) -- Raw result rows in effective query order.

Return type:

CursorPagination[dict[str, Any]]

Returns:

Items in declared order with page availability and cursor tokens.

get_cache_key()

Return SQL-affecting configuration without the signing secret.

Return type:

tuple[Any, ...]

extract_parameters()

Return the proposed names and values for cursor parameters.

Return type:

tuple[list[Any], dict[str, Any]]

class sqlspec.core.filters.CursorKey

Bases: object

Declare a sort key and its result-column name.

Parameters:

• field_name (str) -- Nonempty SQL column name, optionally qualified.

• sort_order (Literal['asc', 'desc']) -- Ascending or descending traversal.

• nulls (Optional[Literal['first', 'last']]) -- Explicit NULL placement, or None for a non-nullable key.

• result_name (str | None) -- Selected column name used to extract cursor values.

Raises:

ValueError -- A name, direction, or NULL placement is invalid.

__init__(field_name, sort_order='asc', nulls=None, result_name=None)

class sqlspec.core.CursorPagination

Bases: Generic[T]

Container for data returned using cursor (keyset) pagination.

Parameters:

• items (Sequence[TypeVar(T)]) -- List of data being sent as part of the response.

• limit (int) -- Maximal number of items to send.

• next_cursor (str | None) -- Token for the next page, or None on the last page.

• previous_cursor (str | None) -- Token for the previous page, or None on the first page.

• has_next (bool) -- Whether a page exists after this one.

• has_previous (bool) -- Whether a page exists before this one.

__init__(items, limit, next_cursor, previous_cursor, has_next, has_previous)

Cursor value types

sqlspec.core.register_cursor_type(value_type, tag, encoder, decoder)

Register a cursor value type at application initialization.

Parameters:

• value_type (type) -- Python type to encode.

• tag (str) -- Unique lowercase wire-format type tag.

• encoder (Callable[[Any], str]) -- Convert a value to text.

• decoder (Callable[[str], Any]) -- Restore a value from text.

Raises:

ValueError -- The type or tag is already registered or the tag is invalid.

Return type:

None

Ordering

class sqlspec.core.filters.OrderByFilter

Bases: StatementFilter

Filter for ORDER BY clauses.

Adds sorting capability to SQL queries.

Parameters:

• field_name (str | Expression) -- Column name or expression to order by.

• sort_order (Literal['asc', 'desc']) -- Ascending or descending direction.

• nulls (Optional[Literal['first', 'last']]) -- Explicit NULL placement, or None for the database default.

__init__(field_name, sort_order='asc', nulls=None)

property nulls: Literal['first', 'last'] | None

Return the requested NULL placement, or None for the database default.

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

append_to_statement(statement)

Append the filter to the statement.

This method modifies the SQL expression and adds parameters via add_named_parameter() on the returned statement.

Return type:

SQL

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

Search

class sqlspec.core.filters.SearchFilter

Bases: _TextSearchFilter

Filter for text search queries.

Constructs WHERE field_name LIKE '%value%' clauses.

__eq__(other)

Compare filters by their cache key for round-trip and deduplication checks.

Return type:

bool

__hash__()

Hash via the canonical filter key so unhashable members are normalized.

Return type:

int

__init__(field_name, value, ignore_case=False)

__reduce__()

Reconstruct via cls(*ctor_args) — works on mypyc native classes.

Return type:

tuple[typing.Any, ...]

append_to_statement(statement)

Append the filter to the statement.

This method modifies the SQL expression and adds parameters via add_named_parameter() on the returned statement.

Return type:

SQL

static escape_like_value(value)

Escape % and _ metacharacters for safe LIKE/ILIKE pattern matching.

Backslash is the default LIKE escape character in PostgreSQL (with standard_conforming_strings on), SQLite, and MySQL, so the produced pattern is safe without an explicit ESCAPE clause on those engines.

Parameters:

value (str) -- Raw user input to escape before wrapping in wildcards.

Return type:

str

Returns:

The input with \, %, and _ escaped using \.

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

get_param_name()

Get parameter name without storing it.

Return type:

str | None

property like_pattern: str | None

Return the search value wrapped in LIKE wildcards.

Returns:

The pattern for a LIKE operation, or None if no value.

class sqlspec.core.filters.NotInSearchFilter

Bases: SearchFilter

Filter for negated text search queries.

Constructs WHERE field_name NOT LIKE '%value%' clauses.

__eq__(other)

Compare filters by their cache key for round-trip and deduplication checks.

Return type:

bool

__hash__()

Hash via the canonical filter key so unhashable members are normalized.

Return type:

int

__init__(field_name, value, ignore_case=False)

__reduce__()

Reconstruct via cls(*ctor_args) — works on mypyc native classes.

Return type:

tuple[typing.Any, ...]

append_to_statement(statement)

Append the filter to the statement.

This method modifies the SQL expression and adds parameters via add_named_parameter() on the returned statement.

Return type:

SQL

static escape_like_value(value)

Escape % and _ metacharacters for safe LIKE/ILIKE pattern matching.

Backslash is the default LIKE escape character in PostgreSQL (with standard_conforming_strings on), SQLite, and MySQL, so the produced pattern is safe without an explicit ESCAPE clause on those engines.

Parameters:

value (str) -- Raw user input to escape before wrapping in wildcards.

Return type:

str

Returns:

The input with \, %, and _ escaped using \.

extract_parameters()

Extract filter parameters.

Return type:

tuple[list[typing.Any], dict[str, typing.Any]]

get_cache_key()

Return cache key for this filter configuration.

Return type:

tuple[typing.Any, ...]

get_param_name()

Get parameter name without storing it.

Return type:

str | None

property like_pattern: str | None

Return the search value wrapped in LIKE wildcards.

Returns:

The pattern for a LIKE operation, or None if no value.

Type Aliases

sqlspec.core.Pagination

Generic alias for CursorPagination[T] | OffsetPagination[T].

sqlspec.core.filters.FilterTypes

Union type of all filter classes.

sqlspec.core.filters.FilterTypeT

Type variable bound to StatementFilter.