graflo.architecture.schema.db_aware

DB-aware projections for logical graph configs.

These wrappers materialize database-specific naming/defaults from logical VertexConfig/EdgeConfig and DPProfile without mutating logical models.

Attributes

Classes

EdgeConfigDBAware

DB-aware projection wrapper for EdgeConfig.

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

class EdgeConfigDBAware:
    """DB-aware projection wrapper for `EdgeConfig`."""

    def __init__(
        self,
        logical: EdgeConfig,
        vertex_config: VertexConfigDBAware,
        database_features: DatabaseProfile,
        ingestion_overlay: EdgeIngestionOverlay | None = None,
    ):
        self.logical = logical
        self.vertex_config = vertex_config
        self.db_profile = database_features
        self.ingestion_overlay = ingestion_overlay

    def _uses_relation_from_key(self, edge_id: EdgeId) -> bool:
        if self.ingestion_overlay is not None:
            return self.ingestion_overlay.uses_relation_from_key(edge_id)
        return False

    @property
    def edges(self) -> list[Edge]:
        return self.logical.edges

    def __iter__(self) -> Iterator[Edge]:
        return self.values()

    def values(self) -> Iterator[Edge]:
        return self.logical.values()

    def items(self) -> Iterator[tuple[EdgeId, Edge]]:
        return self.logical.items()

    @property
    def vertices(self):
        return self.logical.vertices

    def relation_dbname(self, edge: Edge) -> str | None:
        relation = edge.relation
        if self.db_profile.db_flavor == DBType.TIGERGRAPH and relation is None:
            relation = DEFAULT_TIGERGRAPH_RELATION
        return self.db_profile.edge_relation_name(
            edge.edge_id,
            default_relation=relation,
        )

    def effective_weights(self, edge: Edge) -> WeightConfig | None:
        def _as_weight_config() -> WeightConfig | None:
            if not edge.properties:
                return None
            return WeightConfig(
                direct=[f.model_copy(deep=True) for f in edge.properties],
            )

        if self.db_profile.db_flavor != DBType.TIGERGRAPH:
            return _as_weight_config()

        # Typed TigerGraph edge: per-document relation label stored under a stable attribute.
        needs_relation_attr = edge.relation is None or self._uses_relation_from_key(
            edge.edge_id
        )
        if not needs_relation_attr:
            return _as_weight_config()

        base = _as_weight_config() or WeightConfig()
        if DEFAULT_TIGERGRAPH_RELATION_WEIGHTNAME not in base.direct_names:
            base.direct.append(
                Field(
                    name=DEFAULT_TIGERGRAPH_RELATION_WEIGHTNAME, type=FieldType.STRING
                )
            )
        return base

    def runtime(self, edge: Edge) -> EdgeRuntime:
        needs_tg_relation_attr = self.db_profile.db_flavor == DBType.TIGERGRAPH and (
            edge.relation is None or self._uses_relation_from_key(edge.edge_id)
        )
        runtime = EdgeRuntime(
            edge=edge,
            source_storage=self.vertex_config.vertex_dbname(edge.source),
            target_storage=self.vertex_config.vertex_dbname(edge.target),
            relation_name=self.relation_dbname(edge),
            store_extracted_relation_as_weight=needs_tg_relation_attr,
            effective_relation_field=(
                DEFAULT_TIGERGRAPH_RELATION_WEIGHTNAME
                if needs_tg_relation_attr
                else None
            ),
            db_profile=self.db_profile,
        )
        return runtime

    def relationship_merge_property_names(self, edge: Edge) -> list[str]:
        """Relationship properties that, with the endpoints, key an edge (per backend).

        An edge is keyed by its endpoints and relation. The **first** logical
        ``identities`` key adds the properties it names, which is how parallel
        edges between the same two vertices stay distinct; every other property
        is written, never matched on. Additional ``identities`` keys are
        compiled into separate indexes via :meth:`compile_identity_indexes` but
        do not change the writer's key. Without ``identities`` the list is empty.
        """
        return relationship_merge_properties(edge, self.db_profile.db_flavor)

    def compile_identity_indexes(self) -> None:
        db_flavor = self.db_profile.db_flavor
        for edge in self.logical.edges:
            for identity_key in edge.identities:
                index = edge_identity_index(identity_key, db_flavor)
                if index is not None:
                    self.db_profile.add_edge_index(edge.edge_id, index, purpose=None)

Attributes

db_profile = database_features instance-attribute

edges property

ingestion_overlay = ingestion_overlay instance-attribute

