"""Filter-clause construction for typed Pydantic filter trees."""
from datetime import date, datetime
from enum import StrEnum
from typing import TYPE_CHECKING, Any, Literal
from sqlalchemy import and_, func, or_
from fsh_lib.relative_dates import resolve_relative_date
if TYPE_CHECKING:
from collections.abc import Callable, Sequence
from pydantic import BaseModel
from sqlalchemy import Select
from sqlalchemy.sql.elements import ColumnElement
[docs]
class FilterOperator(StrEnum):
"""The closed set of operators a filter condition may use.
Single source of truth shared by two layers: the
:data:`_FILTER_OPS` dispatch table here (execution -- each
member maps to a clause builder) and the field specs in
:mod:`fsh_lib.resource_registry` (declaration -- each field
advertises the subset it permits). Being a ``StrEnum`` lets
members double as dispatch keys and as plain wire strings: a
condition node's ``op`` arrives as a bare string yet still
indexes :data:`_FILTER_OPS`, since each member hashes and
compares equal to its value.
"""
EQ = "eq"
NEQ = "neq"
GT = "gt"
GTE = "gte"
LT = "lt"
LTE = "lte"
CONTAINS = "contains"
STARTS_WITH = "starts_with"
IN = "in"
IS_NULL = "is_null"
# Each entry takes ``(column, value)`` and returns the SQL clause.
# A callable map (rather than a method-name string) keeps unary-ish
# operators like ``is_null`` in the same dispatch table as the binary
# ones — value semantics differ per op but the call site doesn't.
_FILTER_OPS: dict[FilterOperator, Callable[[Any, Any], ColumnElement[bool]]] = {
FilterOperator.EQ: lambda col, v: col == v,
FilterOperator.NEQ: lambda col, v: col != v,
FilterOperator.GT: lambda col, v: col > v,
FilterOperator.GTE: lambda col, v: col >= v,
FilterOperator.LT: lambda col, v: col < v,
FilterOperator.LTE: lambda col, v: col <= v,
FilterOperator.CONTAINS: lambda col, v: col.contains(v),
FilterOperator.STARTS_WITH: lambda col, v: col.startswith(v),
FilterOperator.IN: lambda col, v: col.in_(v),
# Truthy ``value`` → ``IS NULL``; falsy → ``IS NOT NULL``.
# Matches how most filter UIs render an "is empty / is not
# empty" toggle.
FilterOperator.IS_NULL: (
lambda col, v: col.is_(None) if v else col.is_not(None)
),
}
_COMBINERS = {"and_": and_, "or_": or_}
# A pg_trgm trigram is 3 chars; shorter needles can't form one, so
# word_similarity / the gin_trgm_ops index can't match them.
_TRIGRAM_LEN = 3
[docs]
def trigram_clause(col: Any, needle: str) -> ColumnElement[bool]:
"""One column's fuzzy-match predicate for the ``trigram`` strategy.
pg_trgm word_similarity (``col %> needle`` == ``needle <% col``):
is *needle* similar to some continuous extent within the column?
The indexed column sits on the left so a ``gin_trgm_ops`` index
drives it. Needles shorter than a trigram can't form a
word_similarity match (and the index can't serve the pattern), so
they fall back to a case-insensitive substring.
Shared by :func:`apply_search` (the list search box) and the
resource_registry ``_values`` matcher (filter autocomplete) so the
two stay identical -- tune both at once with
``SET pg_trgm.word_similarity_threshold`` (default ``0.6``).
"""
if len(needle) < _TRIGRAM_LEN:
return col.icontains(needle, autoescape=True)
return col.bool_op("%>")(needle)
[docs]
def apply_filters(
stmt: Select,
node: BaseModel | None,
model: type,
*,
timezone: str = "UTC",
) -> Select:
"""Build WHERE clauses from a typed filter expression.
Accepts a typed Pydantic filter model -- either a single
condition (with ``field``, ``op``, ``value``) or a combiner
(with ``and_`` / ``or_`` lists of nested conditions). Models
that match none of these shapes are treated as a no-op and
the statement is returned unchanged. ``None`` is also a
no-op so call sites can invoke ``apply_filters(stmt,
body.filter, Model)`` unconditionally without an outer
``if body.filter is not None`` branch.
Args:
stmt: The SQLAlchemy SELECT statement to filter.
node: A Pydantic model representing the filter tree, or
``None`` to skip filtering entirely.
model: The SQLAlchemy model class providing columns.
timezone: Requesting user's ``Session.timezone``.
Defaults to ``"UTC"``.
Returns:
The statement with WHERE clauses applied.
"""
if node is None:
return stmt
clause = _build_filter_clause(node, model, timezone=timezone)
if clause is None:
return stmt
return stmt.where(clause)
def _build_filter_clause(
node: BaseModel,
model: type,
*,
timezone: str,
) -> ColumnElement[bool] | None:
"""Recursively build a SQLAlchemy clause from a filter node.
Dispatches on node shape via attribute presence:
- ``and_`` / ``or_`` attribute: combine child clauses.
- ``field`` / ``op`` / ``value`` attributes: leaf condition.
- Anything else: no-op, returns ``None``.
Args:
node: A Pydantic model representing a filter node.
model: The SQLAlchemy model class.
timezone: Timezone for time filters.
Returns:
A SQLAlchemy clause element, or ``None`` for empty or
shapeless nodes.
"""
for attr, combiner in _COMBINERS.items():
children = getattr(node, attr, None)
if children is not None:
return _combine(children, combiner, model, timezone=timezone)
field_name = getattr(node, "field", None)
if field_name is None:
return None
# Normalize the wire ``op`` (a bare string on the Pydantic node)
# into the enum once, so the dispatch lookup and comparisons below
# are over a single type.
op = FilterOperator(getattr(node, "op", FilterOperator.EQ))
value = getattr(node, "value", None)
col = getattr(model, field_name)
if isinstance(value, dict) and value.get("kind") == "relativeDate":
value = resolve_relative_date(
value,
timezone=timezone,
as_datetime=_column_python_type(col) is datetime,
)
if op != FilterOperator.IS_NULL and _is_bool_column(col):
if op == FilterOperator.IN and isinstance(value, (list, tuple)):
value = [_as_bool(item) for item in value]
else:
value = _as_bool(value)
elif op != FilterOperator.IS_NULL:
if op == FilterOperator.IN and isinstance(value, (list, tuple)):
value = [_coerce_to_column_type(col, item) for item in value]
else:
value = _coerce_to_column_type(col, value)
return _FILTER_OPS[op](col, value)
def _column_python_type(col: Any) -> type | None:
"""Return the Python type *col* maps to, or ``None`` if it has none."""
try:
return col.type.python_type
except AttributeError, NotImplementedError:
return None
def _coerce_to_column_type(col: Any, value: Any) -> Any:
"""Parse a string filter value into the column's Python type."""
if not isinstance(value, str):
return value
python_type = _column_python_type(col)
if python_type is datetime:
try:
return datetime.fromisoformat(value)
except ValueError:
return value
if python_type is date:
try:
return date.fromisoformat(value)
except ValueError:
return value
return value
def _is_bool_column(col: Any) -> bool:
"""Whether *col* maps to a Python ``bool``."""
return _column_python_type(col) is bool
def _as_bool(value: Any) -> Any:
"""Map a wire filter value bound for a boolean column to a real bool.
A condition's ``value`` arrives untyped, and an option-filter chip
sends strings -- including ``"true"`` / ``"false"`` for a boolean
field now that bool options come from ``_values``. A boolean
column compared against the string ``"false"`` would be wrong
(SQLAlchemy coerces it via ``bool("false")`` -> ``True``), so map
the bool strings to real booleans. A non-string value (an actual
JSON ``true`` / ``false``) is already correct and passes through.
"""
if isinstance(value, str):
return value.strip().lower() == "true"
return value
[docs]
def apply_search(
stmt: Select,
model: type,
columns: Sequence[str],
q: str | None,
strategy: Literal["trigram", "tsvector"] = "trigram",
) -> Select:
"""Filter *stmt* to rows where any of *columns* matches *q*.
Powers the list endpoint's free-text search box -- the typed
counterpart to :func:`apply_filters`, driven by the resource's
configured ``search`` fields rather than an explicit filter
tree. ``strategy`` is chosen per resource in the BE config
(``SearchConfig.strategy``):
* ``"trigram"`` -- each column is matched with ``pg_trgm``
word_similarity (``col %> q``, the commutator of ``q <% col``):
is *q* similar to some continuous extent within the column?
One index-backed operator that is both typo-tolerant and
substring-position-aware, so it catches ``"cat"`` inside
``"concatenate"`` where whole-string ``similarity`` would miss
it. Tune recall with ``SET pg_trgm.word_similarity_threshold``
(default ``0.6``). Needles shorter than a trigram can't form a
word_similarity match, so they fall back to a case-insensitive
substring (``ILIKE``). Requires the ``pg_trgm`` extension (a
``gin_trgm_ops`` index on each column to stay off a seq scan).
``columns`` are the resource's text columns.
* ``"tsvector"`` -- each column is a maintained ``tsvector``
column matched with ``@@`` against ``websearch_to_tsquery``.
Word/lexeme search with stemming -- right for prose columns.
``columns`` are ``tsvector`` column name(s); the ``english``
config here must match the one the column was built with.
``q`` of ``None`` or blank is a no-op, and so is an empty
``columns``, so call sites can invoke this unconditionally
without an outer guard.
Args:
stmt: The SQLAlchemy SELECT statement to filter.
model: The SQLAlchemy model class providing columns.
columns: Model attribute names to match ``q`` against --
text columns for ``trigram``, ``tsvector`` columns for
``tsvector``.
q: The free-text query, or ``None`` to skip search.
strategy: Matching strategy -- ``"trigram"`` (default) or
``"tsvector"``.
Returns:
The statement with the search WHERE clause applied.
"""
if q is None or not q.strip() or not columns:
return stmt
needle = q.strip()
clauses: list[ColumnElement[bool]] = []
for name in columns:
col = getattr(model, name)
if strategy == "tsvector":
clauses.append(
col.bool_op("@@")(
func.websearch_to_tsquery("english", needle),
),
)
else:
clauses.append(trigram_clause(col, needle))
return stmt.where(or_(*clauses))
def _combine(
children: Sequence[BaseModel],
combiner: Callable[..., ColumnElement[bool]],
model: type,
*,
timezone: str,
) -> ColumnElement[bool] | None:
"""Build and combine clauses for a combiner node's children.
Args:
children: The child filter nodes.
combiner: :func:`sqlalchemy.and_` or :func:`sqlalchemy.or_`.
model: The SQLAlchemy model class.
timezone: Forwarded to each child's :func:`_build_filter_clause`.
Returns:
The combined clause, or ``None`` if every child built
to ``None``.
"""
built = (
_build_filter_clause(child, model, timezone=timezone)
for child in children
)
clauses = [clause for clause in built if clause is not None]
return combiner(*clauses) if clauses else None