Skip to content

graflo.architecture.query

DB-agnostic read contract: what an agent may ask of a live graph.

Layer 2. Imports filter/ and graph_types/ (layer 1) and nothing from db/ — enforcement has to be expressible without a driver, or "enforced in core" would mean "enforced wherever a driver happens to be imported".

Not in contract/, which is the write-side manifest at layer 3: mixing a read contract into it would muddy a boundary the layering test already guards.

AggregateQuery

Bases: GraphQuery

Count or summarise one vertex type.

Source code in graflo/architecture/query/models.py
class AggregateQuery(GraphQuery):
    """Count or summarise one vertex type."""

    vertex_type: str = PydanticField(...)
    function: AggregationType = PydanticField(
        default=AggregationType.COUNT, description="Aggregation to apply."
    )
    group_by: str | None = PydanticField(
        default=None, description="Property to group by. COUNT only."
    )
    aggregated_field: str | None = PydanticField(
        default=None,
        description="Property to aggregate. Required for everything but COUNT.",
    )
    filters: FilterExpression | None = PydanticField(default=None)

    @property
    def aggregation(self) -> AggregationType:
        """`function` as an enum. See `NeighborQuery.edge_direction`."""
        return AggregationType(self.function)

    def _validate_specific(self, caps: QueryCaps) -> None:
        if self.function != AggregationType.COUNT and not self.aggregated_field:
            raise ValueError(
                f"aggregated_field is required for {self.function}; only COUNT "
                "can aggregate without naming a property"
            )
        if self.group_by and self.function != AggregationType.COUNT:
            raise ValueError(
                f"group_by is only supported for COUNT, not {self.function}"
            )

aggregation property

function as an enum. See NeighborQuery.edge_direction.

CapExceededError

Bases: ValueError

A query asked for more than a cap allows.

Carries the cap's name so the surface can say which limit was hit rather than returning a generic validation failure — an agent that is told "too many hops, max is 3" can retry; one told "invalid request" cannot.

Source code in graflo/architecture/query/caps.py
class CapExceededError(ValueError):
    """A query asked for more than a cap allows.

    Carries the cap's name so the surface can say *which* limit was hit rather
    than returning a generic validation failure — an agent that is told "too
    many hops, max is 3" can retry; one told "invalid request" cannot.
    """

    def __init__(self, cap: str, requested: object, allowed: object) -> None:
        self.cap = cap
        self.requested = requested
        self.allowed = allowed
        super().__init__(
            f"{cap} exceeded: requested {requested!r}, maximum is {allowed!r}"
        )

GraphQuery

Bases: ConfigBaseModel

Shared envelope for every read request.

Note

There is intentionally no caps field. Caps come from :data:HARD_CAPS and are lowered only through :meth:narrowed, so a request body cannot raise its own ceiling.

Source code in graflo/architecture/query/models.py
class GraphQuery(ConfigBaseModel):
    """Shared envelope for every read request.

    Note:
        There is intentionally no ``caps`` field. Caps come from
        :data:`HARD_CAPS` and are lowered only through :meth:`narrowed`, so a
        request body cannot raise its own ceiling.
    """

    limit: int = PydanticField(
        default=100, ge=1, description="Rows to return, within `max_rows`."
    )
    timeout_s: float = PydanticField(
        default=10.0, gt=0, description="Seconds to allow, within `timeout_s`."
    )
    projection: list[str] | None = PydanticField(
        default=None,
        description=(
            "Property names to return. None returns whatever the backend "
            "stores, subject to the cap's allow-list."
        ),
    )

    def finish_init(self, caps: QueryCaps | None = None) -> Self:
        """Validate against *caps*, defaulting to the core ceiling.

        Returns:
            Self, so the call composes: ``NodeQuery(...).finish_init()``.

        Raises:
            CapExceededError: Naming the cap that was exceeded.
        """
        effective = caps or HARD_CAPS
        if self.limit > effective.max_rows:
            raise CapExceededError("max_rows", self.limit, effective.max_rows)
        if self.timeout_s > effective.timeout_s:
            raise CapExceededError("timeout_s", self.timeout_s, effective.timeout_s)
        if self.projection is not None and effective.projection_allow_list is not None:
            permitted = set(effective.projection_allow_list)
            denied = sorted(n for n in self.projection if n not in permitted)
            if denied:
                raise CapExceededError(
                    "projection_allow_list", denied, effective.projection_allow_list
                )
        self._validate_specific(effective)
        return self

    def _validate_specific(self, caps: QueryCaps) -> None:
        """Per-query-kind checks. Overridden by subclasses."""

    def narrowed(self, caps: QueryCaps) -> Self:
        """Return a copy fitted to *caps*, raising on anything explicitly asked for.

        The split is what makes caps both enforceable and usable:

        - A value the caller **explicitly set** above a cap raises
          :class:`CapExceededError`. Silently clamping it would hand back a
          partial answer the caller believes is complete — the failure mode this
          codebase already rejected for TigerGraph edge direction.
        - A value the caller **left at its default** is clamped. A policy of
          ``max_rows=5`` must not 422 every request that simply did not mention
          a limit; that would make strict policies unusable rather than safe.

        ``model_fields_set`` is what distinguishes the two, which is precisely
        why it must be consulted rather than comparing against the default.

        Projection is always intersected rather than raising, since an
        allow-list exists to *hide* properties: refusing the request would tell
        the caller which forbidden property they guessed correctly.
        """
        effective = HARD_CAPS.narrow(caps)
        explicit = self.model_fields_set
        narrowed = self.model_copy(deep=True)

        if "limit" in explicit and self.limit > effective.max_rows:
            raise CapExceededError("max_rows", self.limit, effective.max_rows)
        narrowed.limit = min(self.limit, effective.max_rows)

        if "timeout_s" in explicit and self.timeout_s > effective.timeout_s:
            raise CapExceededError("timeout_s", self.timeout_s, effective.timeout_s)
        narrowed.timeout_s = min(self.timeout_s, effective.timeout_s)

        if (
            narrowed.projection is not None
            and effective.projection_allow_list is not None
        ):
            permitted = set(effective.projection_allow_list)
            narrowed.projection = [n for n in narrowed.projection if n in permitted]

        narrowed._narrow_specific(effective, explicit)
        return narrowed

    def _narrow_specific(self, caps: QueryCaps, explicit: set[str]) -> None:
        """Per-query-kind fitting. Overridden by subclasses.

        Args:
            caps: The effective ceiling.
            explicit: Field names the caller actually set, so an explicit
                over-ask can raise while a default is clamped.
        """

finish_init(caps=None)

Validate against caps, defaulting to the core ceiling.

Returns:

Type Description
Self

Self, so the call composes: NodeQuery(...).finish_init().

Raises:

Type Description
CapExceededError

Naming the cap that was exceeded.

Source code in graflo/architecture/query/models.py
def finish_init(self, caps: QueryCaps | None = None) -> Self:
    """Validate against *caps*, defaulting to the core ceiling.

    Returns:
        Self, so the call composes: ``NodeQuery(...).finish_init()``.

    Raises:
        CapExceededError: Naming the cap that was exceeded.
    """
    effective = caps or HARD_CAPS
    if self.limit > effective.max_rows:
        raise CapExceededError("max_rows", self.limit, effective.max_rows)
    if self.timeout_s > effective.timeout_s:
        raise CapExceededError("timeout_s", self.timeout_s, effective.timeout_s)
    if self.projection is not None and effective.projection_allow_list is not None:
        permitted = set(effective.projection_allow_list)
        denied = sorted(n for n in self.projection if n not in permitted)
        if denied:
            raise CapExceededError(
                "projection_allow_list", denied, effective.projection_allow_list
            )
    self._validate_specific(effective)
    return self

narrowed(caps)

Return a copy fitted to caps, raising on anything explicitly asked for.

The split is what makes caps both enforceable and usable:

  • A value the caller explicitly set above a cap raises :class:CapExceededError. Silently clamping it would hand back a partial answer the caller believes is complete — the failure mode this codebase already rejected for TigerGraph edge direction.
  • A value the caller left at its default is clamped. A policy of max_rows=5 must not 422 every request that simply did not mention a limit; that would make strict policies unusable rather than safe.

model_fields_set is what distinguishes the two, which is precisely why it must be consulted rather than comparing against the default.

Projection is always intersected rather than raising, since an allow-list exists to hide properties: refusing the request would tell the caller which forbidden property they guessed correctly.

Source code in graflo/architecture/query/models.py
def narrowed(self, caps: QueryCaps) -> Self:
    """Return a copy fitted to *caps*, raising on anything explicitly asked for.

    The split is what makes caps both enforceable and usable:

    - A value the caller **explicitly set** above a cap raises
      :class:`CapExceededError`. Silently clamping it would hand back a
      partial answer the caller believes is complete — the failure mode this
      codebase already rejected for TigerGraph edge direction.
    - A value the caller **left at its default** is clamped. A policy of
      ``max_rows=5`` must not 422 every request that simply did not mention
      a limit; that would make strict policies unusable rather than safe.

    ``model_fields_set`` is what distinguishes the two, which is precisely
    why it must be consulted rather than comparing against the default.

    Projection is always intersected rather than raising, since an
    allow-list exists to *hide* properties: refusing the request would tell
    the caller which forbidden property they guessed correctly.
    """
    effective = HARD_CAPS.narrow(caps)
    explicit = self.model_fields_set
    narrowed = self.model_copy(deep=True)

    if "limit" in explicit and self.limit > effective.max_rows:
        raise CapExceededError("max_rows", self.limit, effective.max_rows)
    narrowed.limit = min(self.limit, effective.max_rows)

    if "timeout_s" in explicit and self.timeout_s > effective.timeout_s:
        raise CapExceededError("timeout_s", self.timeout_s, effective.timeout_s)
    narrowed.timeout_s = min(self.timeout_s, effective.timeout_s)

    if (
        narrowed.projection is not None
        and effective.projection_allow_list is not None
    ):
        permitted = set(effective.projection_allow_list)
        narrowed.projection = [n for n in narrowed.projection if n in permitted]

    narrowed._narrow_specific(effective, explicit)
    return narrowed

NeighborQuery

Bases: GraphQuery

What is adjacent to one anchor vertex.

The instance-plane counterpart of SchemaGraph.schema_neighbors. They are different questions and must never share a name: this one asks which rows are adjacent, the other which types can be.

Source code in graflo/architecture/query/models.py
class NeighborQuery(GraphQuery):
    """What is adjacent to one anchor vertex.

    The instance-plane counterpart of ``SchemaGraph.schema_neighbors``. They are
    different questions and must never share a name: this one asks which *rows*
    are adjacent, the other which *types* can be.
    """

    vertex_type: str = PydanticField(..., description="Logical type of the anchor.")
    key: str | dict[str, Any] = PydanticField(
        ..., description="Anchor identity value, or a single-field mapping."
    )
    hops: int = PydanticField(
        default=1, ge=1, description="Hop distance, within `max_hops`."
    )
    direction: EdgeDirection = PydanticField(
        default=EdgeDirection.OUT,
        description=(
            "Orientation followed from the anchor. Defaults to OUT, matching "
            "`Connection.fetch_edges`; an edge declared `directed: false` is "
            "followed both ways regardless."
        ),
    )
    edge_relations: list[str] | None = PydanticField(
        default=None, description="Restrict to these relations. None means all."
    )
    filters: FilterExpression | None = PydanticField(default=None)

    @property
    def edge_direction(self) -> EdgeDirection:
        """`direction` as an enum, for handing to a driver.

        `ConfigBaseModel` sets ``use_enum_values=True``, so the stored value is
        the bare string. Backends compare against `EdgeDirection` members, and a
        string silently matches none of them — which each backend then resolves
        differently, so the same query returns different neighbourhoods
        depending on where it runs. Always cross the boundary through this.
        """
        return EdgeDirection(self.direction)

    def _validate_specific(self, caps: QueryCaps) -> None:
        if self.hops > caps.max_hops:
            raise CapExceededError("max_hops", self.hops, caps.max_hops)
        if self.edge_relations and len(self.edge_relations) > caps.max_edge_types:
            raise CapExceededError(
                "max_edge_types", len(self.edge_relations), caps.max_edge_types
            )

    def _narrow_specific(self, caps: QueryCaps, explicit: set[str]) -> None:
        if "hops" in explicit and self.hops > caps.max_hops:
            raise CapExceededError("max_hops", self.hops, caps.max_hops)
        self.hops = min(self.hops, caps.max_hops)
        if self.edge_relations and len(self.edge_relations) > caps.max_edge_types:
            raise CapExceededError(
                "max_edge_types", len(self.edge_relations), caps.max_edge_types
            )

edge_direction property

direction as an enum, for handing to a driver.

ConfigBaseModel sets use_enum_values=True, so the stored value is the bare string. Backends compare against EdgeDirection members, and a string silently matches none of them — which each backend then resolves differently, so the same query returns different neighbourhoods depending on where it runs. Always cross the boundary through this.

NodeQuery

Bases: GraphQuery

List vertices of one type, optionally filtered.

Source code in graflo/architecture/query/models.py
class NodeQuery(GraphQuery):
    """List vertices of one type, optionally filtered."""

    vertex_type: str = PydanticField(..., description="Logical vertex type name.")
    filters: FilterExpression | None = PydanticField(
        default=None, description="Predicate applied before limiting."
    )

QueryCaps

Bases: ConfigBaseModel

Ceilings a read request must fit inside.

Defaults are the core ceiling. A deployment narrows them per connection via :meth:GraphQuery.narrowed; nothing widens them.

Source code in graflo/architecture/query/caps.py
class QueryCaps(ConfigBaseModel):
    """Ceilings a read request must fit inside.

    Defaults are the core ceiling. A deployment narrows them per connection via
    :meth:`GraphQuery.narrowed`; nothing widens them.
    """

    max_hops: int = PydanticField(
        default=3, ge=1, description="Deepest traversal permitted."
    )
    max_rows: int = PydanticField(
        default=1000, ge=1, description="Most rows a single query may return."
    )
    max_elements: int = PydanticField(
        default=5000,
        ge=1,
        description="Most vertices plus edges a response may carry.",
    )
    timeout_s: float = PydanticField(
        default=30.0, gt=0, description="Longest a query may run."
    )
    max_edge_types: int = PydanticField(
        default=20,
        ge=1,
        description="Most distinct relations one request may name.",
    )
    max_seeds: int = PydanticField(
        default=10,
        ge=1,
        description="Most anchor vertices a traversal may start from.",
    )
    projection_allow_list: list[str] | None = PydanticField(
        default=None,
        description=(
            "Property names a response may include. None means unrestricted; "
            "an empty list means nothing may be projected, which is not the "
            "same thing."
        ),
    )

    def narrow(self, other: QueryCaps) -> QueryCaps:
        """Combine two cap sets, taking the stricter of each.

        Narrowing is a lattice meet, not an override: a policy that tried to
        raise a ceiling silently becomes a no-op rather than a privilege
        escalation.
        """
        allow: list[str] | None
        if self.projection_allow_list is None:
            allow = other.projection_allow_list
        elif other.projection_allow_list is None:
            allow = self.projection_allow_list
        else:
            # Intersection, order fixed for reproducibility.
            permitted = set(other.projection_allow_list)
            allow = [n for n in self.projection_allow_list if n in permitted]
        return QueryCaps(
            max_hops=min(self.max_hops, other.max_hops),
            max_rows=min(self.max_rows, other.max_rows),
            max_elements=min(self.max_elements, other.max_elements),
            timeout_s=min(self.timeout_s, other.timeout_s),
            max_edge_types=min(self.max_edge_types, other.max_edge_types),
            max_seeds=min(self.max_seeds, other.max_seeds),
            projection_allow_list=allow,
        )

narrow(other)

Combine two cap sets, taking the stricter of each.

Narrowing is a lattice meet, not an override: a policy that tried to raise a ceiling silently becomes a no-op rather than a privilege escalation.

Source code in graflo/architecture/query/caps.py
def narrow(self, other: QueryCaps) -> QueryCaps:
    """Combine two cap sets, taking the stricter of each.

    Narrowing is a lattice meet, not an override: a policy that tried to
    raise a ceiling silently becomes a no-op rather than a privilege
    escalation.
    """
    allow: list[str] | None
    if self.projection_allow_list is None:
        allow = other.projection_allow_list
    elif other.projection_allow_list is None:
        allow = self.projection_allow_list
    else:
        # Intersection, order fixed for reproducibility.
        permitted = set(other.projection_allow_list)
        allow = [n for n in self.projection_allow_list if n in permitted]
    return QueryCaps(
        max_hops=min(self.max_hops, other.max_hops),
        max_rows=min(self.max_rows, other.max_rows),
        max_elements=min(self.max_elements, other.max_elements),
        timeout_s=min(self.timeout_s, other.timeout_s),
        max_edge_types=min(self.max_edge_types, other.max_edge_types),
        max_seeds=min(self.max_seeds, other.max_seeds),
        projection_allow_list=allow,
    )

QueryResult

Bases: ConfigBaseModel

A query's answer plus the honest caveats.

Source code in graflo/architecture/query/result.py
class QueryResult(ConfigBaseModel):
    """A query's answer plus the honest caveats."""

    container: GraphContainer = PydanticField(
        default_factory=GraphContainer,
        description="DB-agnostic vertices and edges, identical across backends.",
    )
    element_count: int = PydanticField(
        default=0, ge=0, description="Vertices plus edges carried."
    )
    truncated: bool = PydanticField(
        default=False, description="Whether a cap cut the answer short."
    )
    caps_hit: list[str] = PydanticField(
        default_factory=list,
        description=(
            "Which caps bound this answer, by name. Empty when nothing bound "
            "it — the only way a caller can tell a complete answer from one "
            "that happens to fit."
        ),
    )
    elapsed_ms: int = PydanticField(default=0, ge=0)

    @classmethod
    def of(
        cls,
        container: GraphContainer,
        *,
        caps_hit: list[str] | None = None,
        elapsed_ms: int = 0,
    ) -> QueryResult:
        """Build a result, deriving the element count from the container."""
        hit = caps_hit or []
        count = sum(len(docs) for docs in container.vertices.values()) + sum(
            len(rows) for rows in container.edges.values()
        )
        return cls(
            container=container,
            element_count=count,
            truncated=bool(hit),
            caps_hit=hit,
            elapsed_ms=elapsed_ms,
        )

of(container, *, caps_hit=None, elapsed_ms=0) classmethod

Build a result, deriving the element count from the container.

Source code in graflo/architecture/query/result.py
@classmethod
def of(
    cls,
    container: GraphContainer,
    *,
    caps_hit: list[str] | None = None,
    elapsed_ms: int = 0,
) -> QueryResult:
    """Build a result, deriving the element count from the container."""
    hit = caps_hit or []
    count = sum(len(docs) for docs in container.vertices.values()) + sum(
        len(rows) for rows in container.edges.values()
    )
    return cls(
        container=container,
        element_count=count,
        truncated=bool(hit),
        caps_hit=hit,
        elapsed_ms=elapsed_ms,
    )

TraverseQuery

Bases: GraphQuery

What is reachable from a set of anchors.

Source code in graflo/architecture/query/models.py
class TraverseQuery(GraphQuery):
    """What is reachable from a set of anchors."""

    seeds: list[dict[str, Any]] = PydanticField(
        ...,
        min_length=1,
        description=(
            "Anchors, each `{vertex_type, key}`. Bounded by `max_seeds`: a "
            "traversal fans out per seed, so seed count multiplies cost."
        ),
    )
    max_hops: int = PydanticField(default=2, ge=1)
    direction: EdgeDirection = PydanticField(default=EdgeDirection.ANY)
    edge_relations: list[str] | None = PydanticField(default=None)
    filters: FilterExpression | None = PydanticField(default=None)

    @property
    def edge_direction(self) -> EdgeDirection:
        """`direction` as an enum. See `NeighborQuery.edge_direction`."""
        return EdgeDirection(self.direction)

    def _validate_specific(self, caps: QueryCaps) -> None:
        if self.max_hops > caps.max_hops:
            raise CapExceededError("max_hops", self.max_hops, caps.max_hops)
        if len(self.seeds) > caps.max_seeds:
            raise CapExceededError("max_seeds", len(self.seeds), caps.max_seeds)
        if self.edge_relations and len(self.edge_relations) > caps.max_edge_types:
            raise CapExceededError(
                "max_edge_types", len(self.edge_relations), caps.max_edge_types
            )
        for seed in self.seeds:
            missing = {"vertex_type", "key"} - set(seed)
            if missing:
                raise ValueError(
                    f"seed {seed!r} is missing {sorted(missing)}; each seed needs "
                    "a vertex_type and a key"
                )

    def _narrow_specific(self, caps: QueryCaps, explicit: set[str]) -> None:
        if "max_hops" in explicit and self.max_hops > caps.max_hops:
            raise CapExceededError("max_hops", self.max_hops, caps.max_hops)
        self.max_hops = min(self.max_hops, caps.max_hops)
        # Seeds are always explicit — there is no default set of anchors — so
        # dropping any of them would silently answer a different question.
        if len(self.seeds) > caps.max_seeds:
            raise CapExceededError("max_seeds", len(self.seeds), caps.max_seeds)
        if self.edge_relations and len(self.edge_relations) > caps.max_edge_types:
            raise CapExceededError(
                "max_edge_types", len(self.edge_relations), caps.max_edge_types
            )

edge_direction property

direction as an enum. See NeighborQuery.edge_direction.