logical = logical instance-attribute

vertex_config = vertex_config instance-attribute

vertices property

Methods:

__init__(logical, vertex_config, database_features, ingestion_overlay=None)

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

def __init__(
    self,
    logical: EdgeConfig,
    vertex_config: VertexConfigDBAware,
    database_features: DatabaseProfile,
    ingestion_overlay: EdgeIngestionOverlay | None = None,
):
    self.logical = logical
    self.vertex_config = vertex_config
    self.db_profile = database_features
    self.ingestion_overlay = ingestion_overlay

__iter__()

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

def __iter__(self) -> Iterator[Edge]:
    return self.values()

compile_identity_indexes()

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

def compile_identity_indexes(self) -> None:
    db_flavor = self.db_profile.db_flavor
    for edge in self.logical.edges:
        for identity_key in edge.identities:
            index = edge_identity_index(identity_key, db_flavor)
            if index is not None:
                self.db_profile.add_edge_index(edge.edge_id, index, purpose=None)

effective_weights(edge)

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

def effective_weights(self, edge: Edge) -> WeightConfig | None:
    def _as_weight_config() -> WeightConfig | None:
        if not edge.properties:
            return None
        return WeightConfig(
            direct=[f.model_copy(deep=True) for f in edge.properties],
        )

    if self.db_profile.db_flavor != DBType.TIGERGRAPH:
        return _as_weight_config()

    # Typed TigerGraph edge: per-document relation label stored under a stable attribute.
    needs_relation_attr = edge.relation is None or self._uses_relation_from_key(
        edge.edge_id
    )
    if not needs_relation_attr:
        return _as_weight_config()

    base = _as_weight_config() or WeightConfig()
    if DEFAULT_TIGERGRAPH_RELATION_WEIGHTNAME not in base.direct_names:
        base.direct.append(
            Field(
                name=DEFAULT_TIGERGRAPH_RELATION_WEIGHTNAME, type=FieldType.STRING
            )
        )
    return base

items()

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

def items(self) -> Iterator[tuple[EdgeId, Edge]]:
    return self.logical.items()

relation_dbname(edge)

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

def relation_dbname(self, edge: Edge) -> str | None:
    relation = edge.relation
    if self.db_profile.db_flavor == DBType.TIGERGRAPH and relation is None:
        relation = DEFAULT_TIGERGRAPH_RELATION
    return self.db_profile.edge_relation_name(
        edge.edge_id,
        default_relation=relation,
    )

relationship_merge_property_names(edge)

Relationship properties that, with the endpoints, key an edge (per backend).

An edge is keyed by its endpoints and relation. The first logical identities key adds the properties it names, which is how parallel edges between the same two vertices stay distinct; every other property is written, never matched on. Additional identities keys are compiled into separate indexes via :meth:compile_identity_indexes but do not change the writer's key. Without identities the list is empty.

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

def relationship_merge_property_names(self, edge: Edge) -> list[str]:
    """Relationship properties that, with the endpoints, key an edge (per backend).

    An edge is keyed by its endpoints and relation. The **first** logical
    ``identities`` key adds the properties it names, which is how parallel
    edges between the same two vertices stay distinct; every other property
    is written, never matched on. Additional ``identities`` keys are
    compiled into separate indexes via :meth:`compile_identity_indexes` but
    do not change the writer's key. Without ``identities`` the list is empty.
    """
    return relationship_merge_properties(edge, self.db_profile.db_flavor)

runtime(edge)

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

def runtime(self, edge: Edge) -> EdgeRuntime:
    needs_tg_relation_attr = self.db_profile.db_flavor == DBType.TIGERGRAPH and (
        edge.relation is None or self._uses_relation_from_key(edge.edge_id)
    )
    runtime = EdgeRuntime(
        edge=edge,
        source_storage=self.vertex_config.vertex_dbname(edge.source),
        target_storage=self.vertex_config.vertex_dbname(edge.target),
        relation_name=self.relation_dbname(edge),
        store_extracted_relation_as_weight=needs_tg_relation_attr,
        effective_relation_field=(
            DEFAULT_TIGERGRAPH_RELATION_WEIGHTNAME
            if needs_tg_relation_attr
            else None
        ),
        db_profile=self.db_profile,
    )
    return runtime

values()

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

def values(self) -> Iterator[Edge]:
    return self.logical.values()

EdgeIngestionOverlay

Bases: Protocol

Ingestion-only signals that affect DB projection (e.g. TigerGraph DDL).

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

