graflo.filter

Filter expression system for database queries.

This package provides a flexible system for creating and evaluating filter expressions that can be translated into different database query languages (AQL, Cypher, SQL, Python).

Key Components

• LogicalOperator: Logical operations (AND, OR, NOT, IMPLICATION / IF_THEN)
• ComparisonOperator: Comparison operations (==, !=, >, <, etc.)
• FilterExpression: Filter expression (leaf or composite logical formulae)
• parse_filter_expression: Unified YAML/JSON loader (Bindings, SelectSpec, graph DB)
• BoundParams: Values a rendered filter names by placeholder, for the driver to bind

SQL notes

• IF_THEN (IMPLICATION) renders as (NOT antecedent OR consequent).
• Use ExpressionFlavor.PYTHON for in-memory implication evaluation on documents.

Example

from graflo.filter import FilterExpression, parse_filter_expression expr = parse_filter_expression({ ... "AND": [ ... {"field": "age", "cmp_operator": ">=", "value": 18}, ... {"field": "status", "cmp_operator": "==", "value": "active"}, ... ] ... })

Modules:

Name	Description
onto	Filter expression system for database queries.
select	Declarative SQL-like view specifications.
sql	SQL filter helpers built on top of FilterExpression.

Attributes

ALL_BASE_COLUMNS = 'all_base' module-attribute

__all__ = ['ALL_BASE_COLUMNS', 'BoundParams', 'ComparisonOperator', 'FilterExpression', 'JoinClause', 'LogicalOperator', 'SelectSpec', 'parse_filter_expression', 'render_conjunct'] module-attribute

Classes

BoundParams

Values a rendered filter names by placeholder, for the driver to bind.

Pass one instance per query as params= to every filter rendered into it, and the query's :attr:values to the driver: placeholder names run on across filters. Only flavors whose drivers bind parameters take one; nGQL and GSQL filters are written as escaped literals.

Source code in graflo/filter/onto.py

class BoundParams:
    """Values a rendered filter names by placeholder, for the driver to bind.

    Pass one instance per query as ``params=`` to every filter rendered into
    it, and the query's :attr:`values` to the driver: placeholder names run on
    across filters. Only flavors whose drivers bind parameters take one; nGQL
    and GSQL filters are written as escaped literals.
    """

    _PLACEHOLDERS: Mapping[ExpressionFlavor, str] = MappingProxyType(
        {
            ExpressionFlavor.AQL: "@{}",
            ExpressionFlavor.CYPHER: "${}",
            ExpressionFlavor.SQL: "%({})s",
        }
    )

    def __init__(self, kind: ExpressionFlavor, prefix: str = "f") -> None:
        if kind not in self._PLACEHOLDERS:
            raise ValueError(
                f"{kind} filters take no bound parameters; they are written as "
                "escaped literals"
            )
        self.kind = kind
        self._prefix = prefix
        self.values: dict[str, Any] = {}

    def bind(self, value: Any) -> str:
        """Record *value* and return the placeholder that names it."""
        name = f"{self._prefix}{len(self.values)}"
        self.values[name] = value
        return self._PLACEHOLDERS[self.kind].format(name)

Attributes

kind = kind instance-attribute

values = {} instance-attribute

Methods:

__init__(kind, prefix='f')

Source code in graflo/filter/onto.py

def __init__(self, kind: ExpressionFlavor, prefix: str = "f") -> None:
    if kind not in self._PLACEHOLDERS:
        raise ValueError(
            f"{kind} filters take no bound parameters; they are written as "
            "escaped literals"
        )
    self.kind = kind
    self._prefix = prefix
    self.values: dict[str, Any] = {}

bind(value)

Record value and return the placeholder that names it.

Source code in graflo/filter/onto.py

def bind(self, value: Any) -> str:
    """Record *value* and return the placeholder that names it."""
    name = f"{self._prefix}{len(self.values)}"
    self.values[name] = value
    return self._PLACEHOLDERS[self.kind].format(name)

ComparisonOperator

Bases: BaseEnum

Comparison operators for field comparisons.

Attributes:

Name	Type	Description
NEQ		Not equal (!=)
EQ		Equal (==)
GE		Greater than or equal (>=)
LE		Less than or equal (<=)
GT		Greater than (>)
LT		Less than (<)
IN		Membership test (IN)
IS_NULL		Null check (IS NULL)
IS_NOT_NULL		Non-null check (IS NOT NULL)

Source code in graflo/filter/onto.py

class ComparisonOperator(BaseEnum):
    """Comparison operators for field comparisons.

    Attributes:
        NEQ: Not equal (!=)
        EQ: Equal (==)
        GE: Greater than or equal (>=)
        LE: Less than or equal (<=)
        GT: Greater than (>)
        LT: Less than (<)
        IN: Membership test (IN)
        IS_NULL: Null check (IS NULL)
        IS_NOT_NULL: Non-null check (IS NOT NULL)
    """

    NEQ = "!="
    EQ = "=="
    GE = ">="
    LE = "<="
    GT = ">"
    LT = "<"
    IN = "IN"
    IS_NULL = "IS_NULL"
    IS_NOT_NULL = "IS_NOT_NULL"

Attributes

EQ = '==' class-attribute instance-attribute

GE = '>=' class-attribute instance-attribute

GT = '>' class-attribute instance-attribute

IN = 'IN' class-attribute instance-attribute

IS_NOT_NULL = 'IS_NOT_NULL' class-attribute instance-attribute

IS_NULL = 'IS_NULL' class-attribute instance-attribute

LE = '<=' class-attribute instance-attribute

LT = '<' class-attribute instance-attribute

NEQ = '!=' class-attribute instance-attribute

FilterExpression

Bases: ConfigBaseModel

Unified filter expression (discriminated: leaf or composite).

• kind="leaf": single field comparison (field, cmp_operator, value, optional unary_op).
• kind="composite": logical combination (operator AND/OR/NOT/IF_THEN, deps).

Source code in graflo/filter/onto.py

