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]

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)

__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

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"

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)

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.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 = _merge_duplicate_fields(self.name, list(self.properties))
        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

digest_source_fields property

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

has_identity_funnel property

True when identity is derived from ordered funnel branches.

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.

property_names property

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

secondary_identity_names property

Names of declared secondary identities, in declaration order.

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

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

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()

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

hash_identity_vertices property

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

identity_funnel_vertices property

Vertex names whose synthetic identity comes from a funnel.

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

__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

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]

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> or INT.

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

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