@runtime_checkable
class EdgeIngestionOverlay(Protocol):
    """Ingestion-only signals that affect DB projection (e.g. TigerGraph DDL)."""

    def uses_relation_from_key(self, edge_id: EdgeId) -> bool: ...

Methods:

uses_relation_from_key(edge_id)

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

def uses_relation_from_key(self, edge_id: EdgeId) -> bool: ...

EdgeRuntime dataclass

Resolved DB-facing runtime data for one logical edge.

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

@dataclass(frozen=True)
class EdgeRuntime:
    """Resolved DB-facing runtime data for one logical edge."""

    edge: Edge
    source_storage: str
    target_storage: str
    relation_name: str | None
    store_extracted_relation_as_weight: bool
    effective_relation_field: str | None
    db_profile: DatabaseProfile

    @property
    def edge_id(self) -> EdgeId:
        return self.edge.edge_id

    def storage_name(self, *, purpose: str | None = None) -> str | None:
        return self.db_profile.edge_storage_name(
            self.edge.edge_id,
            source_storage=self.source_storage,
            target_storage=self.target_storage,
            purpose=purpose,
        )

    def graph_name(self, *, purpose: str | None = None) -> str | None:
        return self.db_profile.edge_graph_name(
            self.edge.edge_id,
            source_storage=self.source_storage,
            target_storage=self.target_storage,
            purpose=purpose,
        )

    def physical_variants(self) -> list[dict[str, str | None | list[Index]]]:
        return self.db_profile.edge_physical_variants(
            self.edge.edge_id,
            source_storage=self.source_storage,
            target_storage=self.target_storage,
        )

Attributes

db_profile instance-attribute

edge instance-attribute

edge_id property

effective_relation_field instance-attribute

relation_name instance-attribute

source_storage instance-attribute

store_extracted_relation_as_weight instance-attribute

target_storage instance-attribute

Methods:

__init__(edge, source_storage, target_storage, relation_name, store_extracted_relation_as_weight, effective_relation_field, db_profile)

graph_name(*, purpose=None)

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

def graph_name(self, *, purpose: str | None = None) -> str | None:
    return self.db_profile.edge_graph_name(
        self.edge.edge_id,
        source_storage=self.source_storage,
        target_storage=self.target_storage,
        purpose=purpose,
    )

physical_variants()

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

def physical_variants(self) -> list[dict[str, str | None | list[Index]]]:
    return self.db_profile.edge_physical_variants(
        self.edge.edge_id,
        source_storage=self.source_storage,
        target_storage=self.target_storage,
    )

storage_name(*, purpose=None)

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

def storage_name(self, *, purpose: str | None = None) -> str | None:
    return self.db_profile.edge_storage_name(
        self.edge.edge_id,
        source_storage=self.source_storage,
        target_storage=self.target_storage,
        purpose=purpose,
    )

SchemaDBAware dataclass

DB-aware schema runtime view.

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

@dataclass(frozen=True)
class SchemaDBAware:
    """DB-aware schema runtime view."""

    vertex_config: VertexConfigDBAware
    edge_config: EdgeConfigDBAware
    db_profile: DatabaseProfile

Attributes

db_profile instance-attribute

edge_config instance-attribute

vertex_config instance-attribute

Methods:

__init__(vertex_config, edge_config, db_profile)

VertexConfigDBAware

DB-aware projection wrapper for VertexConfig.

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