class FilterExpression(ConfigBaseModel):
    """Unified filter expression (discriminated: leaf or composite).

    - kind="leaf": single field comparison (field, cmp_operator, value, optional unary_op).
    - kind="composite": logical combination (operator AND/OR/NOT/IF_THEN, deps).
    """

    kind: Literal["leaf", "composite"]

    # Leaf fields (used when kind="leaf")
    cmp_operator: ComparisonOperator | None = None
    value: list[Any] = Field(default_factory=list)
    field: str | None = None
    unary_op: str | None = (
        None  # optional operator before comparison (YAML key: "operator")
    )

    # Composite fields (used when kind="composite")
    operator: LogicalOperator | None = None  # AND, OR, NOT, IF_THEN
    deps: list[FilterExpression] = Field(default_factory=list)

    @field_validator("value", mode="before")
    @classmethod
    def value_to_list(cls, v: list[Any] | Any) -> list[Any]:
        """Convert single value to list if necessary. Explicit None becomes [None] for null comparison."""
        if v is None:
            return [None]
        if isinstance(v, list):
            return v
        return [v]

    @model_validator(mode="before")
    @classmethod
    def leaf_operator_to_unary_op(cls, data: Any) -> Any:
        """Map leaf 'operator' (YAML/kwargs) to unary_op; infer kind and cmp_operator."""
        if not isinstance(data, dict):
            return data
        data = dict(data)
        if data.get("kind") == "composite":
            return data

        if data.get("kind") is None:
            op = data.get("operator")
            if isinstance(op, LogicalOperator):
                data["kind"] = "composite"
                return data
            if isinstance(op, str) and op in _LOGICAL_OPERATOR_JSON_VALUES:
                data["kind"] = "composite"
                return data
            deps = data.get("deps")
            if isinstance(deps, list) and len(deps) > 0:
                data["kind"] = "composite"
                return data
            if data.get("cmp_operator") is not None or data.get("field") is not None:
                data["kind"] = "leaf"

        if "foo" in data:
            raise ValueError(_FOO_REMOVED)
        raw_op = None
        if "operator" in data and isinstance(data["operator"], str):
            raw_op = data.pop("operator")
        if raw_op is not None:
            data["unary_op"] = raw_op
            if data.get("cmp_operator") is None and raw_op in DUNDER_TO_CMP:
                data["cmp_operator"] = DUNDER_TO_CMP[raw_op]
            if data.get("kind") is None:
                data["kind"] = "leaf"
        return data

    @model_validator(mode="after")
    def check_discriminated_shape(self) -> FilterExpression:
        """Enforce exactly one shape per kind and normalise null-check operators."""
        if self.kind == "leaf":
            if self.operator is not None or self.deps:
                raise ValueError("leaf expression must not have operator or deps")
            if self.cmp_operator is None and self.unary_op is None:
                raise ValueError(
                    "leaf expression requires cmp_operator or operator"
                    + (f" (field {self.field!r})" if self.field else "")
                )
            # IS_NULL / IS_NOT_NULL are unary; clear any spurious value list
            if self.cmp_operator in (
                ComparisonOperator.IS_NULL,
                ComparisonOperator.IS_NOT_NULL,
            ):
                object.__setattr__(self, "value", [])
        else:
            if self.operator is None:
                raise ValueError("composite expression must have operator")
        return self

    @field_validator("deps", mode="before")
    @classmethod
    def parse_deps(cls, v: list[Any]) -> list[Any]:
        """Parse dict/list items into FilterExpression instances."""
        if not isinstance(v, list):
            return v
        result = []
        for item in v:
            if isinstance(item, (dict, list)):
                # Not `from_dict`: a serialized expression (`kind` / `operator`
                # + `deps`) has to come back as the expression it was.
                result.append(parse_filter_expression(item))
            else:
                result.append(item)
        return result

    @classmethod
    def from_list(cls, current: list[Any]) -> FilterExpression:
        """Build a leaf expression from list form [cmp_operator, value, field?, unary_op?]."""
        cmp_operator = current[0]
        value = current[1]
        field = current[2] if len(current) > 2 else None
        unary_op = current[3] if len(current) > 3 else None
        return cls(
            kind="leaf",
            cmp_operator=cmp_operator,
            value=value,
            field=field,
            unary_op=unary_op,
        )

    @classmethod
    def from_dict(cls, data: dict[str, Any] | list[Any]) -> Self:
        """Create a filter expression from a dictionary or list.

        Returns FilterExpression (leaf or composite). LSP-compliant: return type is Self.
        """
        if isinstance(data, list):
            if data[0] in ComparisonOperator:
                return cast(Self, cls.from_list(data))
            elif data[0] in LogicalOperator:
                return cls(kind="composite", operator=data[0], deps=data[1])
            raise ValueError(
                f"filter list must start with a comparison or logical operator, "
                f"got {data[0]!r}"
            )
        elif isinstance(data, dict):
            if data.get("kind") is not None:
                return cls.model_validate(data)
            k = next(iter(data.keys()))
            norm_k = k.upper() if isinstance(k, str) else k
            if norm_k in LogicalOperator:
                if len(data) > 1:
                    raise ValueError(
                        "a logical filter entry takes one logical operator key, "
                        f"got {sorted(str(key) for key in data)}"
                    )
                deps: list[FilterExpression] = [cls.from_dict(v) for v in data[k]]
                return cls(
                    kind="composite", operator=LogicalOperator(norm_k), deps=deps
                )
            else:
                # The model is built by hand below, so `extra="forbid"` never
                # sees these keys: an unknown one would be dropped silently.
                if "foo" in data:
                    raise ValueError(_FOO_REMOVED)
                unknown = sorted(str(key) for key in data if key not in _LEAF_KEYS)
                if unknown:
                    raise ValueError(
                        f"unknown filter key(s) {unknown}; a leaf takes "
                        f"{sorted(_LEAF_KEYS - {'kind', 'unary_op'})}, a "
                        f"logical entry one of {sorted(_LOGICAL_OPERATOR_JSON_VALUES)}"
                    )
                unary_op = data.get("operator")
                cmp_operator = data.get("cmp_operator")
                if cmp_operator is None and unary_op is not None:
                    cmp_operator = DUNDER_TO_CMP.get(unary_op)
                return cls(
                    kind="leaf",
                    cmp_operator=cmp_operator,
                    value=data.get("value", []),
                    field=data.get("field"),
                    unary_op=unary_op,
                )
        raise ValueError(f"expected dict or list, got {type(data)}")

    def rename_fields(self, renames: Mapping[str, str]) -> FilterExpression:
        """This expression with every leaf ``field`` rewritten by *renames*.

        Pure: returns a new expression and leaves this one untouched. Fields
        absent from *renames* keep their name.
        """
        if not renames:
            return self
        if self.kind == "leaf":
            if self.field is None or self.field not in renames:
                return self
            return self.model_copy(update={"field": renames[self.field]})
        return self.model_copy(
            update={"deps": [dep.rename_fields(renames) for dep in self.deps]}
        )

    def matches(self, doc: Mapping[str, Any]) -> bool:
        """Whether *doc* satisfies the expression, evaluated in Python.

        Prefer this to ``expr(kind=ExpressionFlavor.PYTHON, **doc)``, which
        cannot evaluate a document with a field named ``kind`` or ``doc_name``.
        """
        if self.kind == "leaf":
            return self._cast_python(doc)
        return self._cast_python_composite(doc)

    def __call__(
        self,
        doc_name="doc",
        kind: ExpressionFlavor = ExpressionFlavor.AQL,
        **kwargs,
    ) -> str | bool:
        """Render or evaluate the expression in the target language.

        With ``params=`` a :class:`BoundParams`, values are written as its
        placeholders; without it, as literals.
        """
        if self.kind == "leaf":
            return self._call_leaf(doc_name=doc_name, kind=kind, **kwargs)
        return self._call_composite(doc_name=doc_name, kind=kind, **kwargs)

    def _is_null_operator(self) -> bool:
        """Check if this is a null-checking operator (IS_NULL or IS_NOT_NULL)."""
        return self.cmp_operator in (
            ComparisonOperator.IS_NULL,
            ComparisonOperator.IS_NOT_NULL,
        )

    def _call_leaf(
        self,
        doc_name="doc",
        kind: ExpressionFlavor = ExpressionFlavor.AQL,
        **kwargs,
    ) -> str | bool:
        if not self._is_null_operator() and not self.value:
            logger.warning(f"for {self} value is not set : {self.value}")
        if self.cmp_operator is None and kind != ExpressionFlavor.PYTHON:
            raise ValueError(
                "leaf expression requires cmp_operator for non-PYTHON flavor"
            )
        params = _params_for(kind, kwargs)
        if kind == ExpressionFlavor.AQL:
            return self._cast_arango(doc_name, params)
        elif kind == ExpressionFlavor.CYPHER:
            return self._cast_cypher(doc_name, params)
        elif kind == ExpressionFlavor.NGQL:
            return self._cast_ngql(doc_name)
        elif kind == ExpressionFlavor.GSQL:
            if doc_name == "":
                field_types = kwargs.get("field_types")
                return self._cast_restpp(field_types=field_types)
            return self._cast_tigergraph(doc_name)
        elif kind == ExpressionFlavor.SQL:
            return self._cast_sql(params)
        elif kind == ExpressionFlavor.PYTHON:
            return self._cast_python(kwargs)
        raise ValueError(f"kind {kind} not implemented")

    def _call_composite(
        self,
        doc_name="doc",
        kind: ExpressionFlavor = ExpressionFlavor.AQL,
        **kwargs,
    ) -> str | bool:
        if kind in (
            ExpressionFlavor.AQL,
            ExpressionFlavor.CYPHER,
            ExpressionFlavor.NGQL,
            ExpressionFlavor.GSQL,
            ExpressionFlavor.SQL,
        ):
            return self._cast_generic(
                doc_name=doc_name, kind=kind, params=_params_for(kind, kwargs)
            )
        elif kind == ExpressionFlavor.PYTHON:
            return self._cast_python_composite(kwargs)
        raise ValueError(f"kind {kind} not implemented")

    def _in_members(self) -> list[Any]:
        """The members an ``IN`` tests against.

        ``value: [[a, b]]`` names the same set as ``value: [a, b]``.
        """
        if len(self.value) == 1 and isinstance(self.value[0], list):
            return list(self.value[0])
        return list(self.value)

    @staticmethod
    def _graph_literal(value: Any) -> str:
        """One value as a literal of the graph query languages (AQL, Cypher, nGQL, GSQL).

        Raises:
            ValueError: For a value with no literal all of them read the same
                way: a control character other than newline, carriage return
                and tab, a non-finite number, or an object of another type.
        """
        if value is None:
            return "null"
        if isinstance(value, bool):
            return "true" if value else "false"
        if isinstance(value, int):
            return str(value)
        if isinstance(value, float):
            if not math.isfinite(value):
                raise ValueError(f"{value!r} has no literal in a query")
            return repr(value)
        if isinstance(value, (datetime, date, time)):
            value = value.isoformat()
        if isinstance(value, str):
            if _UNWRITABLE.search(value):
                raise ValueError(
                    f"{value!r} holds a control character with no literal in a query"
                )
            escaped = _LITERAL_ESCAPED.sub(lambda m: _LITERAL_ESCAPES[m.group()], value)
            return f'"{escaped}"'
        if isinstance(value, (list, tuple)):
            return (
                "[" + ", ".join(FilterExpression._graph_literal(v) for v in value) + "]"
            )
        raise ValueError(
            f"a {type(value).__name__} value has no literal in a query: {value!r}"
        )

    @staticmethod
    def _sql_literal(value: Any) -> str:
        """One value as a SQL literal: strings and ISO datetimes single-quoted."""
        if value is None:
            return "null"
        if isinstance(value, (int, float)):
            return str(value)
        escaped = str(value).replace("'", "''")
        return f"'{escaped}'"

    def _cast_value(self, params: BoundParams | None = None) -> str:
        """The compared value: a placeholder of *params*, or a literal without it."""
        if self.cmp_operator == ComparisonOperator.IN:
            # Always a list, whatever its length: `x IN "a"` is not membership.
            if params is not None:
                return params.bind(self._in_members())
            members = ", ".join(self._graph_literal(v) for v in self._in_members())
            return f"[{members}]"
        operand: Any = self.value[0] if len(self.value) == 1 else list(self.value)
        if params is not None:
            return params.bind(operand)
        return self._graph_literal(operand)

    def _query_unary_op(self) -> str | None:
        """``unary_op`` as query text, or ``None`` when it only spells ``cmp_operator``.

        A dunder (``__eq__``) is the Python spelling of the comparison; the
        query languages write the comparison itself.
        """
        if self.unary_op is None or self.unary_op in DUNDER_TO_CMP:
            return None
        return self.unary_op

    def _cast_arango(self, doc_name: str, params: BoundParams | None = None) -> str:
        if self.cmp_operator == ComparisonOperator.IS_NULL:
            return f'{doc_name}["{self.field}"] == null'
        if self.cmp_operator == ComparisonOperator.IS_NOT_NULL:
            return f'{doc_name}["{self.field}"] != null'
        const = self._cast_value(params)
        lemma = f"{self.cmp_operator} {const}"
        if self._query_unary_op() is not None:
            lemma = f"{self._query_unary_op()} {lemma}"
        if self.field is not None:
            lemma = f'{doc_name}["{self.field}"] {lemma}'
        return lemma

    def _cast_cypher(self, doc_name: str, params: BoundParams | None = None) -> str:
        if self.cmp_operator == ComparisonOperator.IS_NULL:
            return f"{doc_name}.{self.field} IS NULL"
        if self.cmp_operator == ComparisonOperator.IS_NOT_NULL:
            return f"{doc_name}.{self.field} IS NOT NULL"
        const = self._cast_value(params)
        cmp_op = (
            "=" if self.cmp_operator == ComparisonOperator.EQ else self.cmp_operator
        )
        lemma = f"{cmp_op} {const}"
        if self._query_unary_op() is not None:
            lemma = f"{self._query_unary_op()} {lemma}"
        if self.field is not None:
            lemma = f"{doc_name}.{self.field} {lemma}"
        return lemma

    def _cast_ngql(self, doc_name: str) -> str:
        """Render leaf as nGQL expression (NebulaGraph 3.x).

        Uses dot-access like Cypher but keeps ``==`` for equality (nGQL standard).
        The caller passes *doc_name* as ``"v.TagName"`` so property access becomes
        ``v.TagName.field``.
        """
        if self.cmp_operator == ComparisonOperator.IS_NULL:
            return f"{doc_name}.{self.field} IS EMPTY"
        if self.cmp_operator == ComparisonOperator.IS_NOT_NULL:
            return f"{doc_name}.{self.field} IS NOT EMPTY"
        const = self._cast_value()
        lemma = f"{self.cmp_operator} {const}"
        if self._query_unary_op() is not None:
            lemma = f"{self._query_unary_op()} {lemma}"
        if self.field is not None:
            lemma = f"{doc_name}.{self.field} {lemma}"
        return lemma

    def _cast_tigergraph(self, doc_name: str) -> str:
        if self.cmp_operator == ComparisonOperator.IS_NULL:
            return f"{doc_name}.{self.field} IS NULL"
        if self.cmp_operator == ComparisonOperator.IS_NOT_NULL:
            return f"{doc_name}.{self.field} IS NOT NULL"
        const = self._cast_value()
        cmp_op = (
            "==" if self.cmp_operator == ComparisonOperator.EQ else self.cmp_operator
        )
        lemma = f"{cmp_op} {const}"
        if self._query_unary_op() is not None:
            lemma = f"{self._query_unary_op()} {lemma}"
        if self.field is not None:
            lemma = f"{doc_name}.{self.field} {lemma}"
        return lemma

    @staticmethod
    def _quote_sql_field(field: str) -> str:
        """Quote a SQL field name, handling dotted alias.column references.

        ``record_id``   -> ``"record_id"``
        ``s.record_id`` -> ``s."record_id"``
        """
        if "." in field:
            alias, col = field.split(".", 1)
            return f'{alias}."{col}"'
        return f'"{field}"'

    def _cast_sql(self, params: BoundParams | None = None) -> str:
        """Render leaf as SQL WHERE fragment: \"column\" op value.

        Values are placeholders of *params*, or without it literals with
        strings and dates single-quoted.
        """
        if not self.field:
            # An empty fragment would be dropped from the WHERE clause, and the
            # query would run without the condition.
            raise ValueError(
                f"a SQL condition needs a field: {self.cmp_operator} {self.value}"
            )
        quoted = self._quote_sql_field(self.field)
        if self.cmp_operator == ComparisonOperator.IS_NULL:
            return f"{quoted} IS NULL"
        if self.cmp_operator == ComparisonOperator.IS_NOT_NULL:
            return f"{quoted} IS NOT NULL"
        if self.cmp_operator == ComparisonOperator.IN:
            members = self._in_members()
            if not members:
                raise ValueError(f"IN on {self.field!r} needs at least one value")
            if params is not None:
                rendered = ", ".join(params.bind(v) for v in members)
            else:
                rendered = ", ".join(self._sql_literal(v) for v in members)
            return f"{quoted} IN ({rendered})"
        if self.cmp_operator == ComparisonOperator.EQ:
            op_str = "="
        elif self.cmp_operator == ComparisonOperator.NEQ:
            op_str = "!="
        elif self.cmp_operator in (
            ComparisonOperator.GT,
            ComparisonOperator.LT,
            ComparisonOperator.GE,
            ComparisonOperator.LE,
        ):
            op_str = str(self.cmp_operator)
        else:
            op_str = str(self.cmp_operator)
        value = self.value[0] if self.value else None
        if params is not None:
            return f"{quoted} {op_str} {params.bind(value)}"
        return f"{quoted} {op_str} {self._sql_literal(value)}"

    def _cast_restpp(self, field_types: dict[str, Any] | None = None) -> str:
        if not self.field:
            return ""
        if self.cmp_operator == ComparisonOperator.IS_NULL:
            return f'{self.field}=""'
        if self.cmp_operator == ComparisonOperator.IS_NOT_NULL:
            return f'{self.field}!=""'
        if self.cmp_operator == ComparisonOperator.IN:
            # The REST filter syntax compares one attribute with one value.
            raise ValueError(
                f"IN on {self.field!r} is not expressible as a TigerGraph REST filter"
            )
        if self.cmp_operator == ComparisonOperator.EQ:
            op_str = "="
        elif self.cmp_operator == ComparisonOperator.NEQ:
            op_str = "!="
        elif self.cmp_operator == ComparisonOperator.GT:
            op_str = ">"
        elif self.cmp_operator == ComparisonOperator.LT:
            op_str = "<"
        elif self.cmp_operator == ComparisonOperator.GE:
            op_str = ">="
        elif self.cmp_operator == ComparisonOperator.LE:
            op_str = "<="
        else:
            op_str = str(self.cmp_operator)
        value = self.value[0] if self.value else None
        if value is None:
            value_str = "null"
        elif isinstance(value, (int, float)):
            value_str = str(value)
        elif isinstance(value, str):
            if '"' in value or "," in value:
                # A comma separates REST filter conditions and a quote ends the
                # value; there is no escape for either.
                raise ValueError(
                    f"{value!r} cannot be written in a TigerGraph REST filter"
                )
            is_string_field = True
            if field_types and self.field in field_types:
                field_type = field_types[self.field]
                field_type_str = (
                    field_type.value
                    if hasattr(field_type, "value")
                    else str(field_type).upper()
                )
                if field_type_str in ("INT", "UINT", "FLOAT", "DOUBLE"):
                    is_string_field = False
            value_str = f'"{value}"' if is_string_field else str(value)
        else:
            value_str = str(value)
        return f"{self.field}{op_str}{value_str}"

    def _cast_python(self, doc: Mapping[str, Any]) -> bool:
        if self.field is None:
            return False
        field_val = doc.get(self.field)
        if self.cmp_operator == ComparisonOperator.IS_NULL:
            return field_val is None
        if self.cmp_operator == ComparisonOperator.IS_NOT_NULL:
            return field_val is not None
        if field_val is None:
            return False
        if self.cmp_operator == ComparisonOperator.IN:
            return field_val in self._in_members()
        # List-form expressions (["==", value, field]) carry no unary_op; fall
        # back to the dunder implied by cmp_operator so they evaluate the same
        # way they render.
        dunder = self.unary_op
        if dunder is None and self.cmp_operator is not None:
            dunder = CMP_TO_DUNDER.get(self.cmp_operator)
        if dunder is None or not self.value:
            return False
        comparison = getattr(field_val, dunder, None)
        if comparison is None:
            return False
        result = comparison(self.value[0])
        return result is True

    @staticmethod
    def _wrap_composite_operand(
        dep: FilterExpression, rendered: str, kind: ExpressionFlavor
    ) -> str:
        """Parenthesize nested composite operands so precedence is explicit.

        Every target language binds AND tighter than OR, so an unparenthesized
        ``AND[OR[a, b], c]`` silently means ``a OR (b AND c)``. Wrapping nested
        composites keeps the rendered predicate faithful to the expression tree
        on all flavors.
        """
        if (
            kind
            in (
                ExpressionFlavor.SQL,
                ExpressionFlavor.CYPHER,
                ExpressionFlavor.AQL,
                ExpressionFlavor.NGQL,
                ExpressionFlavor.GSQL,
            )
            and dep.kind == "composite"
        ):
            return f"({rendered})"
        return rendered

    def _render_dep(
        self,
        dep: FilterExpression,
        doc_name: str,
        kind: ExpressionFlavor,
        params: BoundParams | None = None,
    ) -> str:
        rendered = str(dep(kind=kind, doc_name=doc_name, params=params))
        return self._wrap_composite_operand(dep, rendered, kind)

    def _cast_generic(
        self, doc_name: str, kind: ExpressionFlavor, params: BoundParams | None = None
    ) -> str:
        if self.operator is None:
            raise ValueError("composite expression requires operator")
        if (
            kind == ExpressionFlavor.SQL
            and self.operator == LogicalOperator.IMPLICATION
        ):
            if len(self.deps) != 2:
                raise ValueError("IF_THEN composite requires exactly 2 deps")
            antecedent = self._render_dep(self.deps[0], doc_name, kind, params)
            consequent = self._render_dep(self.deps[1], doc_name, kind, params)
            return f"(NOT ({antecedent}) OR ({consequent}))"
        if len(self.deps) == 1:
            if self.operator == LogicalOperator.NOT:
                result = self._render_dep(self.deps[0], doc_name, kind, params)
                if doc_name == "" and kind == ExpressionFlavor.GSQL:
                    return f"!{result}"
                return f"NOT {result}"
            raise ValueError(
                f" length of deps = {len(self.deps)} but operator is not {LogicalOperator.NOT}"
            )
        deps_str_cast = [
            self._render_dep(dep, doc_name, kind, params) for dep in self.deps
        ]
        if doc_name == "" and kind == ExpressionFlavor.GSQL:
            if self.operator == LogicalOperator.AND:
                return " && ".join(deps_str_cast)
            if self.operator == LogicalOperator.OR:
                return " || ".join(deps_str_cast)
        return f" {self.operator} ".join(deps_str_cast)

    def _cast_python_composite(self, doc: Mapping[str, Any]) -> bool:
        if self.operator is None:
            raise ValueError("composite expression requires operator")
        if len(self.deps) == 1:
            if self.operator == LogicalOperator.NOT:
                return not self.deps[0].matches(doc)
            raise ValueError(
                f" length of deps = {len(self.deps)} but operator is not {LogicalOperator.NOT}"
            )
        return OperatorMapping[self.operator]([dep.matches(doc) for dep in self.deps])

