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, Python).

Key Components

• LogicalOperator: Logical operations (AND, OR, NOT, IMPLICATION)
• ComparisonOperator: Comparison operations (==, !=, >, <, etc.)
• FilterExpression: Filter expression (leaf or composite logical formulae)

Example

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

Converts to: "age >= 18 AND status == 'active'"

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"

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' or 'foo' (YAML/kwargs) to unary_op; infer kind and cmp_operator."""
        if not isinstance(data, dict):
            return data
        if data.get("kind") == "composite":
            return data
        raw_op = None
        data = dict(data)
        if "operator" in data and isinstance(data["operator"], str):
            raw_op = data.pop("operator")
        elif "foo" in data and isinstance(data["foo"], str):
            raw_op = data.pop("foo")
        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")
            # 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)):
                result.append(FilterExpression.from_dict(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, current: dict[str, Any] | list[Any]) -> Self:  # type: ignore[override]
        """Create a filter expression from a dictionary or list.

        Returns FilterExpression (leaf or composite). LSP-compliant: return type is Self.
        """
        if isinstance(current, list):
            if current[0] in ComparisonOperator:
                return cls.from_list(current)  # type: ignore[return-value]
            elif current[0] in LogicalOperator:
                return cls(kind="composite", operator=current[0], deps=current[1])
        elif isinstance(current, dict):
            k = list(current.keys())[0]
            norm_k = k.upper() if isinstance(k, str) else k
            if norm_k in LogicalOperator:
                deps = [cls.from_dict(v) for v in current[k]]
                return cls(
                    kind="composite", operator=LogicalOperator(norm_k), deps=deps
                )
            else:
                unary_op = current.get("operator") or current.get("foo")
                cmp_operator = current.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=current.get("value", []),
                    field=current.get("field"),
                    unary_op=unary_op,
                )
        raise ValueError(f"expected dict or list, got {type(current)}")

    def __call__(
        self,
        doc_name="doc",
        kind: ExpressionFlavor = ExpressionFlavor.AQL,
        **kwargs,
    ) -> str | bool:
        """Render or evaluate the expression in the target language."""
        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"
            )
        if kind == ExpressionFlavor.AQL:
            return self._cast_arango(doc_name)
        elif kind == ExpressionFlavor.CYPHER:
            return self._cast_cypher(doc_name)
        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()
        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)
        elif kind == ExpressionFlavor.PYTHON:
            return self._cast_python_composite(kind=kind, **kwargs)
        raise ValueError(f"kind {kind} not implemented")

    def _cast_value(self) -> str:
        value = f"{self.value[0]}" if len(self.value) == 1 else f"{self.value}"
        if len(self.value) == 1:
            if isinstance(self.value[0], str):
                escaped = self.value[0].replace("\\", "\\\\").replace('"', '\\"')
                value = f'"{escaped}"'
            elif self.value[0] is None:
                value = "null"
            else:
                value = f"{self.value[0]}"
        return value

    def _cast_arango(self, doc_name: str) -> 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()
        lemma = f"{self.cmp_operator} {const}"
        if self.unary_op is not None:
            lemma = f"{self.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) -> 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.unary_op is not None:
            lemma = f"{self.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.unary_op is not None:
            lemma = f"{self.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.unary_op is not None:
            lemma = f"{self.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.

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

    def _cast_sql(self) -> str:
        """Render leaf as SQL WHERE fragment: \"column\" op value (strings/dates single-quoted)."""
        if not self.field:
            return ""
        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.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 value is None:
            value_str = "null"
        elif isinstance(value, (int, float)):
            value_str = str(value)
        else:
            # Strings and ISO datetimes: single-quoted for SQL
            value_str = str(value).replace("'", "''")
            value_str = f"'{value_str}'"
        return f"{quoted} {op_str} {value_str}"

    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.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):
            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, **kwargs: Any) -> bool:
        if self.field is not None:
            field_val = kwargs.pop(self.field, None)
            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 not None and self.unary_op is not None:
                foo = getattr(field_val, self.unary_op)
                return foo(self.value[0])
        return False

    def _cast_generic(self, doc_name: str, kind: ExpressionFlavor) -> str:
        if self.operator is None:
            raise ValueError("composite expression requires operator")
        if len(self.deps) == 1:
            if self.operator == LogicalOperator.NOT:
                result = self.deps[0](kind=kind, doc_name=doc_name)
                if doc_name == "" and kind == ExpressionFlavor.GSQL:
                    return f"!{result}"
                return f"{self.operator} {result}"
            raise ValueError(
                f" length of deps = {len(self.deps)} but operator is not {LogicalOperator.NOT}"
            )
        deps_str = [dep(kind=kind, doc_name=doc_name) for dep in self.deps]
        # __call__ returns str | bool; join expects str
        deps_str_cast: list[str] = [str(x) for x in deps_str]
        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, kind: ExpressionFlavor, **kwargs: 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](kind=kind, **kwargs)
            raise ValueError(
                f" length of deps = {len(self.deps)} but operator is not {LogicalOperator.NOT}"
            )
        return OperatorMapping[self.operator](
            [dep(kind=kind, **kwargs) for dep in self.deps]
        )

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