class VertexConfigDBAware:
    """DB-aware projection wrapper for `VertexConfig`."""

    def __init__(self, logical: VertexConfig, database_features: DatabaseProfile):
        self.logical = logical
        self.db_profile = database_features

    @property
    def vertex_set(self):
        return self.logical.vertex_set

    @property
    def blank_vertices(self):
        return self.logical.blank_vertices

    @property
    def assigned_vertices(self):
        return self.logical.assigned_vertices

    @property
    def hash_identity_vertices(self):
        return self.logical.hash_identity_vertices

    def has_vertex(self, name: str) -> bool:
        return self.logical.has_vertex(name)

    def is_blank(self, name: str) -> bool:
        return self.logical.is_blank(name)

    def is_assigned(self, name: str) -> bool:
        return self.logical.is_assigned(name)

    def is_hash_identity(self, name: str) -> bool:
        return self.logical.is_hash_identity(name)

    @property
    def vertices(self):
        return self.logical.vertices

    def vertex_dbname(self, vertex_name: str) -> str:
        return self.db_profile.vertex_storage_name(vertex_name)

    def index(self, vertex_name: str) -> Index:
        """Get primary index for a vertex (DB layer needs Index for collection setup)."""
        return Index(fields=self.identity_fields(vertex_name), unique=True)

    def identity_fields(self, vertex_name: str) -> list[str]:
        identity = self.logical.identity_fields(vertex_name)
        if identity:
            return identity
        if self.logical.is_blank(vertex_name):
            return ["_key"] if self.db_profile.db_flavor == DBType.ARANGO else ["id"]
        return identity

    def secondary_identities(self, vertex_name: str) -> list[SecondaryIdentity]:
        return self.logical.secondary_identities(vertex_name)

    def compile_secondary_identity_indexes(self) -> None:
        """Register a lookup index for every declared secondary identity."""
        compile_secondary_identity_indexes(self.logical, self.db_profile)

    def match_fields(
        self, vertex_name: str, selector: str | list[str] | None
    ) -> list[str]:
        """Fields an edge endpoint is matched on, with DB-aware identity fallback."""
        if selector is None or selector == PRIMARY_IDENTITY_SELECTOR:
            return self.identity_fields(vertex_name)
        return self.logical.match_fields(vertex_name, selector)

    def properties(self, vertex_name: str) -> list[Field]:
        props = self.logical.properties(vertex_name)
        if self.db_profile.db_flavor != DBType.TIGERGRAPH:
            return props
        # TigerGraph needs explicit scalar defaults for schema definition.
        return [
            Field(
                name=f.name,
                type=FieldType.STRING if f.type is None else f.type,
                item_type=f.item_type,
                description=f.description,
            )
            for f in props
        ]

    def property_names(self, vertex_name: str) -> list[str]:
        return [f.name for f in self.properties(vertex_name)]

Attributes

assigned_vertices property

blank_vertices property

db_profile = database_features instance-attribute

hash_identity_vertices property

logical = logical instance-attribute

vertex_set property

vertices property

Methods:

__init__(logical, database_features)

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

def __init__(self, logical: VertexConfig, database_features: DatabaseProfile):
    self.logical = logical
    self.db_profile = database_features

compile_secondary_identity_indexes()

Register a lookup index for every declared secondary identity.

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

def compile_secondary_identity_indexes(self) -> None:
    """Register a lookup index for every declared secondary identity."""
    compile_secondary_identity_indexes(self.logical, self.db_profile)

has_vertex(name)

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

def has_vertex(self, name: str) -> bool:
    return self.logical.has_vertex(name)

identity_fields(vertex_name)

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

def identity_fields(self, vertex_name: str) -> list[str]:
    identity = self.logical.identity_fields(vertex_name)
    if identity:
        return identity
    if self.logical.is_blank(vertex_name):
        return ["_key"] if self.db_profile.db_flavor == DBType.ARANGO else ["id"]
    return identity

index(vertex_name)

Get primary index for a vertex (DB layer needs Index for collection setup).

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

def index(self, vertex_name: str) -> Index:
    """Get primary index for a vertex (DB layer needs Index for collection setup)."""
    return Index(fields=self.identity_fields(vertex_name), unique=True)

is_assigned(name)

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

def is_assigned(self, name: str) -> bool:
    return self.logical.is_assigned(name)

is_blank(name)

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

def is_blank(self, name: str) -> bool:
    return self.logical.is_blank(name)

is_hash_identity(name)

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

def is_hash_identity(self, name: str) -> bool:
    return self.logical.is_hash_identity(name)

match_fields(vertex_name, selector)

Fields an edge endpoint is matched on, with DB-aware identity fallback.

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

def match_fields(
    self, vertex_name: str, selector: str | list[str] | None
) -> list[str]:
    """Fields an edge endpoint is matched on, with DB-aware identity fallback."""
    if selector is None or selector == PRIMARY_IDENTITY_SELECTOR:
        return self.identity_fields(vertex_name)
    return self.logical.match_fields(vertex_name, selector)

properties(vertex_name)

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

def properties(self, vertex_name: str) -> list[Field]:
    props = self.logical.properties(vertex_name)
    if self.db_profile.db_flavor != DBType.TIGERGRAPH:
        return props
    # TigerGraph needs explicit scalar defaults for schema definition.
    return [
        Field(
            name=f.name,
            type=FieldType.STRING if f.type is None else f.type,
            item_type=f.item_type,
            description=f.description,
        )
        for f in props
    ]

property_names(vertex_name)

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

def property_names(self, vertex_name: str) -> list[str]:
    return [f.name for f in self.properties(vertex_name)]