Attributes

cmp_operator = None class-attribute instance-attribute

deps = Field(default_factory=list) class-attribute instance-attribute

field = None class-attribute instance-attribute

kind instance-attribute

operator = None class-attribute instance-attribute

unary_op = None class-attribute instance-attribute

value = Field(default_factory=list) class-attribute instance-attribute

Methods:

__call__(doc_name='doc', kind=ExpressionFlavor.AQL, **kwargs)

Render or evaluate the expression in the target language.

With params= a :class:BoundParams, values are written as its placeholders; without it, as literals.

Source code in graflo/filter/onto.py

def __call__(
    self,
    doc_name="doc",
    kind: ExpressionFlavor = ExpressionFlavor.AQL,
    **kwargs,
) -> str | bool:
    """Render or evaluate the expression in the target language.

    With ``params=`` a :class:`BoundParams`, values are written as its
    placeholders; without it, as literals.
    """
    if self.kind == "leaf":
        return self._call_leaf(doc_name=doc_name, kind=kind, **kwargs)
    return self._call_composite(doc_name=doc_name, kind=kind, **kwargs)

check_discriminated_shape()

Enforce exactly one shape per kind and normalise null-check operators.

Source code in graflo/filter/onto.py

@model_validator(mode="after")
def check_discriminated_shape(self) -> FilterExpression:
    """Enforce exactly one shape per kind and normalise null-check operators."""
    if self.kind == "leaf":
        if self.operator is not None or self.deps:
            raise ValueError("leaf expression must not have operator or deps")
        if self.cmp_operator is None and self.unary_op is None:
            raise ValueError(
                "leaf expression requires cmp_operator or operator"
                + (f" (field {self.field!r})" if self.field else "")
            )
        # IS_NULL / IS_NOT_NULL are unary; clear any spurious value list
        if self.cmp_operator in (
            ComparisonOperator.IS_NULL,
            ComparisonOperator.IS_NOT_NULL,
        ):
            object.__setattr__(self, "value", [])
    else:
        if self.operator is None:
            raise ValueError("composite expression must have operator")
    return self

