Model and Query Objects

The objects you build, pass around, and get back: the model itself, the query against it, and the errors when something does not line up.

Semantic Model

orionbelt.models.semantic.SemanticModel

Bases: BaseModel

Complete semantic model parsed from OBML YAML.

Source code in src/orionbelt/models/semantic.py

class SemanticModel(BaseModel):
    """Complete semantic model parsed from OBML YAML."""

    version: float = 1.0
    name: str | None = Field(
        default=None,
        description=(
            "Optional addressing identifier for multi-model mode (v2.4.0+). "
            "When unset, the multi-model loader uses the filename stem. "
            "After normalization (lowercase + spaces/dots/dashes → "
            "underscores + trim) must match ``^[a-z][a-z0-9_]{0,62}$``. "
            "BI tools select this model via the Flight `database` catalog "
            "or pgwire `database=` URL parameter."
        ),
    )
    description: str | None = None
    settings: ModelSettings | None = None
    data_objects: dict[str, DataObject] = Field(default={}, alias="dataObjects")
    dimensions: dict[str, Dimension] = {}
    measures: dict[str, Measure] = {}
    metrics: dict[str, Metric] = {}
    filters: list[ModelFilter] = Field(default_factory=list)
    rules: dict[str, Rule] = {}
    examples: list[ModelExample] = Field(default_factory=list)
    extends_sources: list[str] = Field(default_factory=list)
    inherits_source: str | None = None
    owner: str | None = None
    expose_counts: bool = Field(
        True,
        alias="exposeCounts",
        description=(
            "When true (default), synthesize a row-count measure for every countable "
            "data object. Set false to suppress all synthesized counts (e.g. wide "
            "models where N facts would balloon the measure list). Declared measures "
            "are unaffected."
        ),
    )
    count_label_pattern: str = Field(
        "{object} Count",
        alias="countLabelPattern",
        description=(
            "Name/label template for synthesized count measures (the count's id is "
            "its label). The only valid token is ``{object}``, which interpolates each "
            "object's display label (e.g. 'Sales' -> 'Sales Count'). A per-object "
            "``countLabel`` overrides it."
        ),
    )
    custom_extensions: list[CustomExtension] = Field(default_factory=list, alias="customExtensions")
    ontology: OntologyConfig | None = Field(
        default=None,
        description=(
            "Ontology settings: the prefixes that compact concept IRIs in "
            "externalConceptMappings expand with."
        ),
    )
    external_concept_mappings: list[ExternalConceptMapping] = Field(
        default_factory=list, alias="externalConceptMappings"
    )

    model_config = {"populate_by_name": True, "extra": "forbid"}

    @field_validator("count_label_pattern", mode="before")
    @classmethod
    def _validate_count_label_pattern(cls, v: object) -> object:
        """The pattern may reference only the ``{object}`` token.

        Reject any other field access (``{name}``, ``{object.__class__}``, ...) —
        cheap insurance even though OBML is author-controlled. Bare/escaped braces
        and positional ``{}`` are rejected too; only the named ``{object}`` field
        is allowed. Delegates to the shared ``count_pattern_error`` so the OBML
        resolver reports the same rule as a structured error.
        """
        from orionbelt.models.synthesis import count_pattern_error

        if not isinstance(v, str):
            return v
        msg = count_pattern_error(v)
        if msg is not None:
            raise ValueError(msg)
        return v

    def effective_joins(
        self,
        object_name: str,
        path_overrides: dict[tuple[str, str], str] | None = None,
    ) -> list[DataObjectJoin]:
        """The joins of *object_name* that are active under *path_overrides*.

        One named secondary path per ``(source, target)`` pair replaces that
        pair's primary join; every other pair keeps its primary. This is the
        rule :class:`~orionbelt.compiler.graph.JoinGraph` traverses by, and
        anything deriving join columns has to use the same one. Filtering
        secondary joins out unconditionally instead read a conformed subquery's
        key off the primary join even when the query had asked for the secondary
        path, which silently answered at the wrong grain.

        Takes a plain mapping rather than the query's ``usePathNames`` because
        ``models.query`` imports this module, not the other way round.
        """
        overrides = path_overrides or {}
        active: list[DataObjectJoin] = []
        obj = self.data_objects.get(object_name)
        for join in obj.joins if obj else []:
            pair = (object_name, join.join_to)
            if join.secondary:
                if overrides.get(pair) == join.path_name:
                    active.append(join)
            elif pair not in overrides:
                active.append(join)
        return active

    def column_reference_objects(self, object_name: str, column_label: str) -> set[str]:
        """Other data objects a column's ``expression`` reads.

        A computed column names a sibling with ``{Column}`` and a column of
        another data object with the qualified ``{[Data Object].[Column]}``
        form. Reading the latter means joining that object in, so callers add
        what this returns to a query's join requirements — it is deliberately
        *not* part of ``measure_source_objects``, which drives multi-fact
        detection: a cross-object computed column is one row of a star, not a
        second fact.

        Follows nested computed columns, across objects as well as within one,
        and returns every object involved except the owning one — which stays
        out however many hops away the walk reaches it again, because it is
        joined already by virtue of owning the column.

        Empty for a plain column and for a computed one that reads only
        siblings, which is what makes this cheap to call on every column.
        """
        found: set[str] = set()
        seen: set[tuple[str, str]] = set()

        def walk(obj_name: str, col_label: str) -> None:
            key = (obj_name, col_label)
            if key in seen:
                return
            seen.add(key)
            obj = self.data_objects.get(obj_name)
            column = obj.columns.get(col_label) if obj else None
            if obj is None or column is None or not column.expression:
                return
            for sibling in find_placeholders(column.expression):
                if sibling in obj.columns:
                    walk(obj_name, sibling)
            for ref_object, ref_column in find_qualified_refs(column.expression):
                if ref_object not in self.data_objects:
                    continue
                if ref_object != object_name:
                    found.add(ref_object)
                walk(ref_object, ref_column)

        walk(object_name, column_label)
        return found

    def measure_join_objects(self, name: str) -> set[str]:
        """Objects a measure needs *joined* without sourcing values from them.

        Two kinds. A ``withinGroup`` column becomes the aggregate's ORDER BY,
        so it has to resolve while contributing no value. And any column the
        measure touches — its own, its filters', its filter context's — may be
        computed from another data object, which the expression names directly
        and so has to be joined.

        Deliberately separate from the objects a measure *sources*: that set
        drives multi-fact detection, and neither an ordering column nor an
        expression's neighbour is a second fact.

        Lives on the model because two callers need the same answer — the
        planner, which adds these to a query's join requirements, and
        composability, which must not advertise a measure the planner will
        then refuse. Computed independently, they drifted.
        """
        measure = self.effective_measures.get(name)
        if measure is None:
            return set()

        result: set[str] = set()
        columns: list[tuple[str, str]] = [
            (c.view, c.column) for c in measure.columns if c.view and c.column
        ]

        within = measure.within_group.column if measure.within_group is not None else None
        if within is not None and within.view:
            result.add(within.view)
            if within.column:
                columns.append((within.view, within.column))

        def collect(item: MeasureFilterItem) -> None:
            if isinstance(item, MeasureFilter):
                if item.column and item.column.view and item.column.column:
                    columns.append((item.column.view, item.column.column))
            elif isinstance(item, MeasureFilterGroup):
                for child in item.filters:
                    collect(child)

        for fi in measure.filters:
            collect(fi)

        # filterContext.include is resolved by a wrapper that runs after join
        # planning, over the joins planning chose — so its dependencies have to
        # be known here. Both field forms the wrapper accepts.
        if measure.filter_context is not None:
            for incl in measure.filter_context.include:
                dim = self.dimensions.get(incl.field)
                if dim is not None:
                    if dim.view and dim.column:
                        result.add(dim.view)
                        columns.append((dim.view, dim.column))
                    continue
                obj_name, _, col_name = incl.field.partition(".")
                obj_name, col_name = obj_name.strip(), col_name.strip()
                obj = self.data_objects.get(obj_name)
                if obj is not None and col_name in obj.columns:
                    result.add(obj_name)
                    columns.append((obj_name, col_name))

        if measure.expression:
            columns.extend(find_qualified_refs(measure.expression))

        for object_name, column_label in columns:
            result |= self.column_reference_objects(object_name, column_label)
        return result

    def dimension_join_objects(self, name: str) -> set[str]:
        """Objects a dimension needs joined besides the one it belongs to.

        Its column may be computed from another data object's column, which is
        inlined wherever the dimension is projected or filtered.
        """
        dim = self.dimensions.get(name)
        if dim is None or not dim.view or not dim.column:
            return set()
        return self.column_reference_objects(dim.view, dim.column)

    def common_join_targets(
        self,
        objects: list[str],
        path_overrides: dict[tuple[str, str], str] | None = None,
    ) -> list[str]:
        """Every data object all of *objects* join to directly, in name order.

        The candidates for conforming independent facts to a shared grain. More
        than one is an ambiguity rather than a tie to break: two facts sharing
        both a calendar and a store conform to different numbers depending which
        is used, so callers refuse rather than pick.

        Which joins count depends on the query's active ``usePathNames``: see
        :meth:`effective_joins`.
        """
        reachable_targets = {
            name: {join.join_to for join in self.effective_joins(name, path_overrides)}
            for name in objects
        }
        # A candidate has to be reached by every object, and counts as reaching
        # itself: an expression may read a column of the very object the facts
        # conform to (``Sales.Qty * Returns.Qty * Calendar.Factor``), and
        # intersecting Calendar's own join targets in would leave nothing, since
        # a conformed dimension typically joins to nothing.
        pool = set(objects) | {
            target for targets in reachable_targets.values() for target in targets
        }
        return sorted(
            candidate
            for candidate in pool
            if all(name == candidate or candidate in reachable_targets[name] for name in objects)
        )

    def unnest_root(self, name: str) -> str:
        """The nearest ancestor of *name* that a FROM clause can actually name.

        A ``nestedIn`` object's rows are an array column on its parent, so they
        exist only inside the parent's row: it can never be a query's base
        object nor a CFL leg's root, and it is never what a plan selects *from*.
        Walking up gives the object that can be, and it always covers what the
        nested one does, since the parent reaches the child and not the reverse.

        Returns *name* unchanged for an ordinary object, and stops on a chain
        that loops - the semantic validator refuses one, but this is reached
        from the compiler, which must terminate on any model it is handed.
        """
        seen: set[str] = set()
        cursor = name
        while cursor not in seen:
            obj = self.data_objects.get(cursor)
            if obj is None or obj.nested_in is None:
                return cursor
            seen.add(cursor)
            cursor = obj.nested_in.data_object
        return cursor

    @property
    def effective_measures(self) -> dict[str, Measure]:
        """Declared measures plus synthesized row-count measures (declared win).

        The single source of truth for the model's queryable measure namespace.
        Synthesized counts are *not* persisted on ``measures`` (they never
        roundtrip through YAML/OSI) — they are computed on demand here so every
        read/resolve surface sees them as ordinary named measures.
        """
        from orionbelt.models.synthesis import synthesize_count_measures

        scope = _effective_measures_scope.get()
        if scope is not None:
            entry = scope.get(id(self))
            if entry is not None and entry[0] is self:
                return entry[1]
        merged = dict(self.measures)
        merged.update(synthesize_count_measures(self))
        if scope is not None:
            scope[id(self)] = (self, merged)
        return merged

    def reaggregated_measure(self, name: str) -> tuple[str, Measure] | None:
        """The measure a reaggregate metric's first stage reads, with its name.

        Follows ``measure`` through nested reaggregate metrics, and a cumulative
        metric a first stage reads, down to the measure at the bottom: that is
        the fact every stage reads and the source of the values each one
        aggregates again. ``None`` when the chain ends anywhere else.
        """
        seen: set[str] = set()
        measures = self.effective_measures
        while name not in measures:
            metric = self.metrics.get(name)
            if (
                metric is None
                or metric.type not in (MetricType.REAGGREGATE, MetricType.CUMULATIVE)
                or metric.measure is None
                or name in seen
            ):
                return None
            seen.add(name)
            name = metric.measure
        return name, measures[name]

    @field_validator("name", mode="before")
    @classmethod
    def _validate_name(cls, v: str | None) -> str | None:
        """Reject invalid names early. Pydantic validators raise ValueError
        which the loader turns into a model-validation error.

        Empty / whitespace-only strings are treated as ``None`` rather than
        passed through, so an empty ``name:`` in YAML falls back to the
        filename stem at startup.
        """
        if v is None:
            return None
        if not isinstance(v, str):
            raise ValueError("name must be a string")
        if not v.strip():
            return None
        # Use the same normalization pipeline the loader uses, so an OBML
        # `name:` that's invalid surfaces during parse-time rather than
        # only at startup. The normalized value is stored on the model.
        from orionbelt.models.identifiers import (
            ModelNameError,
            normalize_model_name,
        )

        try:
            return normalize_model_name(v, source="OBML `name:` field")
        except ModelNameError as exc:
            raise ValueError(str(exc)) from None