secondary_identities(vertex_name)

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

def secondary_identities(self, vertex_name: str) -> list[SecondaryIdentity]:
    return self.logical.secondary_identities(vertex_name)

vertex_dbname(vertex_name)

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

def vertex_dbname(self, vertex_name: str) -> str:
    return self.db_profile.vertex_storage_name(vertex_name)

WeightConfig

Bases: ConfigBaseModel

Configuration for edge weights and relationships.

This class manages the configuration of weights and relationships for edges, including source and target field mappings.

Attributes:

Name	Type	Description
vertices	list[Weight]	List of weight configurations
direct	list[Field]	List of direct field mappings. Can be specified as strings, Field objects, or dicts. Will be normalized to Field objects by the validator. After initialization, this is always list[Field] (type checker sees this).

Examples:

>>> # List of strings
>>> wc1 = WeightConfig(direct=["date", "weight"])

>>> # Typed fields: list of Field objects
>>> wc2 = WeightConfig(direct=[
...     Field(name="date", type="DATETIME"),
...     Field(name="weight", type="FLOAT")
... ])

>>> # From dicts (e.g., from YAML/JSON)
>>> wc3 = WeightConfig(direct=[
...     {"name": "date", "type": "DATETIME"},
...     {"name": "weight"}  # defaults to None type
... ])

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

class WeightConfig(ConfigBaseModel):
    """Configuration for edge weights and relationships.

    This class manages the configuration of weights and relationships for edges,
    including source and target field mappings.

    Attributes:
        vertices: List of weight configurations
        direct: List of direct field mappings. Can be specified as strings, Field objects, or dicts.
               Will be normalized to Field objects by the validator.
               After initialization, this is always list[Field] (type checker sees this).

    Examples:
        >>> # List of strings
        >>> wc1 = WeightConfig(direct=["date", "weight"])

        >>> # Typed fields: list of Field objects
        >>> wc2 = WeightConfig(direct=[
        ...     Field(name="date", type="DATETIME"),
        ...     Field(name="weight", type="FLOAT")
        ... ])

        >>> # From dicts (e.g., from YAML/JSON)
        >>> wc3 = WeightConfig(direct=[
        ...     {"name": "date", "type": "DATETIME"},
        ...     {"name": "weight"}  # defaults to None type
        ... ])
    """

    vertices: list[Weight] = PydanticField(
        default_factory=list,
        description="List of weight definitions for vertex-based edge attributes.",
    )
    direct: list[Field] = PydanticField(
        default_factory=list,
        description="Direct edge attributes (field names, Field objects, or dicts). Normalized to Field objects.",
    )

    @field_validator("direct", mode="before")
    @classmethod
    def normalize_direct(cls, v: Any) -> Any:
        if not isinstance(v, list):
            return v
        return [_normalize_direct_item(item) for item in v]

    @property
    def direct_names(self) -> list[str]:
        """Get list of direct field names (as strings).

        Returns:
            list[str]: List of field names
        """
        return [field.name for field in self.direct]

Attributes

direct = PydanticField(default_factory=list, description='Direct edge attributes (field names, Field objects, or dicts). Normalized to Field objects.') class-attribute instance-attribute

direct_names property

Get list of direct field names (as strings).

Returns:

Type	Description
list[str]	list[str]: List of field names

vertices = PydanticField(default_factory=list, description='List of weight definitions for vertex-based edge attributes.') class-attribute instance-attribute

Methods:

normalize_direct(v) classmethod

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

@field_validator("direct", mode="before")
@classmethod
def normalize_direct(cls, v: Any) -> Any:
    if not isinstance(v, list):
        return v
    return [_normalize_direct_item(item) for item in v]

Functions:

compile_secondary_identity_indexes(vertex_config, db_profile)

Register a lookup index for every declared secondary identity.

Endpoint resolution filters on these fields, so the index is what keeps the lookup from degrading into a scan — and on NebulaGraph a tag index is required outright for the property lookup to run at all.

Indexes are non-unique: secondary identities are softly unique, and asking the database to enforce uniqueness would reject exactly the duplicate data the ambiguity policy exists to handle.

Called from :meth:Schema.finish_init so the profile is populated for every backend, including those whose index definition does not go through :meth:Schema.resolve_db_aware.

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

