graflo.architecture.schema.vertex

Vertex configuration and management for graph databases.

This module provides classes and utilities for managing vertices in graph databases. It handles vertex configuration, property management, identity, and filtering operations. The module supports both ArangoDB and Neo4j through the DBType enum.

Key Components

• Vertex: Represents a vertex with its properties and identity
• VertexConfig: Manages vertices and their configurations

Example

vertex = Vertex(name="user", properties=["id", "name"]) config = VertexConfig(vertices=[vertex]) props = config.properties("user") # Returns list[Field] prop_names = config.property_names("user") # Returns list[str]

Attributes

IdentityMode = Literal['natural', 'hash', 'blank', 'assigned'] module-attribute

PropertiesInputType = list[str] | list['Field'] | list[dict[str, Any]] module-attribute

RELOCATED_VERTEX_KEYS = {'dbname': 'db_profile.vertex_storage_names', 'indexes': 'db_profile.vertex_indexes', 'transforms': 'the ingestion model, as resource pipeline steps'} module-attribute

SCALAR_FIELD_TYPES = frozenset({FieldType.INT, FieldType.UINT, FieldType.FLOAT, FieldType.DOUBLE, FieldType.BOOL, FieldType.STRING, FieldType.DATETIME, FieldType.UUID}) module-attribute

SCALAR_FIELD_TYPE_VALUES = frozenset(ft.value for ft in SCALAR_FIELD_TYPES) module-attribute

VertexName = str module-attribute

logger = logging.getLogger(__name__) module-attribute

Classes

Field

Bases: ConfigBaseModel

Represents a typed field in a vertex.

Field objects behave like strings for backward compatibility. They can be used in sets, as dictionary keys, and in string comparisons. The type information is preserved for databases that need it (like TigerGraph).

Attributes:

Name	Type	Description
name	VertexName	Name of the field
type	FieldType | None	Optional type of the field. Can be FieldType enum, str, or None at construction. Strings are converted to FieldType enum by the validator. None is allowed (most databases like ArangoDB don't require types). Defaults to None.
item_type	FieldType | None	Required when type is LIST; must be a scalar FieldType.

Source code in graflo/architecture/schema/vertex.py

class Field(ConfigBaseModel):
    """Represents a typed field in a vertex.

    Field objects behave like strings for backward compatibility. They can be used
    in sets, as dictionary keys, and in string comparisons. The type information
    is preserved for databases that need it (like TigerGraph).

    Attributes:
        name: Name of the field
        type: Optional type of the field. Can be FieldType enum, str, or None at construction.
              Strings are converted to FieldType enum by the validator.
              None is allowed (most databases like ArangoDB don't require types).
              Defaults to None.
        item_type: Required when ``type`` is ``LIST``; must be a scalar FieldType.
    """

    model_config = ConfigDict(extra="forbid")

    name: VertexName = PydanticField(
        ...,
        description="Name of the field (e.g. column or attribute name).",
    )
    type: FieldType | None = PydanticField(
        default=None,
        description="Optional field type for databases that require it (e.g. TigerGraph: INT, STRING). None for schema-agnostic backends.",
    )
    item_type: FieldType | None = PydanticField(
        default=None,
        description=(
            "Element type when ``type`` is LIST. Must be a scalar "
            "(INT, UINT, FLOAT, DOUBLE, BOOL, STRING, DATETIME, UUID)."
        ),
    )
    description: str | None = PydanticField(
        default=None,
        description="Optional semantic description of the field for schema inference and downstream reasoning.",
    )
    semantics: FieldSemantics | None = PydanticField(
        default=None,
        description="Optional external-vocabulary anchors and unit for this field.",
    )

    @field_validator("type", mode="before")
    @classmethod
    def normalize_type(cls, v: Any) -> FieldType | None:
        return _normalize_field_type_input(v, label="type")

    @field_validator("item_type", mode="before")
    @classmethod
    def normalize_item_type(cls, v: Any) -> FieldType | None:
        return _normalize_field_type_input(v, label="item_type")

    @model_validator(mode="after")
    def validate_list_item_type(self) -> Field:
        type_val = field_type_value(self.type)
        item_val = field_type_value(self.item_type)

        if type_val == FieldType.LIST.value:
            if item_val is None:
                raise ValueError(
                    f"Field '{self.name}': type LIST requires item_type "
                    f"(one of {', '.join(sorted(SCALAR_FIELD_TYPE_VALUES))})"
                )
            if item_val not in SCALAR_FIELD_TYPE_VALUES:
                raise ValueError(
                    f"Field '{self.name}': LIST item_type must be a scalar "
                    f"({', '.join(sorted(SCALAR_FIELD_TYPE_VALUES))}), "
                    f"got '{item_val}'"
                )
        elif item_val is not None:
            raise ValueError(
                f"Field '{self.name}': item_type is only allowed when type is LIST, "
                f"got type '{type_val}'"
            )
        return self

    def __str__(self) -> str:
        """Return field name as string for backward compatibility."""
        return self.name

    def __repr__(self) -> str:
        """Return representation including type information."""
        if is_list_field_type(self.type):
            return (
                f"Field(name='{self.name}', type='LIST', "
                f"item_type='{field_type_value(self.item_type)}')"
            )
        if self.type:
            return f"Field(name='{self.name}', type='{self.type}')"
        return f"Field(name='{self.name}')"

    def __hash__(self) -> int:
        """Hash by name only, allowing Field objects to work in sets and as dict keys."""
        return hash(self.name)

    def __eq__(self, other: object) -> bool:
        """Compare equal to strings with same name, or other Field objects with same name."""
        if isinstance(other, Field):
            return self.name == other.name
        if isinstance(other, str):
            return self.name == other
        return False

    def __ne__(self, other: object) -> bool:
        """Compare not equal."""
        return not self.__eq__(other)

Attributes

description = PydanticField(default=None, description='Optional semantic description of the field for schema inference and downstream reasoning.') class-attribute instance-attribute

item_type = PydanticField(default=None, description='Element type when ``type`` is LIST. Must be a scalar (INT, UINT, FLOAT, DOUBLE, BOOL, STRING, DATETIME, UUID).') class-attribute instance-attribute

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

name = PydanticField(..., description='Name of the field (e.g. column or attribute name).') class-attribute instance-attribute

semantics = PydanticField(default=None, description='Optional external-vocabulary anchors and unit for this field.') class-attribute instance-attribute

type = PydanticField(default=None, description='Optional field type for databases that require it (e.g. TigerGraph: INT, STRING). None for schema-agnostic backends.') class-attribute instance-attribute

Methods:

__eq__(other)

Compare equal to strings with same name, or other Field objects with same name.

Source code in graflo/architecture/schema/vertex.py

def __eq__(self, other: object) -> bool:
    """Compare equal to strings with same name, or other Field objects with same name."""
    if isinstance(other, Field):
        return self.name == other.name
    if isinstance(other, str):
        return self.name == other
    return False

__hash__()

Hash by name only, allowing Field objects to work in sets and as dict keys.

Source code in graflo/architecture/schema/vertex.py

def __hash__(self) -> int:
    """Hash by name only, allowing Field objects to work in sets and as dict keys."""
    return hash(self.name)

__ne__(other)

Compare not equal.

Source code in graflo/architecture/schema/vertex.py

def __ne__(self, other: object) -> bool:
    """Compare not equal."""
    return not self.__eq__(other)

__repr__()

Return representation including type information.

Source code in graflo/architecture/schema/vertex.py

def __repr__(self) -> str:
    """Return representation including type information."""
    if is_list_field_type(self.type):
        return (
            f"Field(name='{self.name}', type='LIST', "
            f"item_type='{field_type_value(self.item_type)}')"
        )
    if self.type:
        return f"Field(name='{self.name}', type='{self.type}')"
    return f"Field(name='{self.name}')"

__str__()

Return field name as string for backward compatibility.

Source code in graflo/architecture/schema/vertex.py

def __str__(self) -> str:
    """Return field name as string for backward compatibility."""
    return self.name

normalize_item_type(v) classmethod

Source code in graflo/architecture/schema/vertex.py

@field_validator("item_type", mode="before")
@classmethod
def normalize_item_type(cls, v: Any) -> FieldType | None:
    return _normalize_field_type_input(v, label="item_type")

normalize_type(v) classmethod

Source code in graflo/architecture/schema/vertex.py

@field_validator("type", mode="before")
@classmethod
def normalize_type(cls, v: Any) -> FieldType | None:
    return _normalize_field_type_input(v, label="type")

validate_list_item_type()

Source code in graflo/architecture/schema/vertex.py

@model_validator(mode="after")
def validate_list_item_type(self) -> Field:
    type_val = field_type_value(self.type)
    item_val = field_type_value(self.item_type)

    if type_val == FieldType.LIST.value:
        if item_val is None:
            raise ValueError(
                f"Field '{self.name}': type LIST requires item_type "
                f"(one of {', '.join(sorted(SCALAR_FIELD_TYPE_VALUES))})"
            )
        if item_val not in SCALAR_FIELD_TYPE_VALUES:
            raise ValueError(
                f"Field '{self.name}': LIST item_type must be a scalar "
                f"({', '.join(sorted(SCALAR_FIELD_TYPE_VALUES))}), "
                f"got '{item_val}'"
            )
    elif item_val is not None:
        raise ValueError(
            f"Field '{self.name}': item_type is only allowed when type is LIST, "
            f"got type '{type_val}'"
        )
    return self

FieldMergeConflict

Bases: Refusal

One property that two declarations describe incompatibly.

The per-property clause and the remedy are kept apart from the rendered message so :func:union_field_lists can report every conflicting property under one owner heading instead of one error per run.

field and conflict are the same two facts in structured form: which property, and whether the disagreement is about types or about units. A caller classifying the refusal reads those rather than the prose.

Source code in graflo/architecture/schema/vertex.py

class FieldMergeConflict(Refusal):
    """One property that two declarations describe incompatibly.

    The per-property clause and the remedy are kept apart from the rendered
    message so :func:`union_field_lists` can report every conflicting property
    under one owner heading instead of one error per run.

    ``field`` and ``conflict`` are the same two facts in structured form: which
    property, and whether the disagreement is about types or about units. A
    caller classifying the refusal reads those rather than the prose.
    """

    def __init__(self, owner: str, reason: str, remedy: str, *, field: str) -> None:
        self.reason = reason
        self.remedy = remedy
        self.owner = owner
        self.field = field
        self.conflict = _conflict_kind([reason])
        super().__init__(
            f"{_conflict_heading([reason], owner)}, {reason}. {remedy}",
            check=_FIELD_CHECK[self.conflict],
        )

Attributes

conflict = _conflict_kind([reason]) instance-attribute

field = field instance-attribute

owner = owner instance-attribute

reason = reason instance-attribute

remedy = remedy instance-attribute

Methods:

__init__(owner, reason, remedy, *, field)

Source code in graflo/architecture/schema/vertex.py

def __init__(self, owner: str, reason: str, remedy: str, *, field: str) -> None:
    self.reason = reason
    self.remedy = remedy
    self.owner = owner
    self.field = field
    self.conflict = _conflict_kind([reason])
    super().__init__(
        f"{_conflict_heading([reason], owner)}, {reason}. {remedy}",
        check=_FIELD_CHECK[self.conflict],
    )

FieldMergeError

Bases: Refusal

Every property two declarations describe incompatibly, in one refusal.

:func:union_field_lists collects rather than raising on the first clash, so an author fixing a large merge sees all of them at once. The individual :class:FieldMergeConflict objects stay on conflicts and their property names on fields, so a caller can point at what is wrong without parsing the rendered message.

check follows :func:_conflict_kind over the whole set, which is what the heading already says -- so a reader of the message and a caller keying on check are told the same thing, and a mixed set reports as a type conflict on both.

Source code in graflo/architecture/schema/vertex.py

class FieldMergeError(Refusal):
    """Every property two declarations describe incompatibly, in one refusal.

    :func:`union_field_lists` collects rather than raising on the first clash,
    so an author fixing a large merge sees all of them at once. The
    individual :class:`FieldMergeConflict` objects stay on ``conflicts`` and
    their property names on ``fields``, so a caller can point at what is
    wrong without parsing the rendered message.

    ``check`` follows :func:`_conflict_kind` over the whole set, which is what
    the heading already says -- so a reader of the message and a caller keying
    on ``check`` are told the same thing, and a mixed set reports as a type
    conflict on both.
    """

    def __init__(
        self,
        message: str,
        *,
        owner: str,
        conflicts: tuple[FieldMergeConflict, ...],
    ) -> None:
        self.owner = owner
        self.conflicts = conflicts
        self.fields = tuple(dict.fromkeys(c.field for c in conflicts))
        self.conflict = _conflict_kind([c.reason for c in conflicts])
        super().__init__(message, check=_FIELD_CHECK[self.conflict])

Attributes

conflict = _conflict_kind([c.reason for c in conflicts]) instance-attribute

conflicts = conflicts instance-attribute

fields = tuple(dict.fromkeys(c.field for c in conflicts)) instance-attribute

owner = owner instance-attribute

Methods:

__init__(message, *, owner, conflicts)

Source code in graflo/architecture/schema/vertex.py

def __init__(
    self,
    message: str,
    *,
    owner: str,
    conflicts: tuple[FieldMergeConflict, ...],
) -> None:
    self.owner = owner
    self.conflicts = conflicts
    self.fields = tuple(dict.fromkeys(c.field for c in conflicts))
    self.conflict = _conflict_kind([c.reason for c in conflicts])
    super().__init__(message, check=_FIELD_CHECK[self.conflict])

FieldType

Bases: BaseEnum

Supported field types for graph databases.

These types are primarily used for TigerGraph, which requires explicit field types. Other databases (ArangoDB, Neo4j) may use different type systems or not require types.

Attributes:

Name	Type	Description
INT		Integer type
UINT		Unsigned integer type
FLOAT		Floating point type
DOUBLE		Double precision floating point type
BOOL		Boolean type
STRING		String type
DATETIME		DateTime type
UUID		Logical UUID scalar (backends store as STRING/TEXT)
LIST		Homogeneous list of scalars (requires Field.item_type)

Source code in graflo/architecture/schema/vertex.py

class FieldType(BaseEnum):
    """Supported field types for graph databases.

    These types are primarily used for TigerGraph, which requires explicit field types.
    Other databases (ArangoDB, Neo4j) may use different type systems or not require types.

    Attributes:
        INT: Integer type
        UINT: Unsigned integer type
        FLOAT: Floating point type
        DOUBLE: Double precision floating point type
        BOOL: Boolean type
        STRING: String type
        DATETIME: DateTime type
        UUID: Logical UUID scalar (backends store as STRING/TEXT)
        LIST: Homogeneous list of scalars (requires ``Field.item_type``)
    """

    INT = "INT"
    UINT = "UINT"
    FLOAT = "FLOAT"
    DOUBLE = "DOUBLE"
    BOOL = "BOOL"
    STRING = "STRING"
    DATETIME = "DATETIME"
    UUID = "UUID"
    LIST = "LIST"

Attributes

BOOL = 'BOOL' class-attribute instance-attribute

DATETIME = 'DATETIME' class-attribute instance-attribute

DOUBLE = 'DOUBLE' class-attribute instance-attribute

FLOAT = 'FLOAT' class-attribute instance-attribute

INT = 'INT' class-attribute instance-attribute

LIST = 'LIST' class-attribute instance-attribute

STRING = 'STRING' class-attribute instance-attribute

UINT = 'UINT' class-attribute instance-attribute

UUID = 'UUID' class-attribute instance-attribute

SecondaryIdentity

Bases: ConfigBaseModel

An alternate field-set that identifies a vertex without upserting it.

Vertices upsert on their primary identity. Edge-only sources often reference endpoints by another field-set — a business key, an ISIN, a source-local code — which is what a secondary identity names.

Uniqueness is soft: it is not enforced by a database constraint, so a lookup may match several vertices and the ingestion-level ambiguity policy decides what happens.

Examples:

>>> SecondaryIdentity(name="by_isin", fields=["isin"])
>>> SecondaryIdentity.model_validate(["org", "local_code"])  # auto-named

Source code in graflo/architecture/schema/vertex.py

class SecondaryIdentity(ConfigBaseModel):
    """An alternate field-set that identifies a vertex without upserting it.

    Vertices upsert on their primary ``identity``. Edge-only sources often
    reference endpoints by another field-set — a business key, an ISIN, a
    source-local code — which is what a secondary identity names.

    Uniqueness is *soft*: it is not enforced by a database constraint, so a
    lookup may match several vertices and the ingestion-level ambiguity policy
    decides what happens.

    Examples:
        >>> SecondaryIdentity(name="by_isin", fields=["isin"])
        >>> SecondaryIdentity.model_validate(["org", "local_code"])  # auto-named
    """

    name: str | None = PydanticField(
        default=None,
        description=(
            "Optional handle used by an edge step to select this field-set "
            "(e.g. source_match: by_isin)."
        ),
    )
    fields: list[str] = PydanticField(
        ...,
        min_length=1,
        description="Property names forming this alternate key.",
    )

    @model_validator(mode="before")
    @classmethod
    def normalize_authored_shape(cls, data: Any) -> Any:
        """Accept a bare ``[field, ...]`` list alongside the mapping form."""
        if isinstance(data, (list, tuple)):
            return {"fields": list(data)}
        if isinstance(data, str):
            return {"fields": [data]}
        return data

    @field_validator("fields", mode="after")
    @classmethod
    def dedupe_fields(cls, v: list[str]) -> list[str]:
        return _dedupe_ordered(v)

    @property
    def field_set(self) -> frozenset[str]:
        return frozenset(self.fields)

Attributes

field_set property

fields = PydanticField(..., min_length=1, description='Property names forming this alternate key.') class-attribute instance-attribute

name = PydanticField(default=None, description='Optional handle used by an edge step to select this field-set (e.g. source_match: by_isin).') class-attribute instance-attribute

Methods:

dedupe_fields(v) classmethod

Source code in graflo/architecture/schema/vertex.py

@field_validator("fields", mode="after")
@classmethod
def dedupe_fields(cls, v: list[str]) -> list[str]:
    return _dedupe_ordered(v)

normalize_authored_shape(data) classmethod

Accept a bare [field, ...] list alongside the mapping form.

Source code in graflo/architecture/schema/vertex.py

@model_validator(mode="before")
@classmethod
def normalize_authored_shape(cls, data: Any) -> Any:
    """Accept a bare ``[field, ...]`` list alongside the mapping form."""
    if isinstance(data, (list, tuple)):
        return {"fields": list(data)}
    if isinstance(data, str):
        return {"fields": [data]}
    return data

Vertex

Bases: ConfigBaseModel

Represents a vertex in the graph database.

A vertex is a fundamental unit in the graph that can have properties, identity, and filters. Properties can be specified as strings, Field objects, or dicts. Internally, properties are stored as Field objects but behave like strings where a string-like Field is needed.

Attributes:

Name	Type	Description
name	str	Name of the vertex
properties	list[Field]	List of field names (str), Field objects, or dicts. Will be normalized to Field objects by the validator.
identity	list[str]	List of property names forming logical primary identity
filters	list[FilterExpression]	List of filter expressions

Examples:

>>> # List of strings
>>> v1 = Vertex(name="user", properties=["id", "name"])

>>> # Typed properties: list of Field objects
>>> v2 = Vertex(name="user", properties=[
...     Field(name="id", type="INT"),
...     Field(name="name", type="STRING")
... ])

>>> # From dicts (e.g., from YAML/JSON)
>>> v3 = Vertex(name="user", properties=[
...     {"name": "id", "type": "INT"},
...     {"name": "name"}  # defaults to None type
... ])

Source code in graflo/architecture/schema/vertex.py

class Vertex(ConfigBaseModel):
    """Represents a vertex in the graph database.

    A vertex is a fundamental unit in the graph that can have properties, identity,
    and filters. Properties can be specified as strings, Field objects, or dicts.
    Internally, properties are stored as Field objects but behave like strings
    where a string-like Field is needed.

    Attributes:
        name: Name of the vertex
        properties: List of field names (str), Field objects, or dicts.
               Will be normalized to Field objects by the validator.
        identity: List of property names forming logical primary identity
        filters: List of filter expressions

    Examples:
        >>> # List of strings
        >>> v1 = Vertex(name="user", properties=["id", "name"])

        >>> # Typed properties: list of Field objects
        >>> v2 = Vertex(name="user", properties=[
        ...     Field(name="id", type="INT"),
        ...     Field(name="name", type="STRING")
        ... ])

        >>> # From dicts (e.g., from YAML/JSON)
        >>> v3 = Vertex(name="user", properties=[
        ...     {"name": "id", "type": "INT"},
        ...     {"name": "name"}  # defaults to None type
        ... ])
    """

    # Unknown keys are an error, like every other config model. They used to be
    # dropped silently, which meant an authored block a given graflo did not know
    # about vanished on the next round-trip with no diagnostic at all.
    model_config = ConfigDict(extra="forbid")

    name: str = PydanticField(
        ...,
        description="Name of the vertex type (e.g. user, post, company).",
    )
    properties: list[Field] = PydanticField(
        default_factory=list,
        description="List of fields (names, Field objects, or dicts). Normalized to Field objects.",
    )
    identity: list[str] = PydanticField(
        default_factory=list,
        description="Logical identity property names (primary key semantics for matching/upserts).",
    )
    filters: list[FilterExpression] = PydanticField(
        default_factory=list,
        description="Filter expressions (logical formulae) applied when querying this vertex.",
    )
    description: str | None = PydanticField(
        default=None,
        description="Optional semantic description of the vertex meaning, role, and intended interpretation.",
    )
    semantics: Semantics | None = PydanticField(
        default=None,
        description="Optional external-vocabulary anchors for this vertex type.",
    )
    blank: bool = PydanticField(
        default=False,
        description=(
            "True when this vertex has no natural identity and gets an auto-generated ID."
        ),
    )
    assigned: bool = PydanticField(
        default=False,
        description=(
            "True when this vertex uses an intentional UUID primary key: empty identity "
            "is filled with uuid4 at assemble time; not a blank-node placeholder."
        ),
    )
    hash_identity_properties: list[str] = PydanticField(
        default_factory=list,
        description=(
            "Source field names whose combined values are SHA256-hashed to produce "
            "a deterministic synthetic 'id'. Non-empty only when no natural key is "
            "narrow enough to store directly. Distinct from blank (random UUID)."
        ),
    )
    identity_funnel: IdentityFunnel | None = PydanticField(
        default=None,
        description=(
            "Ordered fallback branches deriving a deterministic synthetic 'id'. "
            "The first branch whose fields are all present wins. Generalizes "
            "hash_identity_properties (the single-branch case); the two are "
            "mutually exclusive."
        ),
    )
    secondary_identities: list[SecondaryIdentity] = PydanticField(
        default_factory=list,
        description=(
            "Alternate field-sets that identify this vertex for lookup only. Edge "
            "endpoints may be matched on one of these instead of the primary "
            "identity; upserts always use identity. Soft uniqueness."
        ),
    )

    @field_validator("properties", mode="before")
    @classmethod
    def convert_to_properties(cls, v: Any) -> Any:
        if not isinstance(v, list):
            raise ValueError("properties must be a list")
        return [_normalize_fields_item(item) for item in v]

    @field_validator("filters", mode="before")
    @classmethod
    def convert_to_expressions(cls, v: Any) -> Any:
        if not isinstance(v, list):
            return v
        result: list[FilterExpression] = []
        for item in v:
            if isinstance(item, FilterExpression):
                result.append(item)
            elif isinstance(item, (dict, list)):
                from graflo.filter.onto import parse_filter_expression

                result.append(parse_filter_expression(item))
            else:
                raise ValueError(
                    "each filter must be a FilterExpression instance or a dict/list (parsed as FilterExpression)"
                )
        return result

    @field_validator("identity", mode="before")
    @classmethod
    def convert_identity(cls, v: Any) -> Any:
        if v is None:
            return []
        if isinstance(v, tuple):
            return list(v)
        if isinstance(v, list):
            return v
        raise ValueError("identity must be a list[str]")

    @field_validator("hash_identity_properties", mode="before")
    @classmethod
    def convert_hash_identity_properties(cls, v: Any) -> Any:
        if v is None:
            return []
        if isinstance(v, tuple):
            return list(v)
        if isinstance(v, list):
            return v
        raise ValueError("hash_identity_properties must be a list[str]")

    @model_validator(mode="before")
    @classmethod
    def _reject_relocated_keys(cls, data: Any) -> Any:
        """Point at the new home for keys that used to live on a vertex.

        These were silently dropped while this model ignored extra keys, so an
        author moving an old manifest forward got no signal that a physical name
        or an index had gone missing. Naming the destination is the whole value
        of forbidding them.
        """
        if not isinstance(data, dict):
            return data
        relocated = [key for key in RELOCATED_VERTEX_KEYS if key in data]
        if relocated:
            name = data.get("name", "<unnamed>")
            moves = "; ".join(
                f"'{key}' now belongs in {RELOCATED_VERTEX_KEYS[key]}"
                for key in relocated
            )
            raise ValueError(f"Vertex '{name}': {moves}")
        return data

    @model_validator(mode="after")
    def set_identity(self) -> Vertex:
        if self.blank and self.assigned:
            raise ValueError(
                f"Vertex '{self.name}': blank and assigned are mutually exclusive"
            )
        if self.assigned and self.hash_identity_properties:
            raise ValueError(
                f"Vertex '{self.name}': assigned and hash_identity_properties "
                "are mutually exclusive"
            )
        if self.blank and self.hash_identity_properties:
            raise ValueError(
                f"Vertex '{self.name}': blank and hash_identity_properties are "
                "mutually exclusive — identity_mode reads blank first, so the "
                "vertex would key on a generated id and never consult the digest"
            )
        if self.identity_funnel is not None:
            if self.hash_identity_properties:
                raise ValueError(
                    f"Vertex '{self.name}': hash_identity_properties and "
                    "identity_funnel are mutually exclusive — a funnel with one "
                    "branch is the general form of a flat hash key"
                )
            if self.blank:
                raise ValueError(
                    f"Vertex '{self.name}': blank and identity_funnel are "
                    "mutually exclusive — a blank vertex has no source fields "
                    "to digest"
                )
            if self.assigned:
                raise ValueError(
                    f"Vertex '{self.name}': assigned and identity_funnel are "
                    "mutually exclusive"
                )
        merged_properties = union_field_lists(
            self.properties, owner=f"vertex {self.name!r}"
        )
        identity_names = _dedupe_ordered(list(self.identity))
        hash_identity_names = _dedupe_ordered(list(self.hash_identity_properties))
        funnel_field_names = (
            self.identity_funnel.field_names if self.identity_funnel else []
        )
        list_prop_names = {
            f.name for f in merged_properties if is_list_field_type(f.type)
        }
        for name in identity_names:
            if name in list_prop_names:
                raise ValueError(
                    f"Vertex '{self.name}': LIST-typed property '{name}' "
                    "cannot be used as identity"
                )
        for name in hash_identity_names:
            if name in list_prop_names:
                raise ValueError(
                    f"Vertex '{self.name}': LIST-typed property '{name}' "
                    "cannot be used in hash_identity_properties"
                )
        for name in funnel_field_names:
            if name in list_prop_names:
                raise ValueError(
                    f"Vertex '{self.name}': LIST-typed property '{name}' "
                    "cannot be used in identity_funnel"
                )
        seen_names = {f.name for f in merged_properties}
        augmented = list(merged_properties)
        for name in identity_names + hash_identity_names + funnel_field_names:
            if name not in seen_names:
                synth_type = FieldType.UUID if self.assigned and name == "id" else None
                augmented.append(Field(name=name, type=synth_type))
                seen_names.add(name)

        secondary = self._validated_secondary_identities(
            identity_names=identity_names,
            list_prop_names=list_prop_names,
        )
        for entry in secondary:
            for name in entry.fields:
                if name not in seen_names:
                    augmented.append(Field(name=name, type=None))
                    seen_names.add(name)

        object.__setattr__(self, "identity", identity_names)
        object.__setattr__(self, "hash_identity_properties", hash_identity_names)
        object.__setattr__(self, "secondary_identities", secondary)
        object.__setattr__(self, "properties", augmented)
        return self

    def _validated_secondary_identities(
        self, *, identity_names: list[str], list_prop_names: set[str]
    ) -> list[SecondaryIdentity]:
        """Validate secondary identities and assign default names.

        A secondary identity must be usable as a lookup key: LIST-typed fields
        cannot be matched, a blank vertex has no source-visible key to match on,
        and repeating the primary identity would be a no-op.
        """
        if not self.secondary_identities:
            return []
        if self.blank:
            raise ValueError(
                f"Vertex '{self.name}': blank vertices cannot declare "
                "secondary_identities — their identity is generated, not sourced"
            )

        primary = frozenset(identity_names)
        seen_names: set[str] = set()
        seen_field_sets: set[frozenset[str]] = set()
        validated: list[SecondaryIdentity] = []

        for position, entry in enumerate(self.secondary_identities):
            for field_name in entry.fields:
                if field_name in list_prop_names:
                    raise ValueError(
                        f"Vertex '{self.name}': LIST-typed property '{field_name}' "
                        "cannot be used in a secondary identity"
                    )
            if entry.field_set == primary:
                raise ValueError(
                    f"Vertex '{self.name}': secondary identity {entry.fields} "
                    "duplicates the primary identity"
                )
            if entry.field_set in seen_field_sets:
                raise ValueError(
                    f"Vertex '{self.name}': duplicate secondary identity {entry.fields}"
                )
            seen_field_sets.add(entry.field_set)

            name = entry.name or f"secondary_{position}"
            if name in seen_names:
                raise ValueError(
                    f"Vertex '{self.name}': duplicate secondary identity name '{name}'"
                )
            seen_names.add(name)
            validated.append(entry.model_copy(update={"name": name}))

        return validated

    @property
    def secondary_identity_names(self) -> list[str]:
        """Names of declared secondary identities, in declaration order."""
        return [entry.name for entry in self.secondary_identities if entry.name]

    def secondary_identity(self, selector: str | list[str]) -> SecondaryIdentity | None:
        """Resolve *selector* to a declared secondary identity.

        Accepts a declared name, an explicit field list equal to a declared
        field-set, or the literal ``"secondary"`` when exactly one is declared.
        """
        if not self.secondary_identities:
            return None
        if isinstance(selector, str):
            if selector == SECONDARY_IDENTITY_SUGAR:
                if len(self.secondary_identities) == 1:
                    return self.secondary_identities[0]
                return None
            for entry in self.secondary_identities:
                if entry.name == selector:
                    return entry
            return None
        wanted = frozenset(selector)
        for entry in self.secondary_identities:
            if entry.field_set == wanted:
                return entry
        return None

    @property
    def property_names(self) -> list[str]:
        """Property names as strings (Field.name for each entry)."""
        return [field.name for field in self.properties]

    @property
    def has_identity_funnel(self) -> bool:
        """True when identity is derived from ordered funnel branches."""
        return self.identity_funnel is not None

    @property
    def digest_source_fields(self) -> list[str]:
        """Fields feeding the synthetic digest, flat or funnel; empty otherwise."""
        if self.identity_funnel is not None:
            return self.identity_funnel.field_names
        return list(self.hash_identity_properties)

    @property
    def identity_mode(self) -> IdentityMode:
        """Runtime identity mode: natural, hash, blank, or assigned UUID PK.

        A funnel resolves to ``hash``: both derive a deterministic synthetic key
        from source fields and share one writer family. Use
        :attr:`has_identity_funnel` to tell them apart.
        """
        if self.blank:
            return "blank"
        if self.assigned:
            return "assigned"
        if self.hash_identity_properties or self.identity_funnel is not None:
            return "hash"
        return "natural"

    def get_properties(self) -> list[Field]:
        return self.properties

    def finish_init(self):
        """Complete logical initialization for vertex."""
        return

Attributes

assigned = PydanticField(default=False, description='True when this vertex uses an intentional UUID primary key: empty identity is filled with uuid4 at assemble time; not a blank-node placeholder.') class-attribute instance-attribute

blank = PydanticField(default=False, description='True when this vertex has no natural identity and gets an auto-generated ID.') class-attribute instance-attribute

description = PydanticField(default=None, description='Optional semantic description of the vertex meaning, role, and intended interpretation.') class-attribute instance-attribute

digest_source_fields property

Fields feeding the synthetic digest, flat or funnel; empty otherwise.

filters = PydanticField(default_factory=list, description='Filter expressions (logical formulae) applied when querying this vertex.') class-attribute instance-attribute

has_identity_funnel property

True when identity is derived from ordered funnel branches.

hash_identity_properties = PydanticField(default_factory=list, description="Source field names whose combined values are SHA256-hashed to produce a deterministic synthetic 'id'. Non-empty only when no natural key is narrow enough to store directly. Distinct from blank (random UUID).") class-attribute instance-attribute

identity = PydanticField(default_factory=list, description='Logical identity property names (primary key semantics for matching/upserts).') class-attribute instance-attribute

identity_funnel = PydanticField(default=None, description="Ordered fallback branches deriving a deterministic synthetic 'id'. The first branch whose fields are all present wins. Generalizes hash_identity_properties (the single-branch case); the two are mutually exclusive.") class-attribute instance-attribute

identity_mode property

Runtime identity mode: natural, hash, blank, or assigned UUID PK.

A funnel resolves to hash: both derive a deterministic synthetic key from source fields and share one writer family. Use :attr:has_identity_funnel to tell them apart.

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

name = PydanticField(..., description='Name of the vertex type (e.g. user, post, company).') class-attribute instance-attribute

properties = PydanticField(default_factory=list, description='List of fields (names, Field objects, or dicts). Normalized to Field objects.') class-attribute instance-attribute

property_names property

Property names as strings (Field.name for each entry).

secondary_identities = PydanticField(default_factory=list, description='Alternate field-sets that identify this vertex for lookup only. Edge endpoints may be matched on one of these instead of the primary identity; upserts always use identity. Soft uniqueness.') class-attribute instance-attribute

secondary_identity_names property

Names of declared secondary identities, in declaration order.

semantics = PydanticField(default=None, description='Optional external-vocabulary anchors for this vertex type.') class-attribute instance-attribute

Methods:

convert_hash_identity_properties(v) classmethod

Source code in graflo/architecture/schema/vertex.py

@field_validator("hash_identity_properties", mode="before")
@classmethod
def convert_hash_identity_properties(cls, v: Any) -> Any:
    if v is None:
        return []
    if isinstance(v, tuple):
        return list(v)
    if isinstance(v, list):
        return v
    raise ValueError("hash_identity_properties must be a list[str]")

convert_identity(v) classmethod

Source code in graflo/architecture/schema/vertex.py

@field_validator("identity", mode="before")
@classmethod
def convert_identity(cls, v: Any) -> Any:
    if v is None:
        return []
    if isinstance(v, tuple):
        return list(v)
    if isinstance(v, list):
        return v
    raise ValueError("identity must be a list[str]")

convert_to_expressions(v) classmethod

Source code in graflo/architecture/schema/vertex.py

@field_validator("filters", mode="before")
@classmethod
def convert_to_expressions(cls, v: Any) -> Any:
    if not isinstance(v, list):
        return v
    result: list[FilterExpression] = []
    for item in v:
        if isinstance(item, FilterExpression):
            result.append(item)
        elif isinstance(item, (dict, list)):
            from graflo.filter.onto import parse_filter_expression

            result.append(parse_filter_expression(item))
        else:
            raise ValueError(
                "each filter must be a FilterExpression instance or a dict/list (parsed as FilterExpression)"
            )
    return result

convert_to_properties(v) classmethod

Source code in graflo/architecture/schema/vertex.py

@field_validator("properties", mode="before")
@classmethod
def convert_to_properties(cls, v: Any) -> Any:
    if not isinstance(v, list):
        raise ValueError("properties must be a list")
    return [_normalize_fields_item(item) for item in v]

finish_init()

Complete logical initialization for vertex.

Source code in graflo/architecture/schema/vertex.py

def finish_init(self):
    """Complete logical initialization for vertex."""
    return

get_properties()

Source code in graflo/architecture/schema/vertex.py

def get_properties(self) -> list[Field]:
    return self.properties

secondary_identity(selector)

Resolve selector to a declared secondary identity.

Accepts a declared name, an explicit field list equal to a declared field-set, or the literal "secondary" when exactly one is declared.

Source code in graflo/architecture/schema/vertex.py

def secondary_identity(self, selector: str | list[str]) -> SecondaryIdentity | None:
    """Resolve *selector* to a declared secondary identity.

    Accepts a declared name, an explicit field list equal to a declared
    field-set, or the literal ``"secondary"`` when exactly one is declared.
    """
    if not self.secondary_identities:
        return None
    if isinstance(selector, str):
        if selector == SECONDARY_IDENTITY_SUGAR:
            if len(self.secondary_identities) == 1:
                return self.secondary_identities[0]
            return None
        for entry in self.secondary_identities:
            if entry.name == selector:
                return entry
        return None
    wanted = frozenset(selector)
    for entry in self.secondary_identities:
        if entry.field_set == wanted:
            return entry
    return None

set_identity()

Source code in graflo/architecture/schema/vertex.py

@model_validator(mode="after")
def set_identity(self) -> Vertex:
    if self.blank and self.assigned:
        raise ValueError(
            f"Vertex '{self.name}': blank and assigned are mutually exclusive"
        )
    if self.assigned and self.hash_identity_properties:
        raise ValueError(
            f"Vertex '{self.name}': assigned and hash_identity_properties "
            "are mutually exclusive"
        )
    if self.blank and self.hash_identity_properties:
        raise ValueError(
            f"Vertex '{self.name}': blank and hash_identity_properties are "
            "mutually exclusive — identity_mode reads blank first, so the "
            "vertex would key on a generated id and never consult the digest"
        )
    if self.identity_funnel is not None:
        if self.hash_identity_properties:
            raise ValueError(
                f"Vertex '{self.name}': hash_identity_properties and "
                "identity_funnel are mutually exclusive — a funnel with one "
                "branch is the general form of a flat hash key"
            )
        if self.blank:
            raise ValueError(
                f"Vertex '{self.name}': blank and identity_funnel are "
                "mutually exclusive — a blank vertex has no source fields "
                "to digest"
            )
        if self.assigned:
            raise ValueError(
                f"Vertex '{self.name}': assigned and identity_funnel are "
                "mutually exclusive"
            )
    merged_properties = union_field_lists(
        self.properties, owner=f"vertex {self.name!r}"
    )
    identity_names = _dedupe_ordered(list(self.identity))
    hash_identity_names = _dedupe_ordered(list(self.hash_identity_properties))
    funnel_field_names = (
        self.identity_funnel.field_names if self.identity_funnel else []
    )
    list_prop_names = {
        f.name for f in merged_properties if is_list_field_type(f.type)
    }
    for name in identity_names:
        if name in list_prop_names:
            raise ValueError(
                f"Vertex '{self.name}': LIST-typed property '{name}' "
                "cannot be used as identity"
            )
    for name in hash_identity_names:
        if name in list_prop_names:
            raise ValueError(
                f"Vertex '{self.name}': LIST-typed property '{name}' "
                "cannot be used in hash_identity_properties"
            )
    for name in funnel_field_names:
        if name in list_prop_names:
            raise ValueError(
                f"Vertex '{self.name}': LIST-typed property '{name}' "
                "cannot be used in identity_funnel"
            )
    seen_names = {f.name for f in merged_properties}
    augmented = list(merged_properties)
    for name in identity_names + hash_identity_names + funnel_field_names:
        if name not in seen_names:
            synth_type = FieldType.UUID if self.assigned and name == "id" else None
            augmented.append(Field(name=name, type=synth_type))
            seen_names.add(name)

    secondary = self._validated_secondary_identities(
        identity_names=identity_names,
        list_prop_names=list_prop_names,
    )
    for entry in secondary:
        for name in entry.fields:
            if name not in seen_names:
                augmented.append(Field(name=name, type=None))
                seen_names.add(name)

    object.__setattr__(self, "identity", identity_names)
    object.__setattr__(self, "hash_identity_properties", hash_identity_names)
    object.__setattr__(self, "secondary_identities", secondary)
    object.__setattr__(self, "properties", augmented)
    return self

VertexConfig

Bases: ConfigBaseModel

Configuration for managing vertices.

This class manages vertices, providing methods for accessing and manipulating vertex configurations.

Attributes:

Name	Type	Description
vertices	list[Vertex]	List of vertex configurations
force_types	dict[str, list]	Dictionary mapping vertex names to type lists

Source code in graflo/architecture/schema/vertex.py

class VertexConfig(ConfigBaseModel):
    """Configuration for managing vertices.

    This class manages vertices, providing methods for accessing
    and manipulating vertex configurations.

    Attributes:
        vertices: List of vertex configurations
        force_types: Dictionary mapping vertex names to type lists
    """

    # See Vertex: silently dropping unknown keys hides authoring mistakes.
    model_config = ConfigDict(extra="forbid")

    vertices: list[Vertex] = PydanticField(
        ...,
        description="List of vertex type definitions (name, properties, identity, filters).",
    )
    force_types: dict[str, list] = PydanticField(
        default_factory=dict,
        description="Override mapping: vertex name -> list of field type names for type inference.",
    )
    identity_from_all_properties: bool = PydanticField(
        default=True,
        description=(
            "When true, vertices without explicit identity fall back to all property names. "
            "When false, explicit identity is required except for blank or assigned vertices."
        ),
    )
    _vertices_map: dict[VertexName, Vertex] | None = PrivateAttr(default=None)
    _vertex_numeric_fields_map: dict[VertexName, object] | None = PrivateAttr(
        default=None
    )

    @model_validator(mode="after")
    def build_vertices_map(self) -> VertexConfig:
        # `vertices` is the serialized truth and `_vertices_map` the lookup truth.
        # Without this check a duplicate name lets the two disagree silently: the map
        # keeps the last definition, the shadowed one still round-trips through YAML,
        # and `vertex_set` reports a count that the list does not match.
        duplicates = sorted(
            name
            for name, count in Counter(v.name for v in self.vertices).items()
            if count > 1
        )
        if duplicates:
            raise ValueError(f"duplicate vertex names: {duplicates}")
        object.__setattr__(
            self,
            "_vertices_map",
            {item.name: item for item in self.vertices},
        )
        object.__setattr__(self, "_vertex_numeric_fields_map", {})
        self._normalize_vertex_identities()
        return self

    @property
    def blank_vertices(self) -> list[str]:
        """Vertex names marked blank (no natural identity; auto-generated ID)."""
        return [v.name for v in self.vertices if v.blank]

    @property
    def assigned_vertices(self) -> list[str]:
        """Vertex names with intentional UUID primary keys (``assigned: true``)."""
        return [v.name for v in self.vertices if v.assigned]

    @property
    def hash_identity_vertices(self) -> list[str]:
        """Vertex names using digest-derived synthetic identity (flat or funnel)."""
        return [
            v.name
            for v in self.vertices
            if v.hash_identity_properties or v.identity_funnel is not None
        ]

    @property
    def identity_funnel_vertices(self) -> list[str]:
        """Vertex names whose synthetic identity comes from a funnel."""
        return [v.name for v in self.vertices if v.identity_funnel is not None]

    def vertices_by_identity_mode(self, mode: IdentityMode) -> list[str]:
        """Vertex names whose resolved identity mode matches *mode*."""
        return [v.name for v in self.vertices if v.identity_mode == mode]

    def _normalize_vertex_identities(
        self,
    ) -> None:
        blank_id_field = "id"
        for vertex in self.vertices:
            if not vertex.identity:
                if (
                    vertex.hash_identity_properties
                    or vertex.identity_funnel is not None
                    or vertex.blank
                    or vertex.assigned
                ):
                    vertex.identity = [blank_id_field]
                elif self.identity_from_all_properties:
                    vertex.identity = list(vertex.property_names)
                else:
                    raise ValueError(
                        f"Vertex '{vertex.name}' must define identity fields"
                    )
            vertex.identity = _dedupe_ordered(list(vertex.identity))
            vertex.hash_identity_properties = _dedupe_ordered(
                list(vertex.hash_identity_properties)
            )
            missing = _dedupe_ordered(
                [
                    field_name
                    for field_name in vertex.identity
                    + vertex.hash_identity_properties
                    + vertex.digest_source_fields
                    if field_name not in vertex.property_names
                ]
            )
            for field_name in missing:
                synth_type = (
                    FieldType.UUID
                    if vertex.assigned and field_name == blank_id_field
                    else None
                )
                vertex.properties.append(Field(name=field_name, type=synth_type))

    def _get_vertices_map(self) -> dict[VertexName, Vertex]:
        """Return the vertices map (set by model validator)."""
        if self._vertices_map is None:
            raise RuntimeError("VertexConfig not fully initialized")
        return self._vertices_map

    @property
    def vertex_set(self):
        """Get set of vertex names.

        Returns:
            set[str]: Set of vertex names
        """
        return set(self._get_vertices_map().keys())

    @property
    def vertex_list(self):
        """Get list of vertex configurations.

        Returns:
            list[Vertex]: List of vertex configurations
        """
        return list(self._get_vertices_map().values())

    def _get_vertex_by_name(self, identifier: VertexName) -> Vertex:
        """Get vertex by logical vertex name."""
        m = self._get_vertices_map()
        if identifier in m:
            return m[identifier]
        available_names = list(m.keys())
        raise KeyError(
            f"Vertex '{identifier}' not found by logical name. "
            f"Available names: {available_names}"
        )

    def identity_fields(self, vertex_name: VertexName) -> list[str]:
        """Get identity fields for a vertex."""
        return list(self._get_vertices_map()[vertex_name].identity)

    def secondary_identities(self, vertex_name: VertexName) -> list[SecondaryIdentity]:
        """Declared secondary identities for a vertex."""
        return list(self._get_vertex_by_name(vertex_name).secondary_identities)

    def secondary_identity_fields(
        self, vertex_name: VertexName, selector: str | list[str]
    ) -> list[str] | None:
        """Resolve an edge-step selector to a secondary identity field-set.

        Returns ``None`` when *selector* names no declared secondary identity,
        letting callers raise with their own context.
        """
        entry = self._get_vertex_by_name(vertex_name).secondary_identity(selector)
        return list(entry.fields) if entry is not None else None

    def match_fields(
        self, vertex_name: VertexName, selector: str | list[str] | None
    ) -> list[str]:
        """Fields an edge endpoint is matched on for *selector*.

        ``None`` or ``"identity"`` selects the primary identity, which keeps
        every existing edge step on exactly the path it uses today.

        Raises:
            ValueError: if *selector* names no declared secondary identity.
        """
        if selector is None or selector == PRIMARY_IDENTITY_SELECTOR:
            return self.identity_fields(vertex_name)
        fields = self.secondary_identity_fields(vertex_name, selector)
        if fields is None:
            declared = self._get_vertex_by_name(vertex_name).secondary_identity_names
            raise ValueError(
                f"Vertex '{vertex_name}': no secondary identity matches selector "
                f"{selector!r}. Declared: {declared or '(none)'}"
            )
        return fields

    def properties(self, vertex_name: VertexName) -> list[Field]:
        """Vertex properties as Field objects."""

        vertex = self._get_vertex_by_name(vertex_name)

        return vertex.properties

    def property_names(
        self,
        vertex_name: VertexName,
    ) -> list[str]:
        """Vertex property names as strings."""

        vertex = self._get_vertex_by_name(vertex_name)
        return vertex.property_names

    def numeric_fields_list(self, vertex_name):
        """Get list of numeric fields for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            tuple: Tuple of numeric field names

        Raises:
            ValueError: If vertex is not defined in config
        """
        if vertex_name in self.vertex_set:
            nmap = self._vertex_numeric_fields_map
            if nmap is not None and vertex_name in nmap:
                return nmap[vertex_name]
            else:
                return ()
        else:
            raise ValueError(
                " Accessing vertex numeric fields: vertex"
                f" {vertex_name} was not defined in config"
            )

    def filters(self, vertex_name) -> list[FilterExpression]:
        """Get filter clauses for a vertex.

        Args:
            vertex_name: Name of the vertex

        Returns:
            list[FilterExpression]: List of filter expressions
        """
        m = self._get_vertices_map()
        if vertex_name in m:
            return m[vertex_name].filters
        else:
            return []

    def remove_vertices(self, names: set[str]) -> None:
        """Remove vertices by name.

        Removes vertices from the configuration. Mutates the instance in place.

        Args:
            names: Set of vertex names to remove
        """
        if not names:
            return
        self.vertices[:] = [v for v in self.vertices if v.name not in names]
        m = self._get_vertices_map()
        for n in names:
            m.pop(n, None)

    def update_vertex(self, v: Vertex):
        """Update vertex configuration.

        Args:
            v: Vertex configuration to update
        """
        self._get_vertices_map()[v.name] = v

    def __getitem__(self, key: str):
        """Get vertex configuration by name.

        Args:
            key: Vertex name

        Returns:
            Vertex: Vertex configuration

        Raises:
            KeyError: If vertex is not found
        """
        m = self._get_vertices_map()
        if key in m:
            return m[key]
        else:
            raise KeyError(f"Vertex {key} absent")

    def __setitem__(self, key: str, value: Vertex):
        """Set vertex configuration by name.

        Args:
            key: Vertex name
            value: Vertex configuration
        """
        self._get_vertices_map()[key] = value

    def finish_init(self):
        """Complete logical initialization of vertices."""
        self._normalize_vertex_identities()
        for v in self.vertices:
            v.finish_init()

Attributes

assigned_vertices property

Vertex names with intentional UUID primary keys (assigned: true).

blank_vertices property

Vertex names marked blank (no natural identity; auto-generated ID).

force_types = PydanticField(default_factory=dict, description='Override mapping: vertex name -> list of field type names for type inference.') class-attribute instance-attribute

hash_identity_vertices property

Vertex names using digest-derived synthetic identity (flat or funnel).

identity_from_all_properties = PydanticField(default=True, description='When true, vertices without explicit identity fall back to all property names. When false, explicit identity is required except for blank or assigned vertices.') class-attribute instance-attribute

identity_funnel_vertices property

Vertex names whose synthetic identity comes from a funnel.

model_config = ConfigDict(extra='forbid') class-attribute instance-attribute

vertex_list property

Get list of vertex configurations.

Returns:

Type	Description
	list[Vertex]: List of vertex configurations

vertex_set property

Get set of vertex names.

Returns:

Type	Description
	set[str]: Set of vertex names

vertices = PydanticField(..., description='List of vertex type definitions (name, properties, identity, filters).') class-attribute instance-attribute

Methods:

__getitem__(key)

Get vertex configuration by name.

Parameters:

Name	Type	Description	Default
key	str	Vertex name	required

Returns:

Name	Type	Description
Vertex		Vertex configuration

Raises:

Type	Description
KeyError	If vertex is not found

Source code in graflo/architecture/schema/vertex.py

def __getitem__(self, key: str):
    """Get vertex configuration by name.

    Args:
        key: Vertex name

    Returns:
        Vertex: Vertex configuration

    Raises:
        KeyError: If vertex is not found
    """
    m = self._get_vertices_map()
    if key in m:
        return m[key]
    else:
        raise KeyError(f"Vertex {key} absent")

__setitem__(key, value)

Set vertex configuration by name.

Parameters:

Name	Type	Description	Default
key	str	Vertex name	required
value	Vertex	Vertex configuration	required

Source code in graflo/architecture/schema/vertex.py

def __setitem__(self, key: str, value: Vertex):
    """Set vertex configuration by name.

    Args:
        key: Vertex name
        value: Vertex configuration
    """
    self._get_vertices_map()[key] = value

build_vertices_map()

Source code in graflo/architecture/schema/vertex.py

@model_validator(mode="after")
def build_vertices_map(self) -> VertexConfig:
    # `vertices` is the serialized truth and `_vertices_map` the lookup truth.
    # Without this check a duplicate name lets the two disagree silently: the map
    # keeps the last definition, the shadowed one still round-trips through YAML,
    # and `vertex_set` reports a count that the list does not match.
    duplicates = sorted(
        name
        for name, count in Counter(v.name for v in self.vertices).items()
        if count > 1
    )
    if duplicates:
        raise ValueError(f"duplicate vertex names: {duplicates}")
    object.__setattr__(
        self,
        "_vertices_map",
        {item.name: item for item in self.vertices},
    )
    object.__setattr__(self, "_vertex_numeric_fields_map", {})
    self._normalize_vertex_identities()
    return self

filters(vertex_name)

Get filter clauses for a vertex.

Parameters:

Name	Type	Description	Default
vertex_name		Name of the vertex	required

Returns:

Type	Description
list[FilterExpression]	list[FilterExpression]: List of filter expressions

Source code in graflo/architecture/schema/vertex.py

def filters(self, vertex_name) -> list[FilterExpression]:
    """Get filter clauses for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        list[FilterExpression]: List of filter expressions
    """
    m = self._get_vertices_map()
    if vertex_name in m:
        return m[vertex_name].filters
    else:
        return []

finish_init()

Complete logical initialization of vertices.

Source code in graflo/architecture/schema/vertex.py

def finish_init(self):
    """Complete logical initialization of vertices."""
    self._normalize_vertex_identities()
    for v in self.vertices:
        v.finish_init()

identity_fields(vertex_name)

Get identity fields for a vertex.

Source code in graflo/architecture/schema/vertex.py

def identity_fields(self, vertex_name: VertexName) -> list[str]:
    """Get identity fields for a vertex."""
    return list(self._get_vertices_map()[vertex_name].identity)

match_fields(vertex_name, selector)

Fields an edge endpoint is matched on for selector.

None or "identity" selects the primary identity, which keeps every existing edge step on exactly the path it uses today.

Raises:

Type	Description
ValueError	if selector names no declared secondary identity.

Source code in graflo/architecture/schema/vertex.py

def match_fields(
    self, vertex_name: VertexName, selector: str | list[str] | None
) -> list[str]:
    """Fields an edge endpoint is matched on for *selector*.

    ``None`` or ``"identity"`` selects the primary identity, which keeps
    every existing edge step on exactly the path it uses today.

    Raises:
        ValueError: if *selector* names no declared secondary identity.
    """
    if selector is None or selector == PRIMARY_IDENTITY_SELECTOR:
        return self.identity_fields(vertex_name)
    fields = self.secondary_identity_fields(vertex_name, selector)
    if fields is None:
        declared = self._get_vertex_by_name(vertex_name).secondary_identity_names
        raise ValueError(
            f"Vertex '{vertex_name}': no secondary identity matches selector "
            f"{selector!r}. Declared: {declared or '(none)'}"
        )
    return fields

numeric_fields_list(vertex_name)

Get list of numeric fields for a vertex.

Parameters:

Name	Type	Description	Default
vertex_name		Name of the vertex	required

Returns:

Name	Type	Description
tuple		Tuple of numeric field names

Raises:

Type	Description
ValueError	If vertex is not defined in config

Source code in graflo/architecture/schema/vertex.py

def numeric_fields_list(self, vertex_name):
    """Get list of numeric fields for a vertex.

    Args:
        vertex_name: Name of the vertex

    Returns:
        tuple: Tuple of numeric field names

    Raises:
        ValueError: If vertex is not defined in config
    """
    if vertex_name in self.vertex_set:
        nmap = self._vertex_numeric_fields_map
        if nmap is not None and vertex_name in nmap:
            return nmap[vertex_name]
        else:
            return ()
    else:
        raise ValueError(
            " Accessing vertex numeric fields: vertex"
            f" {vertex_name} was not defined in config"
        )

properties(vertex_name)

Vertex properties as Field objects.

Source code in graflo/architecture/schema/vertex.py

def properties(self, vertex_name: VertexName) -> list[Field]:
    """Vertex properties as Field objects."""

    vertex = self._get_vertex_by_name(vertex_name)

    return vertex.properties

property_names(vertex_name)

Vertex property names as strings.

Source code in graflo/architecture/schema/vertex.py

def property_names(
    self,
    vertex_name: VertexName,
) -> list[str]:
    """Vertex property names as strings."""

    vertex = self._get_vertex_by_name(vertex_name)
    return vertex.property_names

remove_vertices(names)

Remove vertices by name.

Removes vertices from the configuration. Mutates the instance in place.

Parameters:

Name	Type	Description	Default
names	set[str]	Set of vertex names to remove	required

Source code in graflo/architecture/schema/vertex.py

def remove_vertices(self, names: set[str]) -> None:
    """Remove vertices by name.

    Removes vertices from the configuration. Mutates the instance in place.

    Args:
        names: Set of vertex names to remove
    """
    if not names:
        return
    self.vertices[:] = [v for v in self.vertices if v.name not in names]
    m = self._get_vertices_map()
    for n in names:
        m.pop(n, None)

secondary_identities(vertex_name)

Declared secondary identities for a vertex.

Source code in graflo/architecture/schema/vertex.py

def secondary_identities(self, vertex_name: VertexName) -> list[SecondaryIdentity]:
    """Declared secondary identities for a vertex."""
    return list(self._get_vertex_by_name(vertex_name).secondary_identities)

secondary_identity_fields(vertex_name, selector)

Resolve an edge-step selector to a secondary identity field-set.

Returns None when selector names no declared secondary identity, letting callers raise with their own context.

Source code in graflo/architecture/schema/vertex.py

def secondary_identity_fields(
    self, vertex_name: VertexName, selector: str | list[str]
) -> list[str] | None:
    """Resolve an edge-step selector to a secondary identity field-set.

    Returns ``None`` when *selector* names no declared secondary identity,
    letting callers raise with their own context.
    """
    entry = self._get_vertex_by_name(vertex_name).secondary_identity(selector)
    return list(entry.fields) if entry is not None else None

update_vertex(v)

Update vertex configuration.

Parameters:

Name	Type	Description	Default
v	Vertex	Vertex configuration to update	required

Source code in graflo/architecture/schema/vertex.py

def update_vertex(self, v: Vertex):
    """Update vertex configuration.

    Args:
        v: Vertex configuration to update
    """
    self._get_vertices_map()[v.name] = v

vertices_by_identity_mode(mode)

Vertex names whose resolved identity mode matches mode.

Source code in graflo/architecture/schema/vertex.py

def vertices_by_identity_mode(self, mode: IdentityMode) -> list[str]:
    """Vertex names whose resolved identity mode matches *mode*."""
    return [v.name for v in self.vertices if v.identity_mode == mode]

Functions:

field_type_value(ft)

Normalize a FieldType / string / None to an uppercase type string.

Source code in graflo/architecture/schema/vertex.py

def field_type_value(ft: FieldType | str | None) -> str | None:
    """Normalize a FieldType / string / None to an uppercase type string."""
    if ft is None:
        return None
    if isinstance(ft, FieldType):
        return ft.value
    return str(ft).upper()

format_field_type_label(field)

Human-readable type label, e.g. LIST<STRING>, INT or untyped.

Source code in graflo/architecture/schema/vertex.py

def format_field_type_label(field: Field) -> str:
    """Human-readable type label, e.g. ``LIST<STRING>``, ``INT`` or ``untyped``."""
    type_val = field_type_value(field.type)
    if type_val is None:
        return "untyped"
    if type_val == FieldType.LIST.value:
        item_val = field_type_value(field.item_type) or "?"
        return f"LIST<{item_val}>"
    return type_val

is_list_field_type(ft)

Source code in graflo/architecture/schema/vertex.py

def is_list_field_type(ft: FieldType | str | None) -> bool:
    return field_type_value(ft) == FieldType.LIST.value

merge_fields(a, b, *, owner)

Merge two same-named fields, refusing a genuine disagreement.

type and item_type are compared and carried as a unit: LIST is only half a type, so electing a type without the item_type that came with it yields a field that cannot be constructed. One untyped side gives way to the other's pair whole; two typed sides that disagree raise, because widening (INT + FLOAT -> DOUBLE) would elect a type neither author wrote.

Descriptions from both sides survive and grounding folds through :func:~graflo.architecture.schema.semantics.merge_field_semantics. Nothing here is decided by which side was seen first.

owner is a rendered label -- vertex 'party', edge ('order', 'invoice', 'places') -- so a merge kernel keyed by into name and a model validator keyed by self.name raise the same sentence.

Source code in graflo/architecture/schema/vertex.py

def merge_fields(a: Field, b: Field, *, owner: str) -> Field:
    """Merge two same-named fields, refusing a genuine disagreement.

    ``type`` and ``item_type`` are compared **and carried as a unit**: ``LIST``
    is only half a type, so electing a ``type`` without the ``item_type`` that
    came with it yields a field that cannot be constructed. One untyped side
    gives way to the other's pair whole; two typed sides that disagree raise,
    because widening (``INT`` + ``FLOAT`` -> ``DOUBLE``) would elect a type
    neither author wrote.

    Descriptions from both sides survive and grounding folds through
    :func:`~graflo.architecture.schema.semantics.merge_field_semantics`. Nothing
    here is decided by which side was seen first.

    ``owner`` is a rendered label -- ``vertex 'party'``, ``edge ('order',
    'invoice', 'places')`` -- so a merge kernel keyed by ``into`` name and a
    model validator keyed by ``self.name`` raise the same sentence.
    """
    if (a.type, a.item_type) != (b.type, b.item_type):
        if a.type is not None and b.type is not None:
            raise FieldMergeConflict(
                owner,
                f"property {a.name!r}: {format_field_type_label(a)!r} vs "
                f"{format_field_type_label(b)!r}",
                _RETYPE_REMEDY,
                field=a.name,
            )
        # Exactly one side is typed: its (type, item_type) pair carries whole.
        base, other = (a, b) if a.type is not None else (b, a)
    else:
        base, other = a, b

    try:
        semantics = merge_field_semantics(
            base.semantics, other.semantics, owner=owner, field=a.name
        )
    except ValueError:
        left_unit = a.semantics.unit if a.semantics else None
        right_unit = b.semantics.unit if b.semantics else None
        raise FieldMergeConflict(
            owner,
            f"property {a.name!r}: units {left_unit!r} vs {right_unit!r}",
            _UNIT_REMEDY,
            field=a.name,
        ) from None

    # Rebuilt from the whole authored field rather than an enumerated
    # constructor: every key Field grows is carried by construction, and
    # ``validate_list_item_type`` re-runs here rather than three frames later.
    return Field.model_validate(
        {
            **base.to_dict(skip_defaults=False),
            "description": _fold_field_description(base.description, other.description),
            "semantics": semantics,
        }
    )

union_field_lists(fields, *, owner)

Fold same-named fields into one, preserving first-declaration order.

Every conflicting property is reported together: merging two large schemas one error per run makes the author re-run the merge to discover the next disagreement, when the merge already knows all of them.

Source code in graflo/architecture/schema/vertex.py

def union_field_lists(fields: Iterable[Field], *, owner: str) -> list[Field]:
    """Fold same-named fields into one, preserving first-declaration order.

    Every conflicting property is reported together: merging two large schemas
    one error per run makes the author re-run the merge to discover the next
    disagreement, when the merge already knows all of them.
    """
    merged: dict[str, Field] = {}
    order: list[str] = []
    conflicts: list[FieldMergeConflict] = []

    for field in fields:
        existing = merged.get(field.name)
        if existing is None:
            merged[field.name] = field
            order.append(field.name)
            continue
        try:
            merged[field.name] = merge_fields(existing, field, owner=owner)
        except FieldMergeConflict as conflict:
            # Collected, not raised: a clash on one property must not hide the
            # next one from an author who has to fix them all anyway.
            conflicts.append(conflict)

    if conflicts:
        reasons = [conflict.reason for conflict in conflicts]
        heading = _conflict_heading(reasons, owner)
        remedies = list(dict.fromkeys(conflict.remedy for conflict in conflicts))
        if len(conflicts) == 1:
            message = f"{heading}, {reasons[0]}. {remedies[0]}"
        else:
            listed = "\n".join(f"  {reason}" for reason in reasons)
            message = f"{heading}:\n{listed}\n" + " ".join(remedies)
        raise FieldMergeError(message, owner=owner, conflicts=tuple(conflicts))

    return [merged[name] for name in order]