Skip to content

graflo.architecture.edge_derivation

Ingestion-time wiring: how an edge step binds to extracted locations and documents.

These fields do not belong in schema edge_config / :class:~graflo.architecture.schema.edge.Edge. They are set on edge pipeline steps (:class:~graflo.architecture.pipeline.runtime.actor.config.models.EdgeActorConfig), threaded through :class:~graflo.architecture.graph_types.EdgeIntent, and used in :func:~graflo.architecture.pipeline.runtime.actor.edge_render.render_edge.

When :attr:EdgeDerivation.relation_from_key is true, the ingestion :class:~graflo.architecture.contract.declarations.edge_derivation_registry.EdgeDerivationRegistry records the edge id so :class:~graflo.architecture.schema.db_aware.EdgeConfigDBAware (with overlay) can align TigerGraph DDL with runtime.

EdgeDerivation

Bases: ConfigBaseModel

How this edge step selects vertex locations and reads per-row relation from data.

Source code in graflo/architecture/edge_derivation.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
class EdgeDerivation(ConfigBaseModel):
    """How this edge step selects vertex locations and reads per-row relation from data."""

    match_source: str | None = Field(
        default=None,
        description="Require this path segment in source vertex locations.",
    )
    match_target: str | None = Field(
        default=None,
        description="Require this path segment in target vertex locations.",
    )
    exclude_source: str | None = Field(
        default=None,
        description="Exclude source locations containing this path segment.",
    )
    exclude_target: str | None = Field(
        default=None,
        description="Exclude target locations containing this path segment.",
    )
    match: str | None = Field(
        default=None,
        description="Require this segment in both source and target locations.",
    )
    relation_field: str | None = Field(
        default=None,
        description="Document/ctx field name for per-row relationship label when schema relation is unset.",
    )
    relation_from_key: bool = Field(
        default=False,
        description="If True, derive the per-row relation label from the location key during assembly.",
    )

    def is_empty(self) -> bool:
        if self.relation_from_key:
            return False
        return all(
            getattr(self, name) is None
            for name in (
                "match_source",
                "match_target",
                "exclude_source",
                "exclude_target",
                "match",
                "relation_field",
            )
        )