from_dict(data) classmethod

Create a filter expression from a dictionary or list.

Returns FilterExpression (leaf or composite). LSP-compliant: return type is Self.

Source code in graflo/filter/onto.py

@classmethod
def from_dict(cls, data: dict[str, Any] | list[Any]) -> Self:
    """Create a filter expression from a dictionary or list.

    Returns FilterExpression (leaf or composite). LSP-compliant: return type is Self.
    """
    if isinstance(data, list):
        if data[0] in ComparisonOperator:
            return cast(Self, cls.from_list(data))
        elif data[0] in LogicalOperator:
            return cls(kind="composite", operator=data[0], deps=data[1])
        raise ValueError(
            f"filter list must start with a comparison or logical operator, "
            f"got {data[0]!r}"
        )
    elif isinstance(data, dict):
        if data.get("kind") is not None:
            return cls.model_validate(data)
        k = next(iter(data.keys()))
        norm_k = k.upper() if isinstance(k, str) else k
        if norm_k in LogicalOperator:
            if len(data) > 1:
                raise ValueError(
                    "a logical filter entry takes one logical operator key, "
                    f"got {sorted(str(key) for key in data)}"
                )
            deps: list[FilterExpression] = [cls.from_dict(v) for v in data[k]]
            return cls(
                kind="composite", operator=LogicalOperator(norm_k), deps=deps
            )
        else:
            # The model is built by hand below, so `extra="forbid"` never
            # sees these keys: an unknown one would be dropped silently.
            if "foo" in data:
                raise ValueError(_FOO_REMOVED)
            unknown = sorted(str(key) for key in data if key not in _LEAF_KEYS)
            if unknown:
                raise ValueError(
                    f"unknown filter key(s) {unknown}; a leaf takes "
                    f"{sorted(_LEAF_KEYS - {'kind', 'unary_op'})}, a "
                    f"logical entry one of {sorted(_LOGICAL_OPERATOR_JSON_VALUES)}"
                )
            unary_op = data.get("operator")
            cmp_operator = data.get("cmp_operator")
            if cmp_operator is None and unary_op is not None:
                cmp_operator = DUNDER_TO_CMP.get(unary_op)
            return cls(
                kind="leaf",
                cmp_operator=cmp_operator,
                value=data.get("value", []),
                field=data.get("field"),
                unary_op=unary_op,
            )
    raise ValueError(f"expected dict or list, got {type(data)}")