def compile_secondary_identity_indexes(
    vertex_config: VertexConfig, db_profile: DatabaseProfile
) -> None:
    """Register a lookup index for every declared secondary identity.

    Endpoint resolution filters on these fields, so the index is what keeps the
    lookup from degrading into a scan — and on NebulaGraph a tag index is
    required outright for the property lookup to run at all.

    Indexes are non-unique: secondary identities are *softly* unique, and asking
    the database to enforce uniqueness would reject exactly the duplicate data
    the ambiguity policy exists to handle.

    Called from :meth:`Schema.finish_init` so the profile is populated for every
    backend, including those whose index definition does not go through
    :meth:`Schema.resolve_db_aware`.
    """
    for vertex in vertex_config.vertices:
        for entry in vertex.secondary_identities:
            if not entry.fields:
                continue
            db_profile.add_vertex_index(
                vertex.name, Index(fields=list(entry.fields), unique=False)
            )

edge_identity_index(identity_key, db_flavor)

The index an edge identities key compiles to on db_flavor, or None.

Logical uniqueness is always (source, *relationship_fields, target).

• ArangoDB — Edge documents carry _from / _to. Unique persistent indexes must include them before other fields, even when the YAML identities entry lists only relationship tokens (e.g. _role).
• Neo4j, FalkorDB, Memgraph, Nebula — Indexed columns are relationship / edge-type properties only; they cannot express endpoint scope. We still register the property fields for lookups but set unique=False so the database is not asked to enforce a misleading global uniqueness on those properties alone. (Application MERGE / ingest semantics remain authoritative.)
• TigerGraph, PostgreSQL — Uniqueness is kept on the index; neither builds it (TigerGraph has no edge indexes, the PostgreSQL edge key covers it).

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

def edge_identity_index(identity_key: Sequence[str], db_flavor: DBType) -> Index | None:
    """The index an edge ``identities`` key compiles to on *db_flavor*, or ``None``.

    Logical uniqueness is always ``(source, *relationship_fields, target)``.

    * **ArangoDB** — Edge documents carry ``_from`` / ``_to``. Unique persistent
      indexes must include them before other fields, even when the YAML
      ``identities`` entry lists only relationship tokens (e.g. ``_role``).
    * **Neo4j, FalkorDB, Memgraph, Nebula** — Indexed columns are relationship /
      edge-type properties only; they cannot express endpoint scope. We still
      register the property fields for lookups but set ``unique=False`` so the
      database is not asked to enforce a misleading global uniqueness on those
      properties alone. (Application MERGE / ingest semantics remain authoritative.)
    * **TigerGraph, PostgreSQL** — Uniqueness is kept on the index; neither builds
      it (TigerGraph has no edge indexes, the PostgreSQL edge key covers it).
    """
    fields: list[str] = []
    for token in identity_key:
        if token == "source":
            if db_flavor == DBType.ARANGO:
                fields.append("_from")
        elif token == "target":
            if db_flavor == DBType.ARANGO:
                fields.append("_to")
        elif token == "relation":
            if db_flavor != DBType.TIGERGRAPH:
                fields.append("relation")
        else:
            fields.append(token)
    fields = list(dict.fromkeys(fields))
    if not fields:
        return None
    if db_flavor == DBType.ARANGO:
        rest = [f for f in fields if f not in ("_from", "_to")]
        return Index(fields=["_from", "_to", *rest], unique=True)
    unique = db_flavor not in (
        DBType.NEO4J,
        DBType.FALKORDB,
        DBType.MEMGRAPH,
        DBType.NEBULA,
    )
    return Index(fields=fields, unique=unique)

relationship_merge_properties(edge, db_flavor)

Relationship properties that, with the endpoints, key edge on db_flavor.

See :meth:EdgeConfigDBAware.relationship_merge_property_names; needs no db-aware projection, only the edge's first identities key.

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

def relationship_merge_properties(edge: Edge, db_flavor: DBType) -> list[str]:
    """Relationship properties that, with the endpoints, key *edge* on *db_flavor*.

    See :meth:`EdgeConfigDBAware.relationship_merge_property_names`; needs no
    db-aware projection, only the edge's first ``identities`` key.
    """
    if not edge.identities:
        return []
    fields: list[str] = []
    for token in edge.identities[0]:
        if token in ("source", "target"):
            continue
        if token == "relation":
            # Where the relation is the stored edge type, it keys nothing more.
            if db_flavor not in (DBType.TIGERGRAPH, DBType.POSTGRES, DBType.NEBULA):
                fields.append("relation")
            continue
        fields.append(token)
    deduped: list[str] = []
    for field in fields:
        if field not in deduped:
            deduped.append(field)
    return deduped