effective_measures property

Declared measures plus synthesized row-count measures (declared win).

The single source of truth for the model's queryable measure namespace. Synthesized counts are not persisted on measures (they never roundtrip through YAML/OSI) — they are computed on demand here so every read/resolve surface sees them as ordinary named measures.

effective_joins(object_name, path_overrides=None)

The joins of object_name that are active under path_overrides.

One named secondary path per (source, target) pair replaces that pair's primary join; every other pair keeps its primary. This is the rule :class:~orionbelt.compiler.graph.JoinGraph traverses by, and anything deriving join columns has to use the same one. Filtering secondary joins out unconditionally instead read a conformed subquery's key off the primary join even when the query had asked for the secondary path, which silently answered at the wrong grain.

Takes a plain mapping rather than the query's usePathNames because models.query imports this module, not the other way round.

Source code in src/orionbelt/models/semantic.py

def effective_joins(
    self,
    object_name: str,
    path_overrides: dict[tuple[str, str], str] | None = None,
) -> list[DataObjectJoin]:
    """The joins of *object_name* that are active under *path_overrides*.

    One named secondary path per ``(source, target)`` pair replaces that
    pair's primary join; every other pair keeps its primary. This is the
    rule :class:`~orionbelt.compiler.graph.JoinGraph` traverses by, and
    anything deriving join columns has to use the same one. Filtering
    secondary joins out unconditionally instead read a conformed subquery's
    key off the primary join even when the query had asked for the secondary
    path, which silently answered at the wrong grain.

    Takes a plain mapping rather than the query's ``usePathNames`` because
    ``models.query`` imports this module, not the other way round.
    """
    overrides = path_overrides or {}
    active: list[DataObjectJoin] = []
    obj = self.data_objects.get(object_name)
    for join in obj.joins if obj else []:
        pair = (object_name, join.join_to)
        if join.secondary:
            if overrides.get(pair) == join.path_name:
                active.append(join)
        elif pair not in overrides:
            active.append(join)
    return active

column_reference_objects(object_name, column_label)

Other data objects a column's expression reads.

A computed column names a sibling with {Column} and a column of another data object with the qualified {[Data Object].[Column]} form. Reading the latter means joining that object in, so callers add what this returns to a query's join requirements — it is deliberately not part of measure_source_objects, which drives multi-fact detection: a cross-object computed column is one row of a star, not a second fact.

Follows nested computed columns, across objects as well as within one, and returns every object involved except the owning one — which stays out however many hops away the walk reaches it again, because it is joined already by virtue of owning the column.

Empty for a plain column and for a computed one that reads only siblings, which is what makes this cheap to call on every column.

Source code in src/orionbelt/models/semantic.py

def column_reference_objects(self, object_name: str, column_label: str) -> set[str]:
    """Other data objects a column's ``expression`` reads.

    A computed column names a sibling with ``{Column}`` and a column of
    another data object with the qualified ``{[Data Object].[Column]}``
    form. Reading the latter means joining that object in, so callers add
    what this returns to a query's join requirements — it is deliberately
    *not* part of ``measure_source_objects``, which drives multi-fact
    detection: a cross-object computed column is one row of a star, not a
    second fact.

    Follows nested computed columns, across objects as well as within one,
    and returns every object involved except the owning one — which stays
    out however many hops away the walk reaches it again, because it is
    joined already by virtue of owning the column.

    Empty for a plain column and for a computed one that reads only
    siblings, which is what makes this cheap to call on every column.
    """
    found: set[str] = set()
    seen: set[tuple[str, str]] = set()

    def walk(obj_name: str, col_label: str) -> None:
        key = (obj_name, col_label)
        if key in seen:
            return
        seen.add(key)
        obj = self.data_objects.get(obj_name)
        column = obj.columns.get(col_label) if obj else None
        if obj is None or column is None or not column.expression:
            return
        for sibling in find_placeholders(column.expression):
            if sibling in obj.columns:
                walk(obj_name, sibling)
        for ref_object, ref_column in find_qualified_refs(column.expression):
            if ref_object not in self.data_objects:
                continue
            if ref_object != object_name:
                found.add(ref_object)
            walk(ref_object, ref_column)

    walk(object_name, column_label)
    return found

measure_join_objects(name)

Objects a measure needs joined without sourcing values from them.

Two kinds. A withinGroup column becomes the aggregate's ORDER BY, so it has to resolve while contributing no value. And any column the measure touches — its own, its filters', its filter context's — may be computed from another data object, which the expression names directly and so has to be joined.

Deliberately separate from the objects a measure sources: that set drives multi-fact detection, and neither an ordering column nor an expression's neighbour is a second fact.

Lives on the model because two callers need the same answer — the planner, which adds these to a query's join requirements, and composability, which must not advertise a measure the planner will then refuse. Computed independently, they drifted.

Source code in src/orionbelt/models/semantic.py

def measure_join_objects(self, name: str) -> set[str]:
    """Objects a measure needs *joined* without sourcing values from them.

    Two kinds. A ``withinGroup`` column becomes the aggregate's ORDER BY,
    so it has to resolve while contributing no value. And any column the
    measure touches — its own, its filters', its filter context's — may be
    computed from another data object, which the expression names directly
    and so has to be joined.

    Deliberately separate from the objects a measure *sources*: that set
    drives multi-fact detection, and neither an ordering column nor an
    expression's neighbour is a second fact.

    Lives on the model because two callers need the same answer — the
    planner, which adds these to a query's join requirements, and
    composability, which must not advertise a measure the planner will
    then refuse. Computed independently, they drifted.
    """
    measure = self.effective_measures.get(name)
    if measure is None:
        return set()

    result: set[str] = set()
    columns: list[tuple[str, str]] = [
        (c.view, c.column) for c in measure.columns if c.view and c.column
    ]

    within = measure.within_group.column if measure.within_group is not None else None
    if within is not None and within.view:
        result.add(within.view)
        if within.column:
            columns.append((within.view, within.column))

    def collect(item: MeasureFilterItem) -> None:
        if isinstance(item, MeasureFilter):
            if item.column and item.column.view and item.column.column:
                columns.append((item.column.view, item.column.column))
        elif isinstance(item, MeasureFilterGroup):
            for child in item.filters:
                collect(child)

    for fi in measure.filters:
        collect(fi)

    # filterContext.include is resolved by a wrapper that runs after join
    # planning, over the joins planning chose — so its dependencies have to
    # be known here. Both field forms the wrapper accepts.
    if measure.filter_context is not None:
        for incl in measure.filter_context.include:
            dim = self.dimensions.get(incl.field)
            if dim is not None:
                if dim.view and dim.column:
                    result.add(dim.view)
                    columns.append((dim.view, dim.column))
                continue
            obj_name, _, col_name = incl.field.partition(".")
            obj_name, col_name = obj_name.strip(), col_name.strip()
            obj = self.data_objects.get(obj_name)
            if obj is not None and col_name in obj.columns:
                result.add(obj_name)
                columns.append((obj_name, col_name))

    if measure.expression:
        columns.extend(find_qualified_refs(measure.expression))

    for object_name, column_label in columns:
        result |= self.column_reference_objects(object_name, column_label)
    return result

dimension_join_objects(name)

Objects a dimension needs joined besides the one it belongs to.

Its column may be computed from another data object's column, which is inlined wherever the dimension is projected or filtered.

Source code in src/orionbelt/models/semantic.py

def dimension_join_objects(self, name: str) -> set[str]:
    """Objects a dimension needs joined besides the one it belongs to.

    Its column may be computed from another data object's column, which is
    inlined wherever the dimension is projected or filtered.
    """
    dim = self.dimensions.get(name)
    if dim is None or not dim.view or not dim.column:
        return set()
    return self.column_reference_objects(dim.view, dim.column)

common_join_targets(objects, path_overrides=None)

Every data object all of objects join to directly, in name order.

The candidates for conforming independent facts to a shared grain. More than one is an ambiguity rather than a tie to break: two facts sharing both a calendar and a store conform to different numbers depending which is used, so callers refuse rather than pick.

Which joins count depends on the query's active usePathNames: see :meth:effective_joins.

Source code in src/orionbelt/models/semantic.py

def common_join_targets(
    self,
    objects: list[str],
    path_overrides: dict[tuple[str, str], str] | None = None,
) -> list[str]:
    """Every data object all of *objects* join to directly, in name order.

    The candidates for conforming independent facts to a shared grain. More
    than one is an ambiguity rather than a tie to break: two facts sharing
    both a calendar and a store conform to different numbers depending which
    is used, so callers refuse rather than pick.

    Which joins count depends on the query's active ``usePathNames``: see
    :meth:`effective_joins`.
    """
    reachable_targets = {
        name: {join.join_to for join in self.effective_joins(name, path_overrides)}
        for name in objects
    }
    # A candidate has to be reached by every object, and counts as reaching
    # itself: an expression may read a column of the very object the facts
    # conform to (``Sales.Qty * Returns.Qty * Calendar.Factor``), and
    # intersecting Calendar's own join targets in would leave nothing, since
    # a conformed dimension typically joins to nothing.
    pool = set(objects) | {
        target for targets in reachable_targets.values() for target in targets
    }
    return sorted(
        candidate
        for candidate in pool
        if all(name == candidate or candidate in reachable_targets[name] for name in objects)
    )

unnest_root(name)

The nearest ancestor of name that a FROM clause can actually name.

A nestedIn object's rows are an array column on its parent, so they exist only inside the parent's row: it can never be a query's base object nor a CFL leg's root, and it is never what a plan selects from. Walking up gives the object that can be, and it always covers what the nested one does, since the parent reaches the child and not the reverse.

Returns name unchanged for an ordinary object, and stops on a chain that loops - the semantic validator refuses one, but this is reached from the compiler, which must terminate on any model it is handed.

Source code in src/orionbelt/models/semantic.py

def unnest_root(self, name: str) -> str:
    """The nearest ancestor of *name* that a FROM clause can actually name.

    A ``nestedIn`` object's rows are an array column on its parent, so they
    exist only inside the parent's row: it can never be a query's base
    object nor a CFL leg's root, and it is never what a plan selects *from*.
    Walking up gives the object that can be, and it always covers what the
    nested one does, since the parent reaches the child and not the reverse.

    Returns *name* unchanged for an ordinary object, and stops on a chain
    that loops - the semantic validator refuses one, but this is reached
    from the compiler, which must terminate on any model it is handed.
    """
    seen: set[str] = set()
    cursor = name
    while cursor not in seen:
        obj = self.data_objects.get(cursor)
        if obj is None or obj.nested_in is None:
            return cursor
        seen.add(cursor)
        cursor = obj.nested_in.data_object
    return cursor

reaggregated_measure(name)

The measure a reaggregate metric's first stage reads, with its name.

Follows measure through nested reaggregate metrics, and a cumulative metric a first stage reads, down to the measure at the bottom: that is the fact every stage reads and the source of the values each one aggregates again. None when the chain ends anywhere else.

Source code in src/orionbelt/models/semantic.py

def reaggregated_measure(self, name: str) -> tuple[str, Measure] | None:
    """The measure a reaggregate metric's first stage reads, with its name.

    Follows ``measure`` through nested reaggregate metrics, and a cumulative
    metric a first stage reads, down to the measure at the bottom: that is
    the fact every stage reads and the source of the values each one
    aggregates again. ``None`` when the chain ends anywhere else.
    """
    seen: set[str] = set()
    measures = self.effective_measures
    while name not in measures:
        metric = self.metrics.get(name)
        if (
            metric is None
            or metric.type not in (MetricType.REAGGREGATE, MetricType.CUMULATIVE)
            or metric.measure is None
            or name in seen
        ):
            return None
        seen.add(name)
        name = metric.measure
    return name, measures[name]

orionbelt.models.semantic.DataObject

Bases: BaseModel

A database table or view with its columns and joins.

Source code in src/orionbelt/models/semantic.py

class DataObject(BaseModel):
    """A database table or view with its columns and joins."""

    name: str
    code: str
    database: str
    schema_name: str = Field(alias="schema")
    columns: dict[str, DataObjectColumn] = {}
    joins: list[DataObjectJoin] = []
    description: str | None = None
    comment: str | None = None
    owner: str | None = None
    countable: bool = Field(
        True,
        description=(
            "When true (default), the model synthesizes a grain-anchored row-count "
            "measure for this data object (name == label, e.g. 'Sales Count'). Set "
            "false to opt out (no count measure is added to the model's measure list)."
        ),
    )
    count_label: str | None = Field(
        None,
        alias="countLabel",
        description=(
            "Optional name/label for this object's synthesized count measure (the "
            "count's id is its label). Overrides the model-level ``countLabelPattern``. "
            "The ``{object}`` token interpolates the object's display label. Ignored "
            "when ``countable`` is false."
        ),
    )
    synonyms: list[str] = Field(default_factory=list)
    custom_extensions: list[CustomExtension] = Field(default_factory=list, alias="customExtensions")
    external_concept_mappings: list[ExternalConceptMapping] = Field(
        default_factory=list, alias="externalConceptMappings"
    )
    refresh: RefreshPolicy | None = Field(
        default=None,
        description=(
            "Optional freshness contract for the physical table this dataObject maps to. "
            "Drives result-cache TTL composition. PLAN_freshness_driven_cache.md §5."
        ),
    )
    nested_in: NestedSource | None = Field(
        default=None,
        alias="nestedIn",
        description=(
            "Take this object's rows by unnesting an array column on another object "
            "rather than from a table of its own. Declared alongside ``code`` rather "
            "than instead of it: where both are present the unnest is used and the "
            "table is the fallback, which is what makes moving a model off a "
            "hand-written flattening view incremental."
        ),
    )

    @property
    def is_nested(self) -> bool:
        """Whether this object's rows come from unnesting a parent's column."""
        return self.nested_in is not None

    @property
    def qualified_code(self) -> str:
        """Full qualified table reference: database.schema.code."""
        self.require_table_source()
        return f"{self.database}.{self.schema_name}.{self.code}"

    def require_table_source(self) -> None:
        """Raise unless this object has a table a FROM clause can name.

        A ``nestedIn`` object without ``code`` has no table: its rows are an
        array column on its parent, reached by an unnest that names the parent
        rather than by selecting from anything. The planner puts it in the FROM
        clause that way and never asks for a table, so reaching here means
        something tried to *select from* it - which is what this refuses, rather
        than falling through to an empty ``code`` and emitting
        ``FROM "" AS "Charge Labels"``.
        """
        if self.code:
            return
        source = self.nested_in
        raise UnrenderableDataObjectError(
            f"Data object '{self.name}' takes its rows by unnesting "
            f"'{source.data_object}.{source.column}', so it has no table to select "
            f"from - its rows exist only inside its parent's. Reach it through "
            f"'{source.data_object}', or declare 'code' alongside 'nestedIn' to read "
            f"a flattening view."
            if source is not None
            else f"Data object '{self.name}' has no 'code' to select from."
        )

    @model_validator(mode="after")
    def _validate_has_a_source(self) -> DataObject:
        """An object needs somewhere for its rows to come from.

        ``code`` or ``nestedIn``, and both together is the supported case rather
        than an error: an object that can be unnested *and* has a flattening
        view behind it stays queryable on an engine that cannot unnest, and
        lets a model migrate one object at a time instead of all at once.
        """
        if not self.code and self.nested_in is None:
            raise ValueError(
                f"Data object '{self.name}' declares neither 'code' nor 'nestedIn', "
                "so it has no rows. Give it a table name, or nest it in a parent "
                "object's array column."
            )
        return self

    @model_validator(mode="after")
    def _validate_count_label(self) -> DataObject:
        """``countLabel`` only has an effect when the object is countable.

        Warn (do not error) so a stray override on a non-countable object is
        surfaced without breaking model load.
        """
        if self.count_label is not None and not self.countable:
            import warnings

            warnings.warn(
                f"Data object '{self.name}' sets 'countLabel' but 'countable' is false; "
                "the label is ignored because no count measure is synthesized.",
                stacklevel=2,
            )
        return self

    model_config = {"populate_by_name": True, "extra": "forbid"}

is_nested property

Whether this object's rows come from unnesting a parent's column.

qualified_code property

Full qualified table reference: database.schema.code.

require_table_source()

Raise unless this object has a table a FROM clause can name.

A nestedIn object without code has no table: its rows are an array column on its parent, reached by an unnest that names the parent rather than by selecting from anything. The planner puts it in the FROM clause that way and never asks for a table, so reaching here means something tried to select from it - which is what this refuses, rather than falling through to an empty code and emitting FROM "" AS "Charge Labels".

Source code in src/orionbelt/models/semantic.py

def require_table_source(self) -> None:
    """Raise unless this object has a table a FROM clause can name.

    A ``nestedIn`` object without ``code`` has no table: its rows are an
    array column on its parent, reached by an unnest that names the parent
    rather than by selecting from anything. The planner puts it in the FROM
    clause that way and never asks for a table, so reaching here means
    something tried to *select from* it - which is what this refuses, rather
    than falling through to an empty ``code`` and emitting
    ``FROM "" AS "Charge Labels"``.
    """
    if self.code:
        return
    source = self.nested_in
    raise UnrenderableDataObjectError(
        f"Data object '{self.name}' takes its rows by unnesting "
        f"'{source.data_object}.{source.column}', so it has no table to select "
        f"from - its rows exist only inside its parent's. Reach it through "
        f"'{source.data_object}', or declare 'code' alongside 'nestedIn' to read "
        f"a flattening view."
        if source is not None
        else f"Data object '{self.name}' has no 'code' to select from."
    )

orionbelt.models.semantic.Dimension

Bases: BaseModel

A named dimension referencing a data object column.

Source code in src/orionbelt/models/semantic.py

class Dimension(BaseModel):
    """A named dimension referencing a data object column."""

    name: str
    view: str = Field(alias="dataObject")
    column: str = ""
    result_type: DataType = Field(DataType.STRING, alias="resultType")
    time_grain: TimeGrain | None = Field(None, alias="timeGrain")
    description: str | None = None
    format: str | None = None
    via: str | None = None
    path_name: str | None = Field(None, alias="pathName")
    """The ``pathName`` of the join from ``via`` to ``dataObject`` this dimension reads through.

    Pins the dimension to one role of a data object that ``via`` joins more
    than once, such as the sales and the support employee of an order. Each
    role compiles to its own aliased join of the table, so several roles can
    appear in one query, which the query-level ``usePathNames`` cannot do: it
    swaps the pair's join for every dimension at once.

    The join must be declared on ``via`` and target ``dataObject`` directly;
    it may be the primary join or a secondary one.
    """
    owner: str | None = None
    synonyms: list[str] = Field(default_factory=list)
    custom_extensions: list[CustomExtension] = Field(default_factory=list, alias="customExtensions")
    external_concept_mappings: list[ExternalConceptMapping] = Field(
        default_factory=list, alias="externalConceptMappings"
    )

    model_config = {"populate_by_name": True, "extra": "forbid"}

path_name = Field(None, alias='pathName') class-attribute instance-attribute

The pathName of the join from via to dataObject this dimension reads through.

Pins the dimension to one role of a data object that via joins more than once, such as the sales and the support employee of an order. Each role compiles to its own aliased join of the table, so several roles can appear in one query, which the query-level usePathNames cannot do: it swaps the pair's join for every dimension at once.

The join must be declared on via and target dataObject directly; it may be the primary join or a secondary one.

orionbelt.models.semantic.Measure

Bases: BaseModel

An aggregation measure with optional expression template.

Source code in src/orionbelt/models/semantic.py

class Measure(BaseModel):
    """An aggregation measure with optional expression template."""

    name: str
    columns: list[DataColumnRef] = []
    result_type: DataType = Field(DataType.FLOAT, alias="resultType")
    aggregation: AggregationType
    expression: str | None = None
    distinct: bool = False
    total: bool = False
    default_value: str | int | float | bool | None = Field(None, alias="defaultValue")
    """Value to report when the aggregate has nothing to add up.

    An aggregate over no rows is NULL in standard SQL, and a filtered measure
    reaches that state routinely — the group exists, the filter matches none of
    it. Whether the answer should read as NULL or as zero is the modeller's
    call, not the engine's, and engines disagree: ClickHouse returns 0 where
    Postgres, DuckDB and the rest return NULL for an aggregate over an empty
    row set. Setting this pins the value on every dialect; leaving it unset
    keeps the SQL-standard NULL.

    Same slot as :attr:`Metric.default_value`, which does the same job for a
    window function with nothing to look at.
    """
    anchor: str | None = None
    """Data object whose grain this measure's expression is evaluated at.

    Only meaningful for an expression reading columns from *independent facts*
    — objects no single join path reaches together. Without it such a measure
    has no defined value: ``UNION ALL`` stacks the facts rather than joining
    them, so no row carries both columns. Naming an anchor says which fact's
    rows the expression runs over; the other facts are aggregated to the key
    they share with it and joined on many-to-one, so nothing fans out.

    It cannot be inferred. ``{[Returns].[Qty]} / {[Sales].[Qty]}`` is symmetric,
    and anchoring it on Returns rather than the shared calendar key changes
    ``AVG`` from 0.5 to 0.3333 (different row populations), so a wrong guess is
    a wrong number rather than an error.

    A bare data-object name, like :attr:`Dimension.via`.
    """

    grain: GrainOverride | None = None
    filter_context: FilterContext | None = Field(None, alias="filterContext")
    filters: list[MeasureFilterItem] = []
    data_type: str | None = Field(None, alias="dataType")
    description: str | None = None
    format: str | None = None
    allow_fan_out: bool = Field(False, alias="allowFanOut")
    delimiter: str | None = None
    within_group: WithinGroup | None = Field(None, alias="withinGroup")
    percentile: float | None = None
    """The fraction of ``percentile_cont`` / ``percentile_disc``, between 0 and
    1 (0.9 for the 90th percentile)."""
    owner: str | None = None
    synonyms: list[str] = Field(default_factory=list)
    custom_extensions: list[CustomExtension] = Field(default_factory=list, alias="customExtensions")
    external_concept_mappings: list[ExternalConceptMapping] = Field(
        default_factory=list, alias="externalConceptMappings"
    )

    model_config = {"populate_by_name": True, "extra": "forbid"}

    @field_validator("aggregation", mode="before")
    @classmethod
    def _normalize_aggregation(cls, v: object) -> object:
        """Lowercase aggregation names so ``SUM`` / ``Sum`` / ``sum`` all
        resolve to the same ``AggregationType.SUM``. The enum's canonical
        spelling is lowercase, but uppercase SQL-style is a common BI/LLM
        convention that pre-v2.7.5 worked by accident (``aggregation``
        was a plain ``str``) — keep accepting it now that the field is
        a validated enum.

        ``AGG`` and ``AGGREGATE`` are accepted as aliases for ``MEASURE``
        (v2.7.7+) so OBML reads naturally for users coming from
        Databricks (``measure``), older Spark docs (``aggregate``), or
        the shorthand most BI tools default to (``agg``).
        """
        if isinstance(v, str):
            lowered = v.lower()
            if lowered in ("agg", "aggregate"):
                return "measure"
            return lowered
        return v

    @field_validator("data_type", mode="before")
    @classmethod
    def _validate_data_type(cls, v: str | None) -> str | None:
        if v is not None:
            parse_data_type(v)
        return v

    @property
    def source_objects(self) -> set[str]:
        """Data objects whose columns this measure reads.

        Covers both declaration forms: the structured ``columns:`` list and
        ``{[Object].[Column]}`` references inside ``expression:``.
        """
        return set(self.referenced_objects)

    @property
    def referenced_objects(self) -> list[str]:
        """:attr:`source_objects`, in the order the declaration mentions them.

        Ordered for determinism, not for meaning. An earlier design used the
        first entry as the default :attr:`anchor` and was removed: it made a
        commutative rewrite change the answer, since
        ``{[Sales].[Qty]} * {[Returns].[Qty]}`` and the operands swapped would
        anchor on different facts and return different averages. Nothing reads
        position now, and nothing should.
        """
        ordered: list[str] = []
        for cref in self.columns:
            if cref.view and cref.view not in ordered:
                ordered.append(cref.view)
        if self.expression:
            for obj, _col in find_qualified_refs(self.expression):
                if obj not in ordered:
                    ordered.append(obj)
        return ordered

    @model_validator(mode="after")
    def _validate_total_grain_exclusion(self) -> Measure:
        if self.total and self.grain is not None:
            raise ValueError("'total: true' and 'grain' are mutually exclusive")
        return self

    @model_validator(mode="after")
    def _validate_measure_delegation(self) -> Measure:
        """``aggregation: measure`` delegates the aggregation to the
        engine's metric-view resolver, so the OBML measure declaration
        must NOT specify ``columns:`` or ``expression:`` — there is no
        source column for OBSL to read; the engine resolves the measure
        by name. Reject the combination at model-load time rather than
        emitting SQL that would silently ignore the column reference.
        """
        if self.aggregation == AggregationType.MEASURE:
            if self.columns:
                raise ValueError(
                    "aggregation: measure delegates resolution to the engine "
                    "(Databricks Metric View); 'columns:' must be omitted. "
                    "The engine resolves the measure by its OBML label."
                )
            if self.expression is not None:
                raise ValueError(
                    "aggregation: measure delegates resolution to the engine "
                    "(Databricks Metric View); 'expression:' must be omitted. "
                    "The engine resolves the measure by its OBML label."
                )
            if self.filters:
                raise ValueError(
                    "aggregation: measure delegates resolution to the engine "
                    "(Databricks Metric View); 'filters:' is not applicable. "
                    "Define the filter inside the metric view itself."
                )
            if self.total:
                raise ValueError(
                    "aggregation: measure cannot be combined with 'total: true' "
                    "(OBSL cannot wrap the engine-resolved aggregation in "
                    "a window function — define the total at the metric-view level)."
                )
            if self.default_value is not None:
                raise ValueError(
                    "aggregation: measure cannot be combined with 'defaultValue' "
                    "(the engine resolves the aggregation, so OBSL never sees "
                    "the empty set it would substitute for — declare the "
                    "default in the metric view itself)."
                )
        return self

    @model_validator(mode="after")
    def _validate_statistical_aggregation_arity(self) -> Measure:
        """Reject malformed statistical aggregates at model-load time.

        Two-column aggregates (``corr``, ``covar_*``, ``regr_*``) require
        exactly two entries in ``columns``. Single-column statistical
        aggregates (``stddev``, ``stddev_pop``, ``variance``, ``var_pop``)
        require exactly one.

        ``expression:`` form is **not allowed** for two-column
        aggregates — a single expression string collapses to one scalar
        argument, producing invalid SQL like ``CORR((a + b))`` instead
        of ``CORR(a, b)``. To express per-argument transformations on
        two-column aggregates, define the inputs as computed columns on
        the data object and reference them via ``columns:``.

        Single-column statistical aggregates (``stddev`` etc.) DO accept
        ``expression:`` — the result ``STDDEV(<scalar expression>)`` is
        valid SQL.
        """
        agg = self.aggregation.lower()
        if self.expression is not None:
            if agg in TWO_COLUMN_AGGREGATIONS:
                raise ValueError(
                    f"Aggregation '{agg}' requires exactly 2 columns and cannot be "
                    "combined with 'expression:'. Use the 'columns:' list with two "
                    "entries (define computed columns on the data object if you need "
                    "per-argument transformations) so the aggregate's argument order "
                    "is explicit."
                )
            return self
        if agg in TWO_COLUMN_AGGREGATIONS and len(self.columns) != 2:
            raise ValueError(
                f"Aggregation '{agg}' requires exactly 2 columns, got {len(self.columns)}"
            )
        if agg in SINGLE_COLUMN_STATISTICAL_AGGREGATIONS and len(self.columns) != 1:
            raise ValueError(
                f"Aggregation '{agg}' requires exactly 1 column, got {len(self.columns)}"
            )
        return self

    @model_validator(mode="after")
    def _validate_ordered_value(self) -> Measure:
        """``first`` / ``last`` read one value per row at a key they require.

        The direction is the aggregation's own (``last`` is the greatest key),
        so a ``withinGroup.order`` would contradict it or say nothing.
        """
        agg = self.aggregation.lower()
        if agg not in ORDERED_VALUE_AGGREGATIONS:
            return self
        if self.within_group is None:
            raise ValueError(
                f"Aggregation '{agg}' requires 'withinGroup' with the column to order by"
            )
        if "order" in self.within_group.model_fields_set:
            raise ValueError(
                f"Aggregation '{agg}' takes no 'withinGroup.order': 'last' is the value at "
                "the greatest key, 'first' at the least"
            )
        if self.expression is None and len(self.columns) != 1:
            raise ValueError(
                f"Aggregation '{agg}' requires exactly 1 column, got {len(self.columns)}"
            )
        if self.distinct:
            raise ValueError(f"Aggregation '{agg}' does not take 'distinct'")
        return self

    @model_validator(mode="after")
    def _validate_percentile(self) -> Measure:
        """A percentile orders one value per row, at a fraction it requires."""
        agg = self.aggregation.lower()
        check_percentile(agg, self.percentile)
        if agg in PERCENTILE_AGGREGATIONS:
            if self.expression is None and len(self.columns) != 1:
                raise ValueError(
                    f"Aggregation '{agg}' requires exactly 1 column, got {len(self.columns)}"
                )
            if self.distinct:
                raise ValueError(f"Aggregation '{agg}' does not take 'distinct'")
        return self

