graflo.filter.select

Declarative SQL-like view specifications.

Provides SelectSpec for describing queries in a structured way, similar to FilterExpression. Supports type_lookup shorthand for edge tables where source/target types come from lookup table(s) via FK joins, with optional per-side table / join key / discriminator column. Asymmetric edges (e.g. one static vertex type) use kind="select".

Attributes

ALL_BASE_COLUMNS = 'all_base' module-attribute

Classes

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

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 _select_columns(self) -> list[_SelectColumn]:
        """The ``select`` items, each column once, in declared order.

        The list stands for a set of named columns: an item repeated is one
        column, and a base column under its own name is already in ``all_base``.
        """
        all_base = ALL_BASE_COLUMNS in self.select
        columns: list[_SelectColumn] = []
        for item in self.select:
            column = _select_column(item)
            if column in columns or (all_base and column.is_base_column):
                continue
            columns.append(column)
        return columns

    def source_column(self, name: str) -> str | None:
        """The column the view's output *name* reads, as a filter field.

        A base column is returned bare (conditions qualify it with the base
        alias), a joined one as ``alias.column``. ``None`` when *name* is not a
        plain column of the output: an expression, or a name the view does
        not resolve.
        """
        if self.kind == "type_lookup":
            return self._type_lookup_source_column(name)
        columns = self._select_columns()
        for column in columns:
            if column.name != name:
                continue
            kind, *ref = column.source
            if kind == "base":
                return ref[0]
            if kind == "join":
                return f"{ref[0]}.{ref[1]}"
            return None
        all_base = any(column.source[0] == "all_base" for column in columns)
        unnamed = any(
            column.name is None and column.source[0] == "sql" for column in columns
        )
        return name if all_base and not unnamed else None

    def _type_lookup_source_column(self, name: str) -> str | None:
        source_type = (
            self.source_type_column
            if self.source_type_column is not None
            else self.type_column
        )
        target_type = (
            self.target_type_column
            if self.target_type_column is not None
            else self.type_column
        )
        outputs = {
            "source_id": self.source,
            "target_id": self.target,
            "source_type": f"s.{source_type}" if source_type else None,
            "target_type": f"t.{target_type}" if target_type else None,
            "relation": self.relation,
        }
        return outputs.get(name)

    @model_validator(mode="after")
    def _validate_select_names(self) -> Self:
        """Two different columns may not take one name."""
        if self.kind != "select":
            return self
        taken: dict[str, _SelectColumn] = {}
        for column in self._select_columns():
            if column.name is None:
                continue
            other = taken.setdefault(column.name, column)
            if other != column:
                raise ValueError(
                    f"select: two columns are named {column.name!r}; "
                    "give one of them another name with 'as'"
                )
        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 = [
            _render_select_column(column, base_alias, join_aliases)
            for column in self._select_columns()
        ]
        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)

source_column(name)

The column the view's output name reads, as a filter field.

A base column is returned bare (conditions qualify it with the base alias), a joined one as alias.column. None when name is not a plain column of the output: an expression, or a name the view does not resolve.

Source code in graflo/filter/select.py

def source_column(self, name: str) -> str | None:
    """The column the view's output *name* reads, as a filter field.

    A base column is returned bare (conditions qualify it with the base
    alias), a joined one as ``alias.column``. ``None`` when *name* is not a
    plain column of the output: an expression, or a name the view does
    not resolve.
    """
    if self.kind == "type_lookup":
        return self._type_lookup_source_column(name)
    columns = self._select_columns()
    for column in columns:
        if column.name != name:
            continue
        kind, *ref = column.source
        if kind == "base":
            return ref[0]
        if kind == "join":
            return f"{ref[0]}.{ref[1]}"
        return None
    all_base = any(column.source[0] == "all_base" for column in columns)
    unnamed = any(
        column.name is None and column.source[0] == "sql" for column in columns
    )
    return name if all_base and not unnamed else None

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: