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

DerivedForm = Literal['normal', 'keys'] module-attribute

FieldMerge = Literal['replace', 'union'] module-attribute

FieldSetBy = Literal['stamp', 'sync'] module-attribute

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

PartsOrder = Literal['general_first', 'specific_first', 'unordered'] 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.",
    )
    # Both flags are left out of a dump at their defaults, so a manifest that
    # uses neither dumps exactly as before they existed.
    reserved: bool = PydanticField(
        default=False,
        exclude_if=lambda value: value is False,
        description=(
            "Set only by the writer, from the caller's stamp. A reserved field is "
            "left out of type sheets, element text and instance schemas, and a "
            "query that names it in a projection, filter or aggregate is refused."
        ),
    )
    set_by: FieldSetBy = PydanticField(
        default="stamp",
        exclude_if=lambda value: value == "stamp",
        description=(
            "Who writes a reserved field: ``stamp``, the writer from the caller's "
            "stamp; or ``sync``, the bulk and sync writers from the record's sync "
            "hash, which an upsert sets to null."
        ),
    )
    merge: FieldMerge = PydanticField(
        default="replace",
        exclude_if=lambda value: value == "replace",
        description=(
            "How an upsert combines a value with the stored one: ``replace`` "
            "overwrites it; ``union`` keeps every member of both lists "
            "(LIST fields only)."
        ),
    )
    # Left out of a dump when absent, as the flags above.
    match: FieldMatch | None = PydanticField(
        default=None,
        exclude_if=lambda value: value is None,
        description="How a stated value is compared with a stored one.",
    )
    derived: FieldDerived | None = PydanticField(
        default=None,
        exclude_if=lambda value: value is None,
        description=(
            "Computed by core from another field of the vertex, before every write "
            "and digest; never supplied by a document."
        ),
    )

    @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}'"
            )
        if self.merge == "union" and type_val != FieldType.LIST.value:
            raise ValueError(
                f"Field '{self.name}': merge 'union' needs type LIST, got "
                f"{type_val or 'untyped'}"
            )
        if self.set_by == "sync" and (not self.reserved or self.merge == "union"):
            raise ValueError(
                f"Field '{self.name}': set_by 'sync' needs a reserved field that "
                "merges by replace"
            )
        if self.derived is not None:
            self._check_derived(self.derived, type_val, item_val)
        return self

    def _check_derived(
        self, derived: FieldDerived, type_val: str | None, item_val: str | None
    ) -> None:
        clashes = [
            what
            for what, present in (
                ("match", self.match is not None),
                ("reserved", self.reserved),
                ("merge 'union'", self.merge == "union"),
            )
            if present
        ]
        if clashes:
            # A derived field is computed, never compared or stamped: its source is.
            raise ValueError(
                f"Field '{self.name}': a derived field cannot also declare "
                f"{', '.join(clashes)}"
            )
        if derived.source == self.name:
            raise ValueError(f"Field '{self.name}': derived from itself")
        if derived.form == "keys" and (type_val, item_val) != ("LIST", "STRING"):
            raise ValueError(
                f"Field '{self.name}': derived keys need type LIST, item_type STRING"
            )
        if derived.form == "normal" and type_val not in (None, "STRING"):
            raise ValueError(
                f"Field '{self.name}': a derived normal form is a STRING, got {type_val}"
            )

    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

derived = PydanticField(default=None, exclude_if=lambda value: value is None, description='Computed by core from another field of the vertex, before every write and digest; never supplied by a document.') class-attribute instance-attribute

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

match = PydanticField(default=None, exclude_if=lambda value: value is None, description='How a stated value is compared with a stored one.') class-attribute instance-attribute

merge = PydanticField(default='replace', exclude_if=lambda value: value == 'replace', description='How an upsert combines a value with the stored one: ``replace`` overwrites it; ``union`` keeps every member of both lists (LIST fields only).') 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

reserved = PydanticField(default=False, exclude_if=lambda value: value is False, description="Set only by the writer, from the caller's stamp. A reserved field is left out of type sheets, element text and instance schemas, and a query that names it in a projection, filter or aggregate is refused.") class-attribute instance-attribute

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

set_by = PydanticField(default='stamp', exclude_if=lambda value: value == 'stamp', description="Who writes a reserved field: ``stamp``, the writer from the caller's stamp; or ``sync``, the bulk and sync writers from the record's sync hash, which an upsert sets to null.") 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}'"
        )
    if self.merge == "union" and type_val != FieldType.LIST.value:
        raise ValueError(
            f"Field '{self.name}': merge 'union' needs type LIST, got "
            f"{type_val or 'untyped'}"
        )
    if self.set_by == "sync" and (not self.reserved or self.merge == "union"):
        raise ValueError(
            f"Field '{self.name}': set_by 'sync' needs a reserved field that "
            "merges by replace"
        )
    if self.derived is not None:
        self._check_derived(self.derived, type_val, item_val)
    return self

FieldDerived

Bases: ConfigBaseModel

A field core computes from another field of the same vertex.

Stored, never supplied: core overwrites a stated value, and a type sheet never shows the field.

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

class FieldDerived(ConfigBaseModel):
    """A field core computes from another field of the same vertex.

    Stored, never supplied: core overwrites a stated value, and a type sheet
    never shows the field.
    """

    source: str = PydanticField(..., description="The field it is computed from.")
    form: DerivedForm = PydanticField(
        ...,
        description=(
            "``normal``: the source's normal form under its ``match.normalizer``. "
            "``keys``: every less specific form of the source under its "
            "``match.parts`` (a LIST of strings)."
        ),
    )

Attributes

form = PydanticField(..., description="``normal``: the source's normal form under its ``match.normalizer``. ``keys``: every less specific form of the source under its ``match.parts`` (a LIST of strings).") class-attribute instance-attribute

source = PydanticField(..., description='The field it is computed from.') class-attribute instance-attribute

FieldMatch

Bases: ConfigBaseModel

How a stated value of the field is compared with a stored one.

It declares the value's structure, never its meaning. See :mod:graflo.architecture.schema.compare for the levels it yields.

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

class FieldMatch(ConfigBaseModel):
    """How a stated value of the field is compared with a stored one.

    It declares the value's structure, never its meaning. See
    :mod:`graflo.architecture.schema.compare` for the levels it yields.
    """

    normalizer: str | None = PydanticField(
        default=None,
        exclude_if=lambda value: value is None,
        description=(
            "A registered scalar normalizer (``graflo.util.transform.NORMALIZERS``), "
            "applied to the whole value and to each part. Without one, values "
            "compare after ``coerce_value`` alone."
        ),
    )
    parts: FieldParts | None = PydanticField(
        default=None,
        exclude_if=lambda value: value is None,
        description="How the value splits into parts; makes ``compatible`` possible.",
    )
    exclusive: bool | None = PydanticField(
        default=None,
        exclude_if=lambda value: value is None,
        description=(
            "Whether two different values are evidence against a match "
            "(``conflict``) rather than ``differs``. Unset: true on a key or "
            "alternate-key field, false otherwise."
        ),
    )

    @field_validator("normalizer")
    @classmethod
    def _registered(cls, v: str | None) -> str | None:
        if v is not None and v not in NORMALIZERS:
            raise ValueError(
                f"unknown normalizer {v!r}; registered: {', '.join(sorted(NORMALIZERS))}"
            )
        return v

Attributes

exclusive = PydanticField(default=None, exclude_if=lambda value: value is None, description='Whether two different values are evidence against a match (``conflict``) rather than ``differs``. Unset: true on a key or alternate-key field, false otherwise.') class-attribute instance-attribute

normalizer = PydanticField(default=None, exclude_if=lambda value: value is None, description='A registered scalar normalizer (``graflo.util.transform.NORMALIZERS``), applied to the whole value and to each part. Without one, values compare after ``coerce_value`` alone.') class-attribute instance-attribute