default_value = Field(None, alias='defaultValue') class-attribute instance-attribute

Value to report when the aggregate has nothing to add up.

An aggregate over no rows is NULL in standard SQL, and a filtered measure reaches that state routinely — the group exists, the filter matches none of it. Whether the answer should read as NULL or as zero is the modeller's call, not the engine's, and engines disagree: ClickHouse returns 0 where Postgres, DuckDB and the rest return NULL for an aggregate over an empty row set. Setting this pins the value on every dialect; leaving it unset keeps the SQL-standard NULL.

Same slot as :attr:Metric.default_value, which does the same job for a window function with nothing to look at.

anchor = None class-attribute instance-attribute

Data object whose grain this measure's expression is evaluated at.

Only meaningful for an expression reading columns from independent facts — objects no single join path reaches together. Without it such a measure has no defined value: UNION ALL stacks the facts rather than joining them, so no row carries both columns. Naming an anchor says which fact's rows the expression runs over; the other facts are aggregated to the key they share with it and joined on many-to-one, so nothing fans out.

It cannot be inferred. {[Returns].[Qty]} / {[Sales].[Qty]} is symmetric, and anchoring it on Returns rather than the shared calendar key changes AVG from 0.5 to 0.3333 (different row populations), so a wrong guess is a wrong number rather than an error.

A bare data-object name, like :attr:Dimension.via.

percentile = None class-attribute instance-attribute

The fraction of percentile_cont / percentile_disc, between 0 and 1 (0.9 for the 90th percentile).

source_objects property

Data objects whose columns this measure reads.

Covers both declaration forms: the structured columns: list and {[Object].[Column]} references inside expression:.

referenced_objects property

:attr:source_objects, in the order the declaration mentions them.

Ordered for determinism, not for meaning. An earlier design used the first entry as the default :attr:anchor and was removed: it made a commutative rewrite change the answer, since {[Sales].[Qty]} * {[Returns].[Qty]} and the operands swapped would anchor on different facts and return different averages. Nothing reads position now, and nothing should.

orionbelt.models.semantic.Metric

Bases: BaseModel

A metric: derived expression, cumulative window, or period-over-period comparison.

Derived (default): references measures by name using {[Measure Name]} syntax. Cumulative: applies a window function to an existing measure, ordered by a time dimension. Supports running totals, rolling windows, and grain-to-date resets. Period-over-Period: compares a measure's value against a prior time period using a synthetical date spine. Supports ratio, difference, previous value, and percent change. Reaggregate: aggregates a measure at the query grain plus the per dimensions, then aggregates those values again to the query grain (e.g. average revenue per customer, by country).

Source code in src/orionbelt/models/semantic.py

class Metric(BaseModel):
    """A metric: derived expression, cumulative window, or period-over-period comparison.

    **Derived** (default): references measures by name using ``{[Measure Name]}`` syntax.
    **Cumulative**: applies a window function to an existing measure, ordered by a time
    dimension.  Supports running totals, rolling windows, and grain-to-date resets.
    **Period-over-Period**: compares a measure's value against a prior time period using
    a synthetical date spine.  Supports ratio, difference, previous value, and percent change.
    **Reaggregate**: aggregates a measure at the query grain plus the ``per`` dimensions,
    then aggregates those values again to the query grain (e.g. average revenue per
    customer, by country).
    """

    name: str
    type: MetricType = MetricType.DERIVED
    # Derived metrics
    expression: str | None = None
    # Cumulative metrics
    measure: str | None = None
    time_dimension: str | None = Field(None, alias="timeDimension")
    cumulative_type: CumulativeAggType = Field(CumulativeAggType.SUM, alias="cumulativeType")
    window: int | None = None
    grain_to_date: GrainToDate | None = Field(None, alias="grainToDate")
    # Per-dimension partitioning for cumulative + window metrics. Each entry
    # must be a model dimension reachable from the measure's source object.
    partition_by: list[str] = Field(default_factory=list, alias="partitionBy")
    # Period-over-Period metrics
    period_over_period: PeriodOverPeriod | None = Field(None, alias="periodOverPeriod")
    # Window metrics (rank / lag / lead / ntile / first_value / last_value)
    window_function: WindowFunctionKind | None = Field(None, alias="windowFunction")
    offset: int | None = None
    buckets: int | None = None
    order_direction: str = Field("desc", alias="orderDirection")
    default_value: str | int | float | bool | None = Field(None, alias="defaultValue")
    # Reaggregate metrics: dimensions added to the query grain for the first
    # stage (``name`` or ``name:grain``), and the second-stage aggregation.
    per: list[str] = Field(default_factory=list)
    aggregation: ReaggregateAggType | None = None
    # Conditions on the first-stage groups; only those meeting all of them are
    # aggregated again.
    having: list[ReaggregateHaving] = Field(default_factory=list)
    # The fraction of a ``percentile_cont`` / ``percentile_disc`` second stage.
    percentile: float | None = None
    # Where the first-stage groups come from: the facts (default) or every
    # member of the ``per`` dimensions.
    population: ReaggregatePopulation | None = None
    # With population: dimension, a member with no row in a period takes its
    # latest earlier value (a semi-additive balance between snapshots).
    carry_forward: bool = Field(False, alias="carryForward")
    # Common
    data_type: str | None = Field(None, alias="dataType")
    description: str | None = None
    format: str | None = None
    owner: str | None = None
    synonyms: list[str] = Field(default_factory=list)
    custom_extensions: list[CustomExtension] = Field(default_factory=list, alias="customExtensions")
    external_concept_mappings: list[ExternalConceptMapping] = Field(
        default_factory=list, alias="externalConceptMappings"
    )

    model_config = {"populate_by_name": True, "extra": "forbid"}

    @field_validator("data_type", mode="before")
    @classmethod
    def _validate_data_type(cls, v: str | None) -> str | None:
        if v is not None:
            parse_data_type(v)
        return v

    @model_validator(mode="after")
    def _validate_metric_type(self) -> Metric:
        if self.type == MetricType.REAGGREGATE:
            self._validate_reaggregate()
            return self
        if (
            self.per
            or self.aggregation is not None
            or self.having
            or self.percentile is not None
            or self.population is not None
            or self.carry_forward
        ):
            raise ValueError(
                "'per', 'aggregation', 'having', 'percentile', 'population' and "
                "'carryForward' are only valid on reaggregate metrics"
            )
        if self.type == MetricType.DERIVED:
            if not self.expression:
                raise ValueError("Derived metrics require 'expression'")
            if self.partition_by:
                raise ValueError("Derived metrics must not have 'partitionBy'")
        elif self.type == MetricType.CUMULATIVE:
            if not self.measure:
                raise ValueError("Cumulative metrics require 'measure'")
            if not self.time_dimension:
                raise ValueError("Cumulative metrics require 'timeDimension'")
            if self.expression:
                raise ValueError("Cumulative metrics must not have 'expression'")
            if self.window is not None and self.grain_to_date is not None:
                raise ValueError("'window' and 'grainToDate' are mutually exclusive")
            if self.window is not None and self.window < 1:
                raise ValueError("'window' must be >= 1")
        elif self.type == MetricType.PERIOD_OVER_PERIOD:
            if not self.expression:
                raise ValueError("Period-over-period metrics require 'expression'")
            if not self.period_over_period:
                raise ValueError("Period-over-period metrics require 'periodOverPeriod'")
            if self.measure:
                raise ValueError(
                    "Period-over-period metrics must not have 'measure' "
                    "(use 'expression' to reference measures)"
                )
            if self.window is not None or self.grain_to_date is not None:
                raise ValueError(
                    "Period-over-period metrics must not have 'window' or 'grainToDate'"
                )
            if self.partition_by:
                raise ValueError("Period-over-period metrics must not have 'partitionBy'")
        elif self.type == MetricType.WINDOW:
            if self.window_function is None:
                raise ValueError("Window metrics require 'windowFunction'")
            if not self.measure and self.window_function not in {
                WindowFunctionKind.ROW_NUMBER,
                WindowFunctionKind.NTILE,
            }:
                # row_number / ntile can rank without an explicit measure, falling back
                # to ordering on the time dimension. All other window functions take
                # the measure as their argument or ORDER BY input.
                raise ValueError(
                    f"Window metric with function '{self.window_function.value}' requires 'measure'"
                )
            if self.expression:
                raise ValueError("Window metrics must not have 'expression'")
            if self.window is not None or self.grain_to_date is not None:
                raise ValueError("Window metrics must not have 'window' or 'grainToDate'")
            if self.window_function in {WindowFunctionKind.LAG, WindowFunctionKind.LEAD}:
                if self.offset is None or self.offset < 1:
                    raise ValueError(
                        f"Window metric with function '{self.window_function.value}' "
                        f"requires positive 'offset'"
                    )
                if not self.time_dimension:
                    raise ValueError(
                        f"Window metric with function '{self.window_function.value}' "
                        f"requires 'timeDimension'"
                    )
            if self.window_function == WindowFunctionKind.NTILE and (
                self.buckets is None or self.buckets < 2
            ):
                raise ValueError("Window metric with function 'ntile' requires 'buckets' >= 2")
            if self.order_direction.lower() not in {"asc", "desc"}:
                raise ValueError("'orderDirection' must be 'asc' or 'desc'")
        return self

    def _validate_reaggregate(self) -> None:
        if not self.measure:
            raise ValueError("Reaggregate metrics require 'measure'")
        if not self.per:
            raise ValueError("Reaggregate metrics require at least one 'per' dimension")
        if len(set(self.per)) != len(self.per):
            raise ValueError("Reaggregate metric 'per' entries must be unique")
        if self.aggregation is None:
            raise ValueError("Reaggregate metrics require 'aggregation'")
        check_percentile(self.aggregation.value, self.percentile)
        # Checked against what was supplied, not against values: a default such
        # as cumulativeType 'sum' is fine, an explicit one is a setting that
        # would have no effect.
        foreign = sorted(
            Metric.model_fields[field].alias or field
            for field in self.model_fields_set - _REAGGREGATE_FIELDS
        )
        if foreign:
            raise ValueError(f"Reaggregate metrics must not have {', '.join(foreign)}")

Query Models

orionbelt.models.query.QueryObject

Bases: BaseModel

A complete YAML analytical query.

Source code in src/orionbelt/models/query.py

class QueryObject(BaseModel):
    """A complete YAML analytical query."""

    select: QuerySelect
    where: list[QueryFilterItem] = []
    having: list[QueryFilterItem] = []
    order_by: list[QueryOrderBy] = Field([], alias="orderBy")
    limit: int | None = None
    offset: int | None = None
    use_path_names: list[UsePathName] = Field([], alias="usePathNames")
    allow_fan_out: bool = Field(False, alias="allowFanOut")
    """Silence fan-out warnings for this query.

    The query-level counterpart of ``Measure.allowFanOut``. A measure reading
    both a base-grain column and one from an object the joins replicate is
    evaluated per base row: right for an extended price
    (``quantity * list price``), wrong for anything that reads the replicated
    row's own magnitude. Nothing in the declarations separates the two, so the
    compiler warns rather than refusing, and this says the duplication is
    understood and intended for this query.
    """

    dimensions_exclude: bool = Field(False, alias="dimensionsExclude")
    grouping: Grouping | None = Field(
        default=None,
        description=(
            "Hierarchical grouping modifier. 'rollup' emits GROUP BY ROLLUP(...) "
            "for hierarchical subtotals + grand total. 'cube' emits GROUP BY CUBE(...) "
            "for the full cross-tab. Adds one GROUPING(dim) AS _g_<dim> column per "
            "selected dimension so callers can distinguish subtotal/grand-total rows."
        ),
    )
    as_of: date | None = Field(
        default=None,
        alias="asOf",
        description=(
            "The date a cumulative metric is evaluated at when the query does not select "
            "its timeDimension, read at that dimension's grain (2025-12-17 is December "
            "for a monthly one). Without it, the metric is evaluated at the latest period "
            "with data under the query's filters. Has no effect on a cumulative metric "
            "whose timeDimension is selected."
        ),
    )

    model_config = {"populate_by_name": True, "extra": "forbid"}

    @model_validator(mode="after")
    def _validate_grouping(self) -> QueryObject:
        """Reject grouping with no dimensions or in raw mode."""
        if self.grouping is None:
            return self
        if self.select.is_raw:
            raise ValueError(
                "select.fields (raw mode) cannot be combined with grouping (rollup/cube)"
            )
        if not self.select.dimensions:
            raise ValueError(
                "grouping (rollup/cube) requires at least one dimension in select.dimensions"
            )
        return self

    @model_validator(mode="after")
    def _validate_raw_mode_exclusivity(self) -> QueryObject:
        """Raw mode (``select.fields``) is mutually exclusive with aggregate
        features. Catch misuse early so the resolver can assume a clean shape.
        """
        if self.select.is_raw:
            if self.select.dimensions:
                raise ValueError(
                    "select.fields (raw mode) cannot be combined with select.dimensions"
                )
            if self.select.measures:
                raise ValueError("select.fields (raw mode) cannot be combined with select.measures")
            if self.having:
                raise ValueError("select.fields (raw mode) cannot be combined with having")
            if self.dimensions_exclude:
                raise ValueError(
                    "select.fields (raw mode) cannot be combined with dimensionsExclude"
                )
        elif self.select.distinct:
            raise ValueError("select.distinct is only valid in raw mode (with select.fields)")
        return self

