Source code for sqlspec.core.filters

"""Filter system for SQL statement manipulation.

This module provides filters that can be applied to SQL statements to add
WHERE clauses, ORDER BY clauses, LIMIT/OFFSET, and other modifications.

Components:
    - StatementFilter: Abstract base class for all filters
    - BeforeAfterFilter: Date range filtering
    - InCollectionFilter: IN clause filtering
    - LimitOffsetFilter: Pagination support
    - OrderByFilter: Sorting support
    - SearchFilter: Text search filtering
    - Various collection and negation filters

Features:
    - Parameter conflict resolution
    - Type-safe filter application
    - Cacheable filter configurations
"""

from abc import abstractmethod
from collections import abc
from datetime import datetime
from typing import TYPE_CHECKING, Any, ClassVar, Generic, Literal, TypeAlias

from mypy_extensions import mypyc_attr
from sqlglot import exp
from typing_extensions import TypeVar

from sqlspec.core._pagination import OffsetPagination
from sqlspec.core.query_modifiers import parse_column_for_condition
from sqlspec.utils.type_guards import has_field_name
from sqlspec.utils.uuids import uuid4

if TYPE_CHECKING:
    from sqlglot.expressions import Condition

    from sqlspec.core.statement import SQL

__all__ = (
    "AnyCollectionFilter",
    "BeforeAfterFilter",
    "BooleanFilter",
    "ChoicesFilter",
    "FilterTypeT",
    "FilterTypes",
    "InAnyFilter",
    "InCollectionFilter",
    "LimitOffsetFilter",
    "NotAnyCollectionFilter",
    "NotInCollectionFilter",
    "NotInSearchFilter",
    "NotNullFilter",
    "NullFilter",
    "OffsetPagination",
    "OnBeforeAfterFilter",
    "OrderByFilter",
    "PaginationFilter",
    "SearchFilter",
    "StatementFilter",
    "apply_filter",
    "canonicalize_filters",
    "find_filter",
)

T = TypeVar("T")
FilterTypeT = TypeVar("FilterTypeT", bound="StatementFilter")



[docs]
@mypyc_attr(allow_interpreted_subclasses=True)
class StatementFilter:
    """Abstract base class for filters that can be appended to a statement."""

    __slots__ = ()

    _is_statement_filter: bool = True


[docs]
    @abstractmethod
    def append_to_statement(self, statement: "SQL") -> "SQL":
        """Append the filter to the statement.

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

[docs]

def extract_parameters(self) -> "tuple[list[Any], dict[str, Any]]": """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: Tuple of (positional_parameters, named_parameters) where: - positional_parameters: List of positional parameter values - named_parameters: Dict of parameter name to value """ return [], {}