from_list(current) classmethod

Build a leaf expression from list form [cmp_operator, value, field?, unary_op?].

Source code in graflo/filter/onto.py

@classmethod
def from_list(cls, current: list[Any]) -> FilterExpression:
    """Build a leaf expression from list form [cmp_operator, value, field?, unary_op?]."""
    cmp_operator = current[0]
    value = current[1]
    field = current[2] if len(current) > 2 else None
    unary_op = current[3] if len(current) > 3 else None
    return cls(
        kind="leaf",
        cmp_operator=cmp_operator,
        value=value,
        field=field,
        unary_op=unary_op,
    )

leaf_operator_to_unary_op(data) classmethod

Map leaf 'operator' (YAML/kwargs) to unary_op; infer kind and cmp_operator.

Source code in graflo/filter/onto.py

@model_validator(mode="before")
@classmethod
def leaf_operator_to_unary_op(cls, data: Any) -> Any:
    """Map leaf 'operator' (YAML/kwargs) to unary_op; infer kind and cmp_operator."""
    if not isinstance(data, dict):
        return data
    data = dict(data)
    if data.get("kind") == "composite":
        return data

    if data.get("kind") is None:
        op = data.get("operator")
        if isinstance(op, LogicalOperator):
            data["kind"] = "composite"
            return data
        if isinstance(op, str) and op in _LOGICAL_OPERATOR_JSON_VALUES:
            data["kind"] = "composite"
            return data
        deps = data.get("deps")
        if isinstance(deps, list) and len(deps) > 0:
            data["kind"] = "composite"
            return data
        if data.get("cmp_operator") is not None or data.get("field") is not None:
            data["kind"] = "leaf"

    if "foo" in data:
        raise ValueError(_FOO_REMOVED)
    raw_op = None
    if "operator" in data and isinstance(data["operator"], str):
        raw_op = data.pop("operator")
    if raw_op is not None:
        data["unary_op"] = raw_op
        if data.get("cmp_operator") is None and raw_op in DUNDER_TO_CMP:
            data["cmp_operator"] = DUNDER_TO_CMP[raw_op]
        if data.get("kind") is None:
            data["kind"] = "leaf"
    return data

matches(doc)

Whether doc satisfies the expression, evaluated in Python.

Prefer this to expr(kind=ExpressionFlavor.PYTHON, **doc), which cannot evaluate a document with a field named kind or doc_name.

Source code in graflo/filter/onto.py

def matches(self, doc: Mapping[str, Any]) -> bool:
    """Whether *doc* satisfies the expression, evaluated in Python.

    Prefer this to ``expr(kind=ExpressionFlavor.PYTHON, **doc)``, which
    cannot evaluate a document with a field named ``kind`` or ``doc_name``.
    """
    if self.kind == "leaf":
        return self._cast_python(doc)
    return self._cast_python_composite(doc)

parse_deps(v) classmethod

Parse dict/list items into FilterExpression instances.

Source code in graflo/filter/onto.py

@field_validator("deps", mode="before")
@classmethod
def parse_deps(cls, v: list[Any]) -> list[Any]:
    """Parse dict/list items into FilterExpression instances."""
    if not isinstance(v, list):
        return v
    result = []
    for item in v:
        if isinstance(item, (dict, list)):
            # Not `from_dict`: a serialized expression (`kind` / `operator`
            # + `deps`) has to come back as the expression it was.
            result.append(parse_filter_expression(item))
        else:
            result.append(item)
    return result

rename_fields(renames)

This expression with every leaf field rewritten by renames.

Pure: returns a new expression and leaves this one untouched. Fields absent from renames keep their name.

Source code in graflo/filter/onto.py

def rename_fields(self, renames: Mapping[str, str]) -> FilterExpression:
    """This expression with every leaf ``field`` rewritten by *renames*.

    Pure: returns a new expression and leaves this one untouched. Fields
    absent from *renames* keep their name.
    """
    if not renames:
        return self
    if self.kind == "leaf":
        if self.field is None or self.field not in renames:
            return self
        return self.model_copy(update={"field": renames[self.field]})
    return self.model_copy(
        update={"deps": [dep.rename_fields(renames) for dep in self.deps]}
    )

value_to_list(v) classmethod

Convert single value to list if necessary. Explicit None becomes [None] for null comparison.

Source code in graflo/filter/onto.py

@field_validator("value", mode="before")
@classmethod
def value_to_list(cls, v: list[Any] | Any) -> list[Any]:
    """Convert single value to list if necessary. Explicit None becomes [None] for null comparison."""
    if v is None:
        return [None]
    if isinstance(v, list):
        return v
    return [v]

JoinClause

Bases: ConfigBaseModel

Specification for a SQL JOIN operation.

Used by TableConnector to describe multi-table queries. Each JoinClause adds one JOIN to the generated SQL. The base row uses TableConnector.base_alias (default base), not a hard-coded name.

Attributes:

Name	Type	Description
table	str	Table name to join (e.g. "all_classes").
schema_name	str | None	Optional schema override for the joined table.
alias	str | None	SQL alias for the joined table (e.g. "s", "t"). Required when the same table is joined more than once.
on_self	str	Column on the base (left) table used in the ON condition.
on_other	str	Column on the joined (right) table used in the ON condition.
join_type	str	Type of join -- LEFT, INNER, etc. Defaults to LEFT.
select_fields	list[str] | None	Explicit list of columns to SELECT from this join. When None every column of the joined table is included (aliased with the join alias prefix).

Source code in graflo/filter/select.py

class JoinClause(ConfigBaseModel):
    """Specification for a SQL JOIN operation.

    Used by TableConnector to describe multi-table queries. Each JoinClause
    adds one JOIN to the generated SQL. The base row uses ``TableConnector.base_alias``
    (default ``base``), not a hard-coded name.

    Attributes:
        table: Table name to join (e.g. "all_classes").
        schema_name: Optional schema override for the joined table.
        alias: SQL alias for the joined table (e.g. "s", "t"). Required when
            the same table is joined more than once.
        on_self: Column on the base (left) table used in the ON condition.
        on_other: Column on the joined (right) table used in the ON condition.
        join_type: Type of join -- LEFT, INNER, etc. Defaults to LEFT.
        select_fields: Explicit list of columns to SELECT from this join.
            When None every column of the joined table is included (aliased
            with the join alias prefix).
    """

    table: str = Field(..., description="Table name to join.")
    schema_name: str | None = Field(
        default=None, description="Schema override for the joined table."
    )
    alias: str | None = Field(
        default=None, description="SQL alias for the joined table."
    )
    on_self: str = Field(
        ..., description="Column on the base table for the ON condition."
    )
    on_other: str = Field(
        ..., description="Column on the joined table for the ON condition."
    )
    join_type: str = Field(default="LEFT", description="JOIN type (LEFT, INNER, etc.).")
    select_fields: list[str] | None = Field(
        default=None,
        description="Columns to SELECT from this join (None = all columns).",
    )

Attributes

alias = Field(default=None, description='SQL alias for the joined table.') class-attribute instance-attribute

join_type = Field(default='LEFT', description='JOIN type (LEFT, INNER, etc.).') class-attribute instance-attribute

on_other = Field(..., description='Column on the joined table for the ON condition.') class-attribute instance-attribute

on_self = Field(..., description='Column on the base table for the ON condition.') class-attribute instance-attribute

schema_name = Field(default=None, description='Schema override for the joined table.') class-attribute instance-attribute

select_fields = Field(default=None, description='Columns to SELECT from this join (None = all columns).') class-attribute instance-attribute

table = Field(..., description='Table name to join.') class-attribute instance-attribute

LogicalOperator

Bases: BaseEnum

Logical operators for combining filter conditions.