Render or evaluate the expression in the target language.

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."""
    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")
        # 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(current) 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, current: dict[str, Any] | list[Any]) -> Self:  # type: ignore[override]
    """Create a filter expression from a dictionary or list.

    Returns FilterExpression (leaf or composite). LSP-compliant: return type is Self.
    """
    if isinstance(current, list):
        if current[0] in ComparisonOperator:
            return cls.from_list(current)  # type: ignore[return-value]
        elif current[0] in LogicalOperator:
            return cls(kind="composite", operator=current[0], deps=current[1])
    elif isinstance(current, dict):
        k = list(current.keys())[0]
        norm_k = k.upper() if isinstance(k, str) else k
        if norm_k in LogicalOperator:
            deps = [cls.from_dict(v) for v in current[k]]
            return cls(
                kind="composite", operator=LogicalOperator(norm_k), deps=deps
            )
        else:
            unary_op = current.get("operator") or current.get("foo")
            cmp_operator = current.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=current.get("value", []),
                field=current.get("field"),
                unary_op=unary_op,
            )
    raise ValueError(f"expected dict or list, got {type(current)}")

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' or 'foo' (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' or 'foo' (YAML/kwargs) to unary_op; infer kind and cmp_operator."""
    if not isinstance(data, dict):
        return data
    if data.get("kind") == "composite":
        return data
    raw_op = None
    data = dict(data)
    if "operator" in data and isinstance(data["operator"], str):
        raw_op = data.pop("operator")
    elif "foo" in data and isinstance(data["foo"], str):
        raw_op = data.pop("foo")
    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

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)):
            result.append(FilterExpression.from_dict(item))
        else:
            result.append(item)
    return result

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]

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"

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] | list[dict[str, Any]]	SELECT list (used when kind="select")
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).

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")
        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).
    """

    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] | list[dict[str, Any]] = Field(default_factory=lambda: ["*"])
    where: FilterExpression | dict[str, Any] | None = None

    # 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

    @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

    def build_sql(
        self,
        schema: str,
        base_table: str,
    ) -> str:
        """Build SQL SELECT query.

        Args:
            schema: Schema name (e.g. "public")
            base_table: Base table name (from TableConnector.table_name)

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

    def _build_type_lookup_sql(self, schema: str, base_table: 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}"'

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

        if rel_col:
            select_parts.append(f'r."{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} r "
            f'LEFT JOIN {s_ref} s ON r."{src_fk}" = s."{src_ident}" '
            f'LEFT JOIN {t_ref} t ON r."{tgt_fk}" = t."{tgt_ident}"'
        )

        where_clause = f's."{src_ident}" IS NOT NULL AND t."{tgt_ident}" IS NOT NULL'
        return f"SELECT {select_clause} FROM {from_clause} WHERE {where_clause}"

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

        # SELECT
        select_parts: list[str] = []
        for item in self.select:
            if isinstance(item, str):
                select_parts.append(item)
            elif isinstance(item, dict):
                expr = item.get("expr", "")
                alias = item.get("alias")
                if alias:
                    select_parts.append(f"{expr} AS {alias}")
                else:
                    select_parts.append(expr)
        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
        if self.where:
            we = (
                FilterExpression.from_dict(self.where)
                if isinstance(self.where, dict)
                else self.where
            )
            where_str = we(kind=ExpressionFlavor.SQL)
            if where_str:
                query += f" WHERE {where_str}"

        return query

    @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, where
        """
        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)

build_sql(schema, base_table)

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

Returns:

Type	Description
str	Complete SQL query string

Source code in graflo/filter/select.py

def build_sql(
    self,
    schema: str,
    base_table: str,
) -> str:
    """Build SQL SELECT query.

    Args:
        schema: Schema name (e.g. "public")
        base_table: Base table name (from TableConnector.table_name)

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

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, where

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, where
    """
    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)