allow_fan_out = Field(False, alias='allowFanOut') class-attribute instance-attribute

Silence fan-out warnings for this query.

The query-level counterpart of Measure.allowFanOut. A measure reading both a base-grain column and one from an object the joins replicate is evaluated per base row: right for an extended price (quantity * list price), wrong for anything that reads the replicated row's own magnitude. Nothing in the declarations separates the two, so the compiler warns rather than refusing, and this says the duplication is understood and intended for this query.

orionbelt.models.query.QuerySelect

Bases: BaseModel

The SELECT part of a query.

Two mutually exclusive modes:

• Aggregate mode (default): dimensions + measures produce a grouped, aggregated result (GROUP BY dimensions, aggregate measures).
• Raw mode: fields returns un-aggregated rows from one or more data objects joined per the model. Set distinct: true for SELECT DISTINCT. Raw mode rejects dimensions, measures, metrics, and HAVING.

Source code in src/orionbelt/models/query.py

class QuerySelect(BaseModel):
    """The SELECT part of a query.

    Two mutually exclusive modes:

    * **Aggregate mode** (default): ``dimensions`` + ``measures`` produce a
      grouped, aggregated result (GROUP BY dimensions, aggregate measures).
    * **Raw mode**: ``fields`` returns un-aggregated rows from one or more
      data objects joined per the model. Set ``distinct: true`` for
      ``SELECT DISTINCT``. Raw mode rejects ``dimensions``, ``measures``,
      ``metrics``, and ``HAVING``.
    """

    dimensions: list[str | CoalesceDimension] = []
    measures: list[str] = []
    fields: list[str] = []
    distinct: bool = False

    model_config = {"populate_by_name": True, "extra": "forbid"}

    @property
    def is_raw(self) -> bool:
        """True when this select is in raw mode (fields-based projection)."""
        return bool(self.fields)

is_raw property

True when this select is in raw mode (fields-based projection).

orionbelt.models.query.QueryFilter

Bases: BaseModel

A filter condition in a query.

Source code in src/orionbelt/models/query.py

class QueryFilter(BaseModel):
    """A filter condition in a query."""

    field: str
    op: FilterOperator
    value: Any = None
    subquery: Subquery | None = None

    model_config = {"populate_by_name": True, "extra": "forbid"}

    @field_validator("value", mode="before")
    @classmethod
    def _validate_filter_value(cls, v: Any) -> Any:
        """Reject arbitrary nested objects — allow scalars, lists of scalars, and dicts
        (for RELATIVE filters which use ``{unit, count, direction}`` objects).
        Date/datetime values are coerced to ISO strings.
        """
        if v is None:
            return v
        if isinstance(v, datetime):
            return v.isoformat()
        if isinstance(v, date):
            return v.isoformat()
        if isinstance(v, (str, int, float, bool, Decimal)):
            return v
        if isinstance(v, list):
            coerced = [i.isoformat() if isinstance(i, (date, datetime)) else i for i in v]
            if all(isinstance(i, (str, int, float, bool, Decimal)) for i in coerced):
                return coerced
        if isinstance(v, dict) and all(isinstance(k, str) for k in v):
            return v
        msg = "Filter value must be a scalar, list of scalars, or object"
        raise ValueError(msg)

    @model_validator(mode="after")
    def _validate_subquery_exclusivity(self) -> QueryFilter:
        """``exists`` / ``nonexists`` require ``subquery`` (and reject ``value``).

        All other operators reject ``subquery`` — the payload would be silently
        ignored, which would mask typos.
        """
        is_subquery_op = self.op in (FilterOperator.EXISTS, FilterOperator.NONEXISTS)
        if is_subquery_op:
            if self.subquery is None:
                raise ValueError(
                    f"Operator '{self.op}' requires a 'subquery' object with 'dataObject'"
                )
            if self.value is not None:
                raise ValueError(f"Operator '{self.op}' takes 'subquery', not 'value' / 'values'")
        elif self.subquery is not None:
            raise ValueError(
                f"Operator '{self.op}' does not accept 'subquery' — use 'exists' or 'nonexists'"
            )
        return self

orionbelt.models.query.UsePathName

Bases: BaseModel

Selects a named secondary join path for a specific (source, target) pair.

Source code in src/orionbelt/models/query.py

class UsePathName(BaseModel):
    """Selects a named secondary join path for a specific (source, target) pair."""

    source: str
    target: str
    path_name: str = Field(alias="pathName")

    model_config = {"populate_by_name": True, "extra": "forbid"}

orionbelt.models.query.DimensionRef

Bases: BaseModel

Reference to a dimension, optionally with time grain.

Supports notation like "customer.country" or "order.order_date:month".

Source code in src/orionbelt/models/query.py

class DimensionRef(BaseModel):
    """Reference to a dimension, optionally with time grain.

    Supports notation like "customer.country" or "order.order_date:month".
    """

    name: str
    grain: TimeGrain | None = None

    model_config = {"populate_by_name": True, "extra": "forbid"}

    @classmethod
    def parse(cls, raw: str) -> DimensionRef:
        """Parse 'name:grain' notation."""
        if ":" in raw:
            name, grain_str = raw.rsplit(":", 1)
            return cls(name=name, grain=TimeGrain(grain_str))
        return cls(name=raw)

parse(raw) classmethod

Parse 'name:grain' notation.

Source code in src/orionbelt/models/query.py

@classmethod
def parse(cls, raw: str) -> DimensionRef:
    """Parse 'name:grain' notation."""
    if ":" in raw:
        name, grain_str = raw.rsplit(":", 1)
        return cls(name=name, grain=TimeGrain(grain_str))
    return cls(name=raw)

Error Models

orionbelt.models.errors.SemanticError

Bases: BaseModel

A structured error or warning with optional source position and remediation.

Used uniformly for errors (severity="error") and warnings (severity="warning"). See models/warnings.py for the stable warning code taxonomy.

Source code in src/orionbelt/models/errors.py

class SemanticError(BaseModel):
    """A structured error or warning with optional source position and remediation.

    Used uniformly for errors (``severity="error"``) and warnings (``severity="warning"``).
    See ``models/warnings.py`` for the stable warning code taxonomy.
    """

    code: str
    message: str
    path: str | None = None
    span: SourceSpan | None = None
    suggestions: list[str] = Field(default_factory=list)
    severity: str = "error"
    hint: str | None = Field(
        default=None,
        description="Optional remediation suggestion (single sentence)",
    )
    context: dict[str, Any] | None = Field(
        default=None,
        description=(
            "Optional structured detail (e.g. which measure / dataObject / column) so "
            "agents can branch on the data without parsing the message."
        ),
    )

orionbelt.models.errors.ValidationResult

Bases: BaseModel

Result of semantic model validation.

Source code in src/orionbelt/models/errors.py

class ValidationResult(BaseModel):
    """Result of semantic model validation."""

    valid: bool
    errors: list[SemanticError] = Field(default_factory=list)
    warnings: list[SemanticError] = Field(default_factory=list)

orionbelt.models.errors.SourceSpan

Bases: BaseModel

Points to exact location in YAML source for error reporting.

Source code in src/orionbelt/models/errors.py

class SourceSpan(BaseModel):
    """Points to exact location in YAML source for error reporting."""

    file: str
    line: int
    column: int
    end_line: int | None = None
    end_column: int | None = None