Attributes:

Name	Type	Description
AND		Logical AND operation
OR		Logical OR operation
NOT		Logical NOT operation
IMPLICATION		Logical IF-THEN operation

Source code in graflo/filter/onto.py

class LogicalOperator(BaseEnum):
    """Logical operators for combining filter conditions.

    Attributes:
        AND: Logical AND operation
        OR: Logical OR operation
        NOT: Logical NOT operation
        IMPLICATION: Logical IF-THEN operation
    """

    AND = "AND"
    OR = "OR"
    NOT = "NOT"
    IMPLICATION = "IF_THEN"

Attributes

AND = 'AND' class-attribute instance-attribute

IMPLICATION = 'IF_THEN' class-attribute instance-attribute

NOT = 'NOT' class-attribute instance-attribute

OR = 'OR' class-attribute instance-attribute

SelectSpec

Bases: ConfigBaseModel

Declarative view specification emulating SQL SELECT structure.

Alternative to TableConnector's table_name + joins + filters. Supports from_dict() for YAML/JSON loading (like FilterExpression).

Attributes:

Name	Type	Description
kind	Literal['select', 'type_lookup']	"select" for full spec, "type_lookup" for shorthand
from_	str | None	Base table (used when kind="select")
joins	list[JoinClause | dict[str, Any]]	JOIN clauses (used when kind="select")
select	list[str | dict[str, Any]]	SELECT list (used when kind="select"). Each entry may be: - A string: all_base (all columns from the base row), * (SQL wildcard), a raw SQL fragment, or a simple identifier for a base column (qualified with base_alias when joins are present). - A dict {expr, alias} (legacy SQL expression). - A dict {base, as?} for a base-table column (optional output alias). - A dict {from_join, column, as?} for a column from a joined table (from_join is the join alias from joins).
where	FilterExpression | dict[str, Any] | None	WHERE clause, reuses FilterExpression
table	str | None	Lookup table (type_lookup only)
identity	str | None	Identity column in lookup table (type_lookup only)
type_column	str | None	Type discriminator column (type_lookup only)
source	str | None	FK column on base table for source (type_lookup only)
target	str | None	FK column on base table for target (type_lookup only)
relation	str | None	Relation column in base table (type_lookup only, optional)
source_table	str | None	Lookup table for the source side (defaults to table).
target_table	str | None	Lookup table for the target side (defaults to table).
source_identity	str | None	Join column on the source lookup (defaults to identity).
target_identity	str | None	Join column on the target lookup (defaults to identity).
source_type_column	str | None	Type discriminator on source lookup (defaults to type_column).
target_type_column	str | None	Type discriminator on target lookup (defaults to type_column).
base_alias	str	SQL alias for the base table row when joins are present (kind="select" and joins, or kind="type_lookup"). Default base; must not collide with join aliases.

Source code in graflo/filter/select.py

