from __future__ import annotations
from dataclasses import dataclass
from datetime import date, datetime
from functools import lru_cache
from logging import getLogger
from typing import Any, Literal
from uuid import UUID
from edri.utility.function import camel2snake
from sqlalchemy import and_, cast, or_
from sqlalchemy.inspection import inspect
from sqlalchemy.orm import RelationshipProperty
from sqlalchemy.sql.elements import ColumnElement
from sqlalchemy.sql.sqltypes import ARRAY, Boolean, Date, DateTime, Enum, Integer, String
logger = getLogger(__name__)
UnknownFieldPolicy = Literal["ignore", "log", "raise", "collect"]
__all__ = ["FilterBuilder", "FilterBuilderConfig", "UnknownFieldPolicy"]
[docs]
@dataclass(frozen=True)
class FilterBuilderConfig:
"""Configuration for strict filter resolution."""
# What to do when a field/path/operator cannot be resolved.
unknown_field_policy: UnknownFieldPolicy = "log"
[docs]
class FilterBuilder:
"""Build SQLAlchemy filter expressions from client-style payloads.
Expected payload shape::
names: ["createdAt", "subject.idSubject", ...]
options: ["before", "is", ...]
values: ["2026-01-17T00:00:00.000Z", "177", ...]
Resolution is deliberately dumb and strict:
- ``name`` resolves only to a mapped scalar column/property on ``table``.
- ``subject.name`` resolves ``subject`` as a mapped relationship and
``name`` as a mapped scalar column/property on the related model.
- deeper paths work the same way, e.g.
``subject.organization.idOrganization``.
- every intermediate dotted segment must be a mapped relationship.
- the final segment must be a mapped scalar column/property.
- a relationship without a final dotted column is filterable only with
``isEmpty``/``isNotEmpty`` (including their null aliases).
- no FK/PK guessing is performed.
- no ``id_<field>`` fallback is performed.
- no association-object inference is performed.
- no relationship-PK inference is performed.
The frontend/API serializer is responsible for sending the exact backend
field path. For example, a DataGrid column may display ``subject.name``
while its API filter field is explicitly mapped to
``subject.id_subject``.
"""
NULL_OPS = frozenset({"isEmpty", "isNotEmpty"})
EQUALITY_OPS = frozenset({"=", "!="})
ORDER_OPS = frozenset({">", ">=", "<", "<="})
RANGE_OPS = frozenset({"between", "notBetween"})
MEMBERSHIP_OPS = frozenset({"isAnyOf", "isNoneOf"})
STRING_MATCH_OPS = frozenset(
{
"equalsIgnoreCase",
"contains",
"doesNotContain",
"startsWith",
"endsWith",
}
)
ARRAY_OPS = frozenset(
{
"=",
"!=",
"contains",
"has",
"equalsArray",
"notEqualsArray",
}
)
ENUM_OPS = EQUALITY_OPS | frozenset({"equalsIgnoreCase"})
# Negative operator -> positive twin. Dotted paths negate the positive
# EXISTS, so rows without a related row also match a negative filter.
NEGATED_OPS = {
"!=": "=",
"isEmpty": "isNotEmpty",
"isNoneOf": "isAnyOf",
"doesNotContain": "contains",
"notBetween": "between",
"notEqualsArray": "equalsArray",
}
OPTION_ALIASES = {
"equals": "=",
"is": "=",
"doesNotEqual": "!=",
"not": "!=",
"isNull": "isEmpty",
"isNotNull": "isNotEmpty",
"before": "<",
"after": ">",
"onOrBefore": "<=",
"beforeOrEqual": "<=",
"beforeOrEquals": "<=",
"onOrAfter": ">=",
"afterOrEqual": ">=",
"afterOrEquals": ">=",
"in": "isAnyOf",
"isIn": "isAnyOf",
"anyOf": "isAnyOf",
"notIn": "isNoneOf",
"isNotIn": "isNoneOf",
"noneOf": "isNoneOf",
"like": "contains",
"notContains": "doesNotContain",
"beginsWith": "startsWith",
"endsIn": "endsWith",
"ieq": "equalsIgnoreCase",
"equalsCI": "equalsIgnoreCase",
"inRange": "between",
"notInRange": "notBetween",
}
TRUE_VALUES = frozenset({"true", "1", "yes", "y", "on", "t"})
FALSE_VALUES = frozenset({"false", "0", "no", "n", "off", "f"})
def __init__(self, unknown_field_policy: UnknownFieldPolicy = "log"):
self.cfg = FilterBuilderConfig(unknown_field_policy=unknown_field_policy)
self.unknown_fields: list[dict[str, Any]] = []
# =========================================================
# PUBLIC
# =========================================================
[docs]
def build(
self,
table: Any,
names: list[str],
options: list[str],
values: list[str],
operator: Literal["and", "or"] | None = "and",
) -> ColumnElement[bool] | None:
"""Build one SQLAlchemy boolean expression from parallel filter arrays.
``names`` are interpreted exactly as model paths after the existing
camelCase -> snake_case normalization of each segment.
"""
if not names:
return None
parts: list[ColumnElement[bool]] = []
for raw_name, option, raw_value in zip(names, options, values):
raw = (raw_name or "").strip()
if not raw:
continue
path = [segment.strip() for segment in raw.split(".")]
if any(not segment for segment in path):
self._handle_unknown_field(
model=table,
name=raw,
option=option,
value=raw_value,
reason="invalid dotted field path",
)
continue
if len(path) == 1:
segment = camel2snake(path[0])
kind, attr, _relprop = self._resolve(table, segment)
if kind == "none":
self._handle_unknown_field(
model=table,
name=raw,
option=option,
value=raw_value,
reason="mapped attribute not found on model",
)
continue
if kind == "relationship":
normalized_option = self._normalize_option(option)
if normalized_option in self.NULL_OPS:
exists = attr.any() if _relprop.uselist else attr.has()
parts.append(
~exists if normalized_option == "isEmpty" else exists
)
continue
self._handle_unknown_field(
model=table,
name=raw,
option=option,
value=raw_value,
reason="relationship filters require dotted field notation",
)
continue
expr = self._scalar_filter(attr, option, raw_value)
if expr is None:
self._handle_unknown_field(
model=table,
name=raw,
option=option,
value=raw_value,
reason="unsupported operator or invalid value for scalar field",
)
continue
parts.append(expr)
continue
expr = self._relationship_path_filter(
model=table,
path=path,
option=option,
value=raw_value,
)
if expr is None:
self._handle_unknown_field(
model=table,
name=raw,
option=option,
value=raw_value,
reason=(
"unknown or unsupported field path; every intermediate "
"segment must be a relationship and the final segment "
"must be a scalar field"
),
)
continue
parts.append(expr)
if not parts:
return None
return and_(*parts) if (operator or "").lower() == "and" else or_(*parts)
# =========================================================
# STRICT MODEL RESOLUTION
# =========================================================
@classmethod
@lru_cache(maxsize=512)
def _mapper(cls, model_cls: Any):
"""Cached SQLAlchemy inspection for mapped classes."""
try:
return inspect(model_cls)
except Exception:
return None
@classmethod
@lru_cache(maxsize=8192)
def _resolve_meta(cls, model_cls: Any, name: str) -> tuple[str, str | None]:
"""Resolve only real mapped columns/properties or relationships.
Unlike the old implementation, this intentionally does not use
``hasattr()`` as field discovery and performs no naming fallbacks.
"""
mapper = cls._mapper(model_cls)
if mapper is None:
return "none", None
try:
relationship = mapper.relationships.get(name)
except Exception:
relationship = None
if isinstance(relationship, RelationshipProperty):
return "relationship", name
try:
column_property = mapper.column_attrs.get(name)
except Exception:
column_property = None
if column_property is not None:
return "column", name
return "none", None
@classmethod
def _resolve(cls, model_cls: Any, name: str) -> tuple[str, Any, RelationshipProperty | None]:
"""Resolve a mapped attribute to ``(kind, attr, relationship_property)``."""
kind, attr_name = cls._resolve_meta(model_cls, name)
if kind == "none" or not attr_name:
return "none", None, None
attr = getattr(model_cls, attr_name, None)
if attr is None:
return "none", None, None
if kind == "relationship":
mapper = cls._mapper(model_cls)
if mapper is None:
return "none", None, None
try:
relprop = mapper.relationships.get(attr_name)
except Exception:
relprop = None
if not isinstance(relprop, RelationshipProperty):
return "none", None, None
return "relationship", attr, relprop
return "column", attr, None
# =========================================================
# SCALAR FILTERS
# =========================================================
@classmethod
def _scalar_filter(cls, col: Any, option: str, raw_value: str) -> ColumnElement[bool] | None:
"""Compare one mapped scalar field to one payload value."""
option = cls._normalize_option(option)
col_type = getattr(col, "type", None)
if not cls._supports_scalar_option_type(col_type, option):
return None
if option == "isEmpty":
return col.is_(None)
if option == "isNotEmpty":
return col.is_not(None)
# PostgreSQL ARRAY handling.
#
# Project semantics:
# is / equals -> array contains all requested values (@>)
# not -> array does not contain all requested values
# isAnyOf -> arrays overlap (&&)
# isNoneOf -> arrays do not overlap
# contains / has -> array contains all requested values (@>)
# equalsArray -> exact equality of the complete array
# notEqualsArray -> exact inequality of the complete array
if isinstance(col_type, ARRAY):
item_type = getattr(col_type, "item_type", None)
ok, items = cls._coerce_csv_for_type(item_type, raw_value)
if not ok or not items:
return None
if option in {"=", "!=", "contains", "has"}:
expr = col.bool_op("@>")(items)
return expr if option in {"=", "contains", "has"} else or_(~expr, col.is_(None))
if option in cls.MEMBERSHIP_OPS:
expr = col.bool_op("&&")(items)
return expr if option == "isAnyOf" else or_(~expr, col.is_(None))
if option == "equalsArray":
return col == items
if option == "notEqualsArray":
return or_(col != items, col.is_(None))
return None
if option in cls.MEMBERSHIP_OPS:
ok, items = cls._coerce_csv_for_type(col_type, raw_value)
if not ok or not items:
return None
expr = col.in_(items)
return expr if option == "isAnyOf" else or_(~expr, col.is_(None))
if isinstance(col_type, Boolean):
ok, value = cls._coerce_value_for_type(col_type, raw_value)
if not ok:
return None
return col.is_(value) if option == "=" else col.is_not(value)
if isinstance(col_type, (Integer, Date, DateTime)):
if option in cls.RANGE_OPS:
raw_range = cls._parse_range(raw_value)
if raw_range is None:
return None
ok, values = cls._coerce_items_for_type(col_type, raw_range)
if not ok:
return None
expr = col.between(values[0], values[1])
return or_(~expr, col.is_(None)) if option == "notBetween" else expr
ok, value = cls._coerce_value_for_type(col_type, raw_value)
if not ok:
return None
return cls._comparison_expr(col, option, value, include_null_on_negative=True)
if isinstance(col_type, Enum):
if option == "equalsIgnoreCase":
return cast(col, String).ilike(str(raw_value))
ok, value = cls._coerce_value_for_type(col_type, raw_value)
if not ok:
return None
return cls._comparison_expr(col, option, value, include_null_on_negative=True)
if option == "=":
return col == raw_value
if option == "!=":
return or_(col != raw_value, col.is_(None))
string_col = cast(col, String)
if option == "equalsIgnoreCase":
return string_col.ilike(str(raw_value))
if option == "contains":
return string_col.ilike(f"%{raw_value}%")
if option == "doesNotContain":
return or_(~string_col.ilike(f"%{raw_value}%"), col.is_(None))
if option == "startsWith":
return string_col.ilike(f"{raw_value}%")
if option == "endsWith":
return string_col.ilike(f"%{raw_value}")
return None
# =========================================================
# DOTTED RELATIONSHIP PATHS
# =========================================================
@classmethod
def _relationship_path_filter(
cls,
model: Any,
path: list[str],
option: str,
value: str,
) -> ColumnElement[bool] | None:
"""Resolve an explicit dotted path and build correlated EXISTS clauses.
Example::
subject.organization.name contains "abc"
becomes conceptually::
Root.subject.has(
Subject.organization.has(
Organization.name.ilike("%abc%")
)
)
Collection relationships use ``any()`` instead of ``has()``.
A negative operator wraps the positive filter in ``NOT EXISTS``::
subject.name != "abc" -> ~Root.subject.has(Subject.name == "abc")
so it also matches rows without a related row, and for collections it
means "no related row matches".
"""
if len(path) < 2:
return None
option = cls._normalize_option(option)
positive = cls.NEGATED_OPS.get(option)
if positive is not None:
expr = cls._relationship_path_filter(model, path, positive, value)
return None if expr is None else ~expr
segment = camel2snake(path[0])
kind, attr, relprop = cls._resolve(model, segment)
if kind != "relationship" or relprop is None:
return None
target_cls = relprop.mapper.class_
remaining = path[1:]
if len(remaining) == 1:
final_segment = camel2snake(remaining[0])
final_kind, final_attr, _ = cls._resolve(target_cls, final_segment)
if final_kind != "column":
return None
inner = cls._scalar_filter(final_attr, option, value)
else:
inner = cls._relationship_path_filter(
model=target_cls,
path=remaining,
option=option,
value=value,
)
if inner is None:
return None
return attr.any(inner) if relprop.uselist else attr.has(inner)
# =========================================================
# UNKNOWN / UNSUPPORTED FILTER HANDLING
# =========================================================
def _handle_unknown_field(
self,
*,
model: Any,
name: str,
option: str,
value: Any,
reason: str,
) -> None:
info = {
"model": getattr(model, "__name__", str(model)),
"name": name,
"option": option,
"value": value,
"reason": reason,
}
policy = self.cfg.unknown_field_policy
if policy == "ignore":
return None
if policy == "collect":
self.unknown_fields.append(info)
return None
msg = (
f"FilterBuilder: unknown/unsupported filter '{name}' on {info['model']} "
f"(op={option!r}, value={value!r}) reason={reason}"
)
if policy == "raise":
raise ValueError(msg)
logger.warning(msg)
return None
# =========================================================
# TYPE / VALUE HELPERS
# =========================================================
@classmethod
def _supports_scalar_option_type(cls, col_type: Any, option: str) -> bool:
if option in cls.NULL_OPS:
return True
if isinstance(col_type, ARRAY):
return option in cls.ARRAY_OPS | cls.MEMBERSHIP_OPS
if option in cls.MEMBERSHIP_OPS:
return not isinstance(col_type, Boolean)
if isinstance(col_type, Integer):
return option in cls.EQUALITY_OPS | cls.ORDER_OPS | cls.RANGE_OPS
if isinstance(col_type, Boolean):
return option in cls.EQUALITY_OPS
if isinstance(col_type, (Date, DateTime)):
return option in cls.EQUALITY_OPS | cls.ORDER_OPS | cls.RANGE_OPS
if isinstance(col_type, Enum):
return option in cls.ENUM_OPS
return option in cls.EQUALITY_OPS | cls.STRING_MATCH_OPS
@staticmethod
def _parse_iso_datetime(raw: str) -> datetime:
value = (raw or "").strip()
if value.endswith("Z"):
value = value[:-1] + "+00:00"
return datetime.fromisoformat(value)
@staticmethod
def _parse_range(raw: str) -> tuple[str, str] | None:
value = (raw or "").strip()
if ".." in value:
left, right = value.split("..", 1)
left, right = left.strip(), right.strip()
return (left, right) if left and right else None
if "," in value:
parts = [part.strip() for part in value.split(",") if part.strip()]
return (parts[0], parts[1]) if len(parts) >= 2 else None
if " - " in value:
left, right = value.split(" - ", 1)
left, right = left.strip(), right.strip()
return (left, right) if left and right else None
return None
@classmethod
@lru_cache(maxsize=256)
def _enum_maps(
cls,
enum_cls: Any,
) -> tuple[tuple[Any, ...], dict[str, Any], dict[str, Any], dict[int, Any]]:
"""Precompute Python Enum/StrEnum lookup tables."""
try:
members = tuple(enum_cls)
by_name: dict[str, Any] = dict(getattr(enum_cls, "__members__", {}))
except Exception:
return (), {}, {}, {}
by_value_str: dict[str, Any] = {}
by_value_int: dict[int, Any] = {}
for member in members:
value = getattr(member, "value", None)
if value is not None:
by_value_str[str(value)] = member
if isinstance(value, int):
by_value_int[value] = member
return members, by_name, by_value_str, by_value_int
@classmethod
def _coerce_value_for_type(cls, attr_type: Any, raw: str) -> tuple[bool, Any]:
value = (raw or "").strip()
if not value:
return False, None
if isinstance(attr_type, Enum):
enum_cls = getattr(attr_type, "enum_class", None)
if enum_cls is None:
return True, value
members, by_name, by_value_str, by_value_int = cls._enum_maps(enum_cls)
# Exact Python Enum/StrEnum representation wins over numeric-index
# compatibility, especially for StrEnum values such as "1".
member = by_name.get(value)
if member is not None:
return True, member
member = by_value_str.get(value)
if member is not None:
return True, member
if value.isdigit():
index = int(value)
member = by_value_int.get(index)
if member is not None:
return True, member
if 1 <= index <= len(members):
return True, members[index - 1]
if index < len(members):
return True, members[index]
return False, None
try:
python_type = attr_type.python_type
except Exception:
python_type = None
try:
if python_type is bool:
lowered = value.lower()
if lowered in cls.TRUE_VALUES:
return True, True
if lowered in cls.FALSE_VALUES:
return True, False
return False, None
if python_type is int:
return True, int(value)
if python_type is float:
return True, float(value)
if python_type is UUID:
return True, UUID(value)
if python_type is datetime:
return True, cls._parse_iso_datetime(value)
if python_type is date:
parsed = (
cls._parse_iso_datetime(value).date()
if "T" in value or value.endswith("Z")
else date.fromisoformat(value)
)
return True, parsed
if python_type is str or python_type is None:
return True, value
return True, python_type(value)
except Exception:
return False, None
@classmethod
def _coerce_items_for_type(
cls,
attr_type: Any,
raw_items: tuple[str, ...] | list[str],
) -> tuple[bool, list[Any]]:
if attr_type is None:
return True, list(raw_items)
values: list[Any] = []
for raw_item in raw_items:
ok, value = cls._coerce_value_for_type(attr_type, raw_item)
if not ok:
return False, []
values.append(value)
return True, values
@classmethod
def _coerce_csv_for_type(cls, attr_type: Any, raw: str) -> tuple[bool, list[Any]]:
raw_items = cls._split_csv(raw)
if not raw_items:
return False, []
return cls._coerce_items_for_type(attr_type, raw_items)
@staticmethod
def _comparison_expr(
col: Any,
option: str,
value: Any,
*,
include_null_on_negative: bool = False,
) -> ColumnElement[bool] | None:
if option == "=":
return col == value
if option == "!=":
expr = col != value
return or_(expr, col.is_(None)) if include_null_on_negative else expr
if option == ">":
return col > value
if option == ">=":
return col >= value
if option == "<":
return col < value
if option == "<=":
return col <= value
return None
@staticmethod
def _split_csv(raw: str) -> list[str]:
if raw is None:
return []
return [item for part in str(raw).split(",") if (item := part.strip())]
@classmethod
def _normalize_option(cls, option: str) -> str:
normalized = (option or "").strip()
return cls.OPTION_ALIASES.get(normalized, normalized)