parts = PydanticField(default=None, exclude_if=lambda value: value is None, description='How the value splits into parts; makes ``compatible`` possible.') class-attribute instance-attribute

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. origins maps each type label to the declarations that carry it, when the caller knows them (a merge does; a single model validator does not).

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.
    ``origins`` maps each type label to the declarations that carry it, when
    the caller knows them (a merge does; a single model validator does not).
    """

    def __init__(
        self,
        owner: str,
        reason: str,
        remedy: str,
        *,
        field: str,
        origins: Mapping[str, Sequence[str]] | None = None,
    ) -> None:
        self.reason = reason
        self.remedy = remedy
        self.owner = owner
        self.field = field
        self.origins = {label: tuple(o) for label, o in (origins or {}).items()}
        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

origins = {label: tuple(o) for label, o in (origins or {}).items()} instance-attribute

owner = owner instance-attribute

reason = reason instance-attribute

remedy = remedy instance-attribute

Methods:

__init__(owner, reason, remedy, *, field, origins=None)

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

def __init__(
    self,
    owner: str,
    reason: str,
    remedy: str,
    *,
    field: str,
    origins: Mapping[str, Sequence[str]] | None = None,
) -> None:
    self.reason = reason
    self.remedy = remedy
    self.owner = owner
    self.field = field
    self.origins = {label: tuple(o) for label, o in (origins or {}).items()}
    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])

FieldParts

Bases: ConfigBaseModel

How a value splits into parts, and which end of it is the general one.

A less specific form of a value is a run of its parts that keeps the general end: "Berlin" of "Berlin, pump hall 2" (general_first), "example.com" of "eu.example.com" (specific_first). unordered parts have no such run: one value is less specific when its parts are a subset of the other's.

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

class FieldParts(ConfigBaseModel):
    """How a value splits into parts, and which end of it is the general one.

    A less specific form of a value is a run of its parts that keeps the general
    end: "Berlin" of "Berlin, pump hall 2" (``general_first``), "example.com" of
    "eu.example.com" (``specific_first``). ``unordered`` parts have no such run:
    one value is less specific when its parts are a subset of the other's.
    """

    split: Literal["words"] | list[str] = PydanticField(
        ...,
        description=(
            "``words`` (whitespace, after the normalizer) or the separators "
            "between parts. Required: no separator is assumed."
        ),
    )
    order: PartsOrder = PydanticField(
        ...,
        description=(
            "Which end is the general one: ``general_first``, ``specific_first`` "
            "or ``unordered``. Required: no order is assumed."
        ),
    )

    @field_validator("split")
    @classmethod
    def _separators(
        cls, v: Literal["words"] | list[str]
    ) -> Literal["words"] | list[str]:
        if isinstance(v, list):
            if not v or any(not separator for separator in v):
                raise ValueError("split needs one or more non-empty separators")
            return _dedupe_ordered(v)
        return v

Attributes

order = PydanticField(..., description='Which end is the general one: ``general_first``, ``specific_first`` or ``unordered``. Required: no order is assumed.') class-attribute instance-attribute

split = PydanticField(..., description='``words`` (whitespace, after the normalizer) or the separators between parts. Required: no separator is assumed.') class-attribute instance-attribute

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

SuperClass

Bases: ConfigBaseModel

A class a vertex type is a subclass of (rdfs:subClassOf).

Either a vertex of the same schema, by name, or a class outside it, by IRI. A logical assertion: every instance of the subclass is an instance of the parent. Core checks the hierarchy and computes what it entails (:mod:graflo.architecture.schema.hierarchy); a subclass inherits no property, identity or storage from its parent.

Examples:

>>> SuperClass(vertex="Person")
>>> SuperClass(iri="http://xmlns.com/foaf/0.1/Agent")

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

class SuperClass(ConfigBaseModel):
    """A class a vertex type is a subclass of (``rdfs:subClassOf``).

    Either a vertex of the same schema, by name, or a class outside it, by IRI.
    A logical assertion: every instance of the subclass is an instance of the
    parent. Core checks the hierarchy and computes what it entails
    (:mod:`graflo.architecture.schema.hierarchy`); a subclass inherits no
    property, identity or storage from its parent.

    Examples:
        >>> SuperClass(vertex="Person")
        >>> SuperClass(iri="http://xmlns.com/foaf/0.1/Agent")
    """

    vertex: str | None = PydanticField(
        default=None, description="Name of a vertex in the same schema."
    )
    iri: str | None = PydanticField(
        default=None, description="Absolute IRI of a class outside the schema."
    )

    @model_validator(mode="after")
    def _one_parent(self) -> SuperClass:
        if (self.vertex is None) == (self.iri is None):
            raise ValueError("a superclass names exactly one of `vertex` or `iri`")
        if self.iri is not None:
            check_iri(self.iri, "superclass iri")
        return self

Attributes

iri = PydanticField(default=None, description='Absolute IRI of a class outside the schema.') class-attribute instance-attribute

vertex = PydanticField(default=None, description='Name of a vertex in the same schema.') class-attribute instance-attribute

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.",
    )
    subclass_of: list[SuperClass] = PydanticField(
        default_factory=list,
        description=(
            "Classes this vertex type is a subclass of (``rdfs:subClassOf``): a "
            "vertex of this schema by name, or an external class by IRI. A "
            "logical assertion, checked and entailed; nothing is inherited."
        ),
    )
    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"
                )
            overlap = sorted(set(self.identity) & set(self.identity_funnel.field_names))
            if overlap:
                # The cast drops a digest vertex's identity fields from each
                # record, so a branch reading one could never complete.
                raise ValueError(
                    f"Vertex '{self.name}': identity field(s) {overlap} are also "
                    "identity_funnel branch fields; the field the digest is "
                    "stored in must be one no branch reads"
                )
        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"
                )
        reserved_names = {f.name for f in merged_properties if f.reserved}
        keyed = sorted(
            reserved_names
            & {*identity_names, *hash_identity_names, *funnel_field_names}
        )
        if keyed:
            # The writer stamps a reserved field, so as a key it would fold
            # every vertex of one stamp into one.
            raise ValueError(
                f"Vertex '{self.name}': reserved property(ies) {keyed} cannot "
                "be part of its identity"
            )
        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)
        self._check_derived_fields(merged_properties, funnel=set(funnel_field_names))

        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 _check_derived_fields(
        self, properties: list[Field], *, funnel: set[str]
    ) -> None:
        """Each derived field reads a plain field of this vertex that says how.

        The normalizer and the split are read from the source's ``match``, never
        defaulted, so a derived field, its lookup and its comparison use one rule.
        """
        by_name = {f.name: f for f in properties}
        for field in properties:
            derived = field.derived
            if derived is None:
                continue
            at = f"Vertex '{self.name}': derived field '{field.name}'"
            source = by_name.get(derived.source)
            if source is None:
                raise ValueError(f"{at} reads '{derived.source}', not a property")
            if source.derived is not None or source.reserved:
                raise ValueError(
                    f"{at} reads '{source.name}', which is derived or reserved"
                )
            match = source.match
            if match is None or match.normalizer is None:
                raise ValueError(
                    f"{at}: its source '{source.name}' declares no match.normalizer"
                )
            if derived.form == "keys" and (
                match.parts is None or match.parts.order == "unordered"
            ):
                raise ValueError(
                    f"{at}: keys need ordered match.parts on '{source.name}'"
                )
            if field.name in funnel:
                raise ValueError(f"{at}: an identity_funnel branch cannot read it")

    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 derived_fields(self) -> list[Field]:
        """Fields core computes from another field, in declared order."""
        return [field for field in self.properties if field.derived is not None]

    @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

derived_fields property

Fields core computes from another field, in declared order.

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

subclass_of = PydanticField(default_factory=list, description='Classes this vertex type is a subclass of (``rdfs:subClassOf``): a vertex of this schema by name, or an external class by IRI. A logical assertion, checked and entailed; nothing is inherited.') 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"
            )
        overlap = sorted(set(self.identity) & set(self.identity_funnel.field_names))
        if overlap:
            # The cast drops a digest vertex's identity fields from each
            # record, so a branch reading one could never complete.
            raise ValueError(
                f"Vertex '{self.name}': identity field(s) {overlap} are also "
                "identity_funnel branch fields; the field the digest is "
                "stored in must be one no branch reads"
            )
    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"
            )
    reserved_names = {f.name for f in merged_properties if f.reserved}
    keyed = sorted(
        reserved_names
        & {*identity_names, *hash_identity_names, *funnel_field_names}
    )
    if keyed:
        # The writer stamps a reserved field, so as a key it would fold
        # every vertex of one stamp into one.
        raise ValueError(
            f"Vertex '{self.name}': reserved property(ies) {keyed} cannot "
            "be part of its identity"
        )
    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)
    self._check_derived_fields(merged_properties, funnel=set(funnel_field_names))

    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
        ]

    def has_vertex(self, name: str) -> bool:
        """Whether *name* is a declared vertex type; O(1), unlike ``vertex_set``."""
        return name in self._get_vertices_map()

    def is_blank(self, name: str) -> bool:
        """Whether *name* is in :attr:`blank_vertices`, without building the list."""
        vertex = self._get_vertices_map().get(name)
        return vertex is not None and vertex.blank

    def is_assigned(self, name: str) -> bool:
        """Whether *name* is in :attr:`assigned_vertices`, without building the list."""
        vertex = self._get_vertices_map().get(name)
        return vertex is not None and vertex.assigned

    def is_hash_identity(self, name: str) -> bool:
        """Whether *name* is in :attr:`hash_identity_vertices`, without building it."""
        vertex = self._get_vertices_map().get(name)
        return vertex is not None and bool(
            vertex.hash_identity_properties or vertex.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:
                digest = bool(vertex.hash_identity_properties or vertex.identity_funnel)
                if digest and blank_id_field in vertex.property_names:
                    # Before normalization assigns it, `id` here is the author's,
                    # and the cast discards a record's own `id` for the digest.
                    raise ValueError(
                        f"Vertex '{vertex.name}' declares a property "
                        f"`{blank_id_field}`, which its digest identity uses for "
                        "its synthetic key: a record's own "
                        f"`{blank_id_field}` is replaced by the digest, never "
                        f"stored. Leave `{blank_id_field}` out of `properties` "
                        "(the key is added for you), or rename it if records "
                        "carry one you want to keep"
                    )
                if digest 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()

has_vertex(name)

Whether name is a declared vertex type; O(1), unlike vertex_set.

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

def has_vertex(self, name: str) -> bool:
    """Whether *name* is a declared vertex type; O(1), unlike ``vertex_set``."""
    return name in self._get_vertices_map()

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)

is_assigned(name)

Whether name is in :attr:assigned_vertices, without building the list.

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

def is_assigned(self, name: str) -> bool:
    """Whether *name* is in :attr:`assigned_vertices`, without building the list."""
    vertex = self._get_vertices_map().get(name)
    return vertex is not None and vertex.assigned

is_blank(name)

Whether name is in :attr:blank_vertices, without building the list.

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

def is_blank(self, name: str) -> bool:
    """Whether *name* is in :attr:`blank_vertices`, without building the list."""
    vertex = self._get_vertices_map().get(name)
    return vertex is not None and vertex.blank

is_hash_identity(name)

Whether name is in :attr:hash_identity_vertices, without building it.

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

def is_hash_identity(self, name: str) -> bool:
    """Whether *name* is in :attr:`hash_identity_vertices`, without building it."""
    vertex = self._get_vertices_map().get(name)
    return vertex is not None and bool(
        vertex.hash_identity_properties or vertex.identity_funnel is not None
    )

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, retype_remedy=None)

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. retype_remedy is the fix a type clash suggests; a merge passes its own.

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

def merge_fields(
    a: Field, b: Field, *, owner: str, retype_remedy: str | None = None
) -> 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.
    ``retype_remedy`` is the fix a type clash suggests; a merge passes its own.
    """
    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 or _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

    if (a.reserved, a.merge, a.set_by) != (b.reserved, b.merge, b.set_by):
        # A writer-only field must not turn into a readable one by a merge.
        raise FieldMergeConflict(
            owner,
            f"property {a.name!r}: reserved={a.reserved}, merge={a.merge!r}, "
            f"set_by={a.set_by!r} vs reserved={b.reserved}, merge={b.merge!r}, "
            f"set_by={b.set_by!r}",
            "Declare the property the same way on both sides before merging.",
            field=a.name,
        )
    # One side declaring how the field matches, or that it is derived, carries;
    # two different declarations would change identity or comparison silently.
    declared: dict[str, Any] = {}
    for attribute in ("match", "derived"):
        left, right = getattr(a, attribute), getattr(b, attribute)
        if left is not None and right is not None and left != right:
            raise FieldMergeConflict(
                owner,
                f"property {a.name!r}: {attribute} {left.to_dict()} vs "
                f"{right.to_dict()}",
                "Declare the property the same way on both sides before merging.",
                field=a.name,
            )
        declared[attribute] = left if left is not None else right

    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,
            **declared,
        }
    )

union_field_lists(fields, *, owner, retype_remedy=None)

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, retype_remedy: str | None = None
) -> 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, retype_remedy=retype_remedy
            )
        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]