class SelectSpec(ConfigBaseModel):
    """Declarative view specification emulating SQL SELECT structure.

    Alternative to TableConnector's table_name + joins + filters.
    Supports from_dict() for YAML/JSON loading (like FilterExpression).

    Attributes:
        kind: "select" for full spec, "type_lookup" for shorthand
        from_: Base table (used when kind="select")
        joins: JOIN clauses (used when kind="select")
        select: SELECT list (used when kind="select"). Each entry may be:
            - A string: ``all_base`` (all columns from the base row), ``*`` (SQL
              wildcard), a raw SQL fragment, or a simple identifier for a **base
              column** (qualified with ``base_alias`` when joins are present).
            - A dict ``{expr, alias}`` (legacy SQL expression).
            - A dict ``{base, as?}`` for a base-table column (optional output alias).
            - A dict ``{from_join, column, as?}`` for a column from a joined table
              (``from_join`` is the join alias from ``joins``).
        where: WHERE clause, reuses FilterExpression
        table: Lookup table (type_lookup only)
        identity: Identity column in lookup table (type_lookup only)
        type_column: Type discriminator column (type_lookup only)
        source: FK column on base table for source (type_lookup only)
        target: FK column on base table for target (type_lookup only)
        relation: Relation column in base table (type_lookup only, optional)
        source_table: Lookup table for the source side (defaults to table).
        target_table: Lookup table for the target side (defaults to table).
        source_identity: Join column on the source lookup (defaults to identity).
        target_identity: Join column on the target lookup (defaults to identity).
        source_type_column: Type discriminator on source lookup (defaults to
            type_column).
        target_type_column: Type discriminator on target lookup (defaults to
            type_column).
        base_alias: SQL alias for the base table row when joins are present
            (``kind="select"`` and ``joins``, or ``kind="type_lookup"``). Default
            ``base``; must not collide with join aliases.
    """

    kind: Literal["select", "type_lookup"] = "select"

    # Full select form
    from_: str | None = Field(default=None, validation_alias="from")
    joins: list[JoinClause | dict[str, Any]] = Field(default_factory=list)
    select: list[str | dict[str, Any]] = Field(
        default_factory=_default_select_columns,
        description=(
            "Default all_base expands to base.* when joins exist, else SELECT *."
        ),
    )
    where: FilterExpression | dict[str, Any] | None = None
    base_alias: str = Field(
        default="base",
        description="SQL table alias for the base row (when joins are generated).",
    )

    # Type-lookup shorthand
    table: str | None = None
    identity: str | None = None
    type_column: str | None = None
    source: str | None = None
    target: str | None = None
    relation: str | None = None
    source_table: str | None = None
    target_table: str | None = None
    source_identity: str | None = None
    target_identity: str | None = None
    source_type_column: str | None = None
    target_type_column: str | None = None

    @field_validator("where", mode="before")
    @classmethod
    def _check_where(cls, v: Any) -> Any:
        """Parse ``where`` when the view loads, so a bad entry is reported then.

        The value is kept as authored: replacing a mapping with its parsed model
        would change the canonical form of the manifest that carries it.
        """
        if isinstance(v, (dict, list)):
            try:
                parse_filter_expression(v)
            except ValueError as e:
                raise ValueError(f"where: {e}") from e
        return v

    def where_expression(self) -> FilterExpression | None:
        """The ``where`` clause as a filter expression, or ``None``."""
        if not self.where:
            return None
        return parse_filter_expression(self.where)

    def effective_base_alias(self) -> str | None:
        """The alias the generated query gives the base table, or ``None``.

        A ``type_lookup`` view always joins, so it always aliases. A ``select``
        view aliases the base table only when it has joins; conditions added to
        a join-less view must name columns unqualified.
        """
        if self.kind == "type_lookup" or self.joins:
            return self.base_alias
        return None

    @model_validator(mode="after")
    def _validate_type_lookup_side_options(self) -> Self:
        if self.kind != "type_lookup":
            return self

        def eff_source_lookup() -> tuple[str | None, str | None, str | None]:
            t = self.source_table if self.source_table is not None else self.table
            i = (
                self.source_identity
                if self.source_identity is not None
                else self.identity
            )
            c = (
                self.source_type_column
                if self.source_type_column is not None
                else self.type_column
            )
            return (t, i, c)

        def eff_target_lookup() -> tuple[str | None, str | None, str | None]:
            t = self.target_table if self.target_table is not None else self.table
            i = (
                self.target_identity
                if self.target_identity is not None
                else self.identity
            )
            c = (
                self.target_type_column
                if self.target_type_column is not None
                else self.type_column
            )
            return (t, i, c)

        t, i, c = eff_source_lookup()
        if not all([t, i, c]):
            raise ValueError(
                "type_lookup requires table, identity, type_column (or "
                "source_table / source_identity / source_type_column) for the "
                "source lookup join"
            )
        t, i, c = eff_target_lookup()
        if not all([t, i, c]):
            raise ValueError(
                "type_lookup requires table, identity, type_column (or "
                "target_table / target_identity / target_type_column) for the "
                "target lookup join"
            )
        return self

    @model_validator(mode="after")
    def _validate_base_alias(self) -> Self:
        if not _SIMPLE_IDENT.match(self.base_alias):
            raise ValueError(
                "base_alias must be a simple SQL identifier "
                "(ASCII letter, digit, underscore)"
            )
        if self.kind == "type_lookup":
            if self.base_alias in ("s", "t"):
                raise ValueError(
                    "base_alias must not be 's' or 't' (reserved for type_lookup "
                    "lookup join aliases)"
                )
            return self
        join_aliases = _collect_join_aliases(self.joins)
        if self.base_alias in join_aliases:
            raise ValueError(
                f"base_alias {self.base_alias!r} conflicts with a join alias "
                f"{sorted(join_aliases)}"
            )
        return self

    def build_sql(
        self,
        schema: str,
        base_table: str,
        extra_conditions: Sequence[str] = (),
    ) -> str:
        """Build SQL SELECT query.

        Args:
            schema: Schema name (e.g. "public")
            base_table: Base table name (from TableConnector.table_name)
            extra_conditions: Rendered conditions to AND onto the view's own
                ``WHERE``. Each must already be safe as one operand of ``AND``
                (see :func:`~graflo.filter.onto.render_conjunct`) and name base
                columns through :meth:`effective_base_alias`.

        Returns:
            Complete SQL query string
        """
        if self.kind == "type_lookup":
            return self._build_type_lookup_sql(schema, base_table, extra_conditions)
        return self._build_select_sql(schema, base_table, extra_conditions)

    def _build_type_lookup_sql(
        self, schema: str, base_table: str, extra_conditions: Sequence[str] = ()
    ) -> str:
        """Expand type_lookup shorthand to full SQL."""
        if not self.source or not self.target:
            raise ValueError("type_lookup requires source and target column names")

        src_fk = self.source
        tgt_fk = self.target
        rel_col = self.relation

        src_tbl = self.source_table if self.source_table is not None else self.table
        src_ident = (
            self.source_identity if self.source_identity is not None else self.identity
        )
        src_type_col = (
            self.source_type_column
            if self.source_type_column is not None
            else self.type_column
        )

        tgt_tbl = self.target_table if self.target_table is not None else self.table
        tgt_ident = (
            self.target_identity if self.target_identity is not None else self.identity
        )
        tgt_type_col = (
            self.target_type_column
            if self.target_type_column is not None
            else self.type_column
        )

        base_ref = f'"{schema}"."{base_table}"'
        ba = self.base_alias

        assert src_type_col is not None and tgt_type_col is not None
        select_parts: list[str] = [
            f'{ba}."{src_fk}" AS source_id',
            f's."{src_type_col}" AS source_type',
            f'{ba}."{tgt_fk}" AS target_id',
            f't."{tgt_type_col}" AS target_type',
        ]

        if rel_col:
            select_parts.append(f'{ba}."{rel_col}" AS relation')
        select_clause = ", ".join(select_parts)

        assert src_tbl is not None and src_ident is not None
        assert tgt_tbl is not None and tgt_ident is not None
        s_ref = f'"{schema}"."{src_tbl}"'
        t_ref = f'"{schema}"."{tgt_tbl}"'
        from_clause = (
            f"{base_ref} {ba} "
            f'LEFT JOIN {s_ref} s ON {ba}."{src_fk}" = s."{src_ident}" '
            f'LEFT JOIN {t_ref} t ON {ba}."{tgt_fk}" = t."{tgt_ident}"'
        )

        conditions = [
            f's."{src_ident}" IS NOT NULL',
            f't."{tgt_ident}" IS NOT NULL',
            *extra_conditions,
        ]
        where_clause = " AND ".join(conditions)
        return f"SELECT {select_clause} FROM {from_clause} WHERE {where_clause}"

    def _build_select_sql(
        self, schema: str, base_table: str, extra_conditions: Sequence[str] = ()
    ) -> str:
        """Build SQL from full select spec."""
        from_table = self.from_ or base_table
        base_ref = f'"{schema}"."{from_table}"'
        base_alias = self.effective_base_alias()
        if base_alias:
            base_ref_aliased = f"{base_ref} {base_alias}"
        else:
            base_ref_aliased = base_ref

        # SELECT
        join_aliases = _collect_join_aliases(self.joins)
        select_parts: list[str] = []
        for item in self.select:
            if isinstance(item, str):
                select_parts.append(_render_select_string(item, base_alias))
            elif isinstance(item, dict):
                select_parts.append(_render_select_dict(item, base_alias, join_aliases))
        select_clause = ", ".join(select_parts) if select_parts else "*"

        # FROM + JOINs
        from_clause = base_ref_aliased
        for j in self.joins:
            jc = JoinClause.model_validate(j) if isinstance(j, dict) else j
            jc_schema = jc.schema_name or schema
            alias = jc.alias or jc.table
            join_ref = f'"{jc_schema}"."{jc.table}"'
            left_col = (
                f'{base_alias}."{jc.on_self}"' if base_alias else f'"{jc.on_self}"'
            )
            right_col = f'{alias}."{jc.on_other}"'
            from_clause += (
                f" {jc.join_type} JOIN {join_ref} {alias} ON {left_col} = {right_col}"
            )

        query = f"SELECT {select_clause} FROM {from_clause}"

        # WHERE
        conditions: list[str] = []
        where = self.where_expression()
        if where is not None:
            conditions.append(render_conjunct(where, kind=ExpressionFlavor.SQL))
        conditions.extend(extra_conditions)
        if conditions:
            query += f" WHERE {' AND '.join(conditions)}"

        return query

    @classmethod
    def concat_select_parts(cls, head: Self, *tail: Self) -> Self:
        """Merge ``kind="select"`` join/select fragments into one spec.

        ``head`` may set ``from_`` and ``where``. Each ``tail`` fragment must omit
        ``from_`` and ``where`` so the logical base table and filters stay on
        ``head``. ``joins`` and ``select`` are concatenated in order.

        Omitting ``from`` on the merged spec is fine: :meth:`build_sql` uses
        ``TableConnector.table_name`` as the base when ``from_`` is unset.

        This is the structured equivalent of listing two join/select blocks in
        one YAML ``view: select``, described in
        ``docs/concepts/connectors/table_views.md``.
        """
        if head.kind != "select":
            raise ValueError("head must use kind='select'")
        for part in tail:
            if part.kind != "select":
                raise ValueError("all fragments must use kind='select'")
            if part.from_ is not None:
                raise ValueError(
                    "tail SelectSpec fragments must omit `from` "
                    "(base table is TableConnector.table_name at build_sql time)"
                )
            if part.where is not None:
                raise ValueError(
                    "tail fragments must omit `where`; set it on `head` only"
                )
        merged_joins: list[JoinClause | dict[str, Any]] = list(head.joins)
        merged_select: list[str | dict[str, Any]] = list(head.select)
        for part in tail:
            merged_joins.extend(part.joins)
            merged_select.extend(part.select)
        return cls(
            kind="select",
            from_=head.from_,
            joins=merged_joins,
            select=merged_select if merged_select else [ALL_BASE_COLUMNS],
            where=head.where,
            base_alias=head.base_alias,
        )

    @classmethod
    def from_dict(cls, data: dict[str, Any] | list[Any]) -> Self:
        """Create SelectSpec from dictionary (YAML/JSON friendly).

        Supports:
        - kind="type_lookup": table, identity, type_column, source, target, relation,
          optional per-side source_* / target_* lookup overrides
        - kind="select": from, joins, select (``all_base``, strings, base/from_join
          dicts, or legacy expr/alias dicts), where, base_alias
        """
        if isinstance(data, list):
            return cls.model_validate(data)
        data = dict(data)
        kind = data.pop("kind", "select")
        if kind == "type_lookup":
            return cls(
                kind="type_lookup",
                **{k: v for k, v in data.items() if k != "from" and v is not None},
            )
        # Normalize "from" -> from_
        if "from" in data:
            data["from_"] = data.pop("from")
        return cls(kind="select", **data)

Attributes

base_alias = Field(default='base', description='SQL table alias for the base row (when joins are generated).') class-attribute instance-attribute

from_ = Field(default=None, validation_alias='from') class-attribute instance-attribute

identity = None class-attribute instance-attribute

joins = Field(default_factory=list) class-attribute instance-attribute

kind = 'select' class-attribute instance-attribute

relation = None class-attribute instance-attribute

select = Field(default_factory=_default_select_columns, description='Default all_base expands to base.* when joins exist, else SELECT *.') class-attribute instance-attribute

source = None class-attribute instance-attribute

source_identity = None class-attribute instance-attribute

source_table = None class-attribute instance-attribute

source_type_column = None class-attribute instance-attribute

table = None class-attribute instance-attribute

target = None class-attribute instance-attribute

target_identity = None class-attribute instance-attribute

target_table = None class-attribute instance-attribute

target_type_column = None class-attribute instance-attribute

type_column = None class-attribute instance-attribute

where = None class-attribute instance-attribute

Methods:

build_sql(schema, base_table, extra_conditions=())

Build SQL SELECT query.

Parameters:

Name	Type	Description	Default
schema	str	Schema name (e.g. "public")	required
base_table	str	Base table name (from TableConnector.table_name)	required
extra_conditions	Sequence[str]	Rendered conditions to AND onto the view's own WHERE. Each must already be safe as one operand of AND (see :func:~graflo.filter.onto.render_conjunct) and name base columns through :meth:effective_base_alias.	()

Returns:

Type	Description
str	Complete SQL query string

Source code in graflo/filter/select.py

def build_sql(
    self,
    schema: str,
    base_table: str,
    extra_conditions: Sequence[str] = (),
) -> str:
    """Build SQL SELECT query.

    Args:
        schema: Schema name (e.g. "public")
        base_table: Base table name (from TableConnector.table_name)
        extra_conditions: Rendered conditions to AND onto the view's own
            ``WHERE``. Each must already be safe as one operand of ``AND``
            (see :func:`~graflo.filter.onto.render_conjunct`) and name base
            columns through :meth:`effective_base_alias`.

    Returns:
        Complete SQL query string
    """
    if self.kind == "type_lookup":
        return self._build_type_lookup_sql(schema, base_table, extra_conditions)
    return self._build_select_sql(schema, base_table, extra_conditions)

concat_select_parts(head, *tail) classmethod

Merge kind="select" join/select fragments into one spec.

head may set from_ and where. Each tail fragment must omit from_ and where so the logical base table and filters stay on head. joins and select are concatenated in order.

Omitting from on the merged spec is fine: :meth:build_sql uses TableConnector.table_name as the base when from_ is unset.

This is the structured equivalent of listing two join/select blocks in one YAML view: select, described in docs/concepts/connectors/table_views.md.

Source code in graflo/filter/select.py

@classmethod
def concat_select_parts(cls, head: Self, *tail: Self) -> Self:
    """Merge ``kind="select"`` join/select fragments into one spec.

    ``head`` may set ``from_`` and ``where``. Each ``tail`` fragment must omit
    ``from_`` and ``where`` so the logical base table and filters stay on
    ``head``. ``joins`` and ``select`` are concatenated in order.

    Omitting ``from`` on the merged spec is fine: :meth:`build_sql` uses
    ``TableConnector.table_name`` as the base when ``from_`` is unset.

    This is the structured equivalent of listing two join/select blocks in
    one YAML ``view: select``, described in
    ``docs/concepts/connectors/table_views.md``.
    """
    if head.kind != "select":
        raise ValueError("head must use kind='select'")
    for part in tail:
        if part.kind != "select":
            raise ValueError("all fragments must use kind='select'")
        if part.from_ is not None:
            raise ValueError(
                "tail SelectSpec fragments must omit `from` "
                "(base table is TableConnector.table_name at build_sql time)"
            )
        if part.where is not None:
            raise ValueError(
                "tail fragments must omit `where`; set it on `head` only"
            )
    merged_joins: list[JoinClause | dict[str, Any]] = list(head.joins)
    merged_select: list[str | dict[str, Any]] = list(head.select)
    for part in tail:
        merged_joins.extend(part.joins)
        merged_select.extend(part.select)
    return cls(
        kind="select",
        from_=head.from_,
        joins=merged_joins,
        select=merged_select if merged_select else [ALL_BASE_COLUMNS],
        where=head.where,
        base_alias=head.base_alias,
    )

effective_base_alias()

The alias the generated query gives the base table, or None.

A type_lookup view always joins, so it always aliases. A select view aliases the base table only when it has joins; conditions added to a join-less view must name columns unqualified.

Source code in graflo/filter/select.py

def effective_base_alias(self) -> str | None:
    """The alias the generated query gives the base table, or ``None``.

    A ``type_lookup`` view always joins, so it always aliases. A ``select``
    view aliases the base table only when it has joins; conditions added to
    a join-less view must name columns unqualified.
    """
    if self.kind == "type_lookup" or self.joins:
        return self.base_alias
    return None

from_dict(data) classmethod

Create SelectSpec from dictionary (YAML/JSON friendly).

Supports: - kind="type_lookup": table, identity, type_column, source, target, relation, optional per-side source_* / target_* lookup overrides - kind="select": from, joins, select (all_base, strings, base/from_join dicts, or legacy expr/alias dicts), where, base_alias

Source code in graflo/filter/select.py

@classmethod
def from_dict(cls, data: dict[str, Any] | list[Any]) -> Self:
    """Create SelectSpec from dictionary (YAML/JSON friendly).

    Supports:
    - kind="type_lookup": table, identity, type_column, source, target, relation,
      optional per-side source_* / target_* lookup overrides
    - kind="select": from, joins, select (``all_base``, strings, base/from_join
      dicts, or legacy expr/alias dicts), where, base_alias
    """
    if isinstance(data, list):
        return cls.model_validate(data)
    data = dict(data)
    kind = data.pop("kind", "select")
    if kind == "type_lookup":
        return cls(
            kind="type_lookup",
            **{k: v for k, v in data.items() if k != "from" and v is not None},
        )
    # Normalize "from" -> from_
    if "from" in data:
        data["from_"] = data.pop("from")
    return cls(kind="select", **data)

where_expression()

The where clause as a filter expression, or None.

Source code in graflo/filter/select.py

def where_expression(self) -> FilterExpression | None:
    """The ``where`` clause as a filter expression, or ``None``."""
    if not self.where:
        return None
    return parse_filter_expression(self.where)

Functions:

parse_filter_expression(raw)

Parse YAML/JSON/dict/list/filter model into a :class:FilterExpression.

Uses :meth:FilterExpression.from_dict for logical shorthand (OR, AND, NOT, IF_THEN keys) and leaf shorthand; uses model_validate for discriminated kind / operator + deps payloads.

Source code in graflo/filter/onto.py

def parse_filter_expression(raw: Any) -> FilterExpression:
    """Parse YAML/JSON/dict/list/filter model into a :class:`FilterExpression`.

    Uses :meth:`FilterExpression.from_dict` for logical shorthand (``OR``, ``AND``,
    ``NOT``, ``IF_THEN`` keys) and leaf shorthand; uses ``model_validate`` for
    discriminated ``kind`` / ``operator`` + ``deps`` payloads.
    """
    if isinstance(raw, FilterExpression):
        return raw
    if isinstance(raw, dict):
        if raw.get("kind") is not None:
            return FilterExpression.model_validate(raw)
        keys = list(raw.keys())
        if len(keys) == 1:
            norm_k = keys[0].upper() if isinstance(keys[0], str) else keys[0]
            if norm_k in _LOGICAL_OPERATOR_JSON_VALUES:
                return FilterExpression.from_dict(raw)
        if "operator" in raw or "deps" in raw:
            return FilterExpression.model_validate(raw)
        return FilterExpression.from_dict(raw)
    if isinstance(raw, list):
        return FilterExpression.from_dict(raw)
    raise ValueError(
        f"expected FilterExpression, dict, or list, got {type(raw).__name__}"
    )

render_conjunct(expr, *, kind, doc_name='doc', **kwargs)

Render expr as one operand of an AND of conditions.

Callers that collect several conditions and join them with AND must render each through this function. Every target language binds AND tighter than OR, so a bare a OR b beside c would mean a OR (b AND c); an OR condition is therefore parenthesised. AND and NOT conditions and single comparisons keep their meaning unwrapped, and IF_THEN renders its own parentheses.

Source code in graflo/filter/onto.py

def render_conjunct(
    expr: FilterExpression,
    *,
    kind: ExpressionFlavor,
    doc_name: str = "doc",
    **kwargs: Any,
) -> str:
    """Render *expr* as one operand of an ``AND`` of conditions.

    Callers that collect several conditions and join them with ``AND`` must
    render each through this function. Every target language binds ``AND``
    tighter than ``OR``, so a bare ``a OR b`` beside ``c`` would mean
    ``a OR (b AND c)``; an ``OR`` condition is therefore parenthesised. ``AND``
    and ``NOT`` conditions and single comparisons keep their meaning unwrapped,
    and ``IF_THEN`` renders its own parentheses.
    """
    rendered = str(expr(doc_name=doc_name, kind=kind, **kwargs))
    if expr.kind == "composite" and expr.operator == LogicalOperator.OR:
        return f"({rendered})"
    return rendered