Skip to content

graflo.architecture.contract.ingestion.sync

Record sync ownership: which resource's records replace which elements.

A resource that declares sync owns one vertex type and a set of edges. In sync mode, re-ingesting one of its records replaces that record's vertex properties and deletes and recreates the edges it owns. Each owned edge belongs to one owner record, its owner key: the source endpoint when the resource owns the source type, else the target endpoint when it owns the target type, else the value of the edge's owner_field.

Attributes

OwnerSide = Literal['source', 'target', 'field'] module-attribute

Classes

OwnedEdge

Bases: NamedTuple

How one declared edge is owned: by which vertex type, read from where.

Source code in graflo/architecture/contract/ingestion/sync.py 
29
30
31
32
33
34
class OwnedEdge(NamedTuple):
    """How one declared edge is owned: by which vertex type, read from where."""

    owner: str
    side: OwnerSide
    owner_field: str | None

Attributes

owner instance-attribute
owner_field instance-attribute
side instance-attribute

ResourceSync

Bases: ConfigBaseModel

What a resource's records replace when re-ingested in sync mode.

Source code in graflo/architecture/contract/ingestion/sync.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class ResourceSync(ConfigBaseModel):
    """What a resource's records replace when re-ingested in sync mode."""

    owns: str = PydanticField(
        ..., description="Vertex type whose properties the resource's records replace."
    )
    edges: list[SyncEdge] = PydanticField(
        default_factory=list,
        description="Edges the resource's records replace, each owned by one record.",
    )

    def owner_side(self, edge: SyncEdge) -> OwnerSide:
        """Where *edge*'s owner key is read; the source wins when both endpoints are owned."""
        if edge.source == self.owns:
            return "source"
        if edge.target == self.owns:
            return "target"
        return "field"

Attributes

edges = PydanticField(default_factory=list, description="Edges the resource's records replace, each owned by one record.") class-attribute instance-attribute
owns = PydanticField(..., description="Vertex type whose properties the resource's records replace.") class-attribute instance-attribute

Methods:

owner_side(edge)

Where edge's owner key is read; the source wins when both endpoints are owned.

Source code in graflo/architecture/contract/ingestion/sync.py 
67
68
69
70
71
72
73
def owner_side(self, edge: SyncEdge) -> OwnerSide:
    """Where *edge*'s owner key is read; the source wins when both endpoints are owned."""
    if edge.source == self.owns:
        return "source"
    if edge.target == self.owns:
        return "target"
    return "field"

SyncEdge

Bases: ConfigBaseModel

An edge a resource's records own.

Source code in graflo/architecture/contract/ingestion/sync.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class SyncEdge(ConfigBaseModel):
    """An edge a resource's records own."""

    source: str = PydanticField(..., description="Edge source vertex type.")
    target: str = PydanticField(..., description="Edge target vertex type.")
    relation: str | None = PydanticField(default=None, description="Edge relation.")
    owner_field: str | None = PydanticField(
        default=None,
        description=(
            "Edge identity field holding the owner key, for an edge whose "
            "endpoints are neither of the owned type."
        ),
    )

    @property
    def edge_id(self) -> EdgeId:
        return self.source, self.target, self.relation

Attributes

edge_id property
owner_field = PydanticField(default=None, description='Edge identity field holding the owner key, for an edge whose endpoints are neither of the owned type.') class-attribute instance-attribute
relation = PydanticField(default=None, description='Edge relation.') class-attribute instance-attribute
source = PydanticField(..., description='Edge source vertex type.') class-attribute instance-attribute
target = PydanticField(..., description='Edge target vertex type.') class-attribute instance-attribute

Functions:

owned_edges(resources)

Every edge the resources own, by its declared id.

Source code in graflo/architecture/contract/ingestion/sync.py 
76
77
78
79
80
81
82
83
84
85
def owned_edges(resources: Iterable[ResourceConfig]) -> dict[EdgeId, OwnedEdge]:
    """Every edge the *resources* own, by its declared id."""
    return {
        entry.edge_id: OwnedEdge(
            resource.sync.owns, resource.sync.owner_side(entry), entry.owner_field
        )
        for resource in resources
        if resource.sync is not None
        for entry in resource.sync.edges
    }

owned_types(resources)

Every vertex type the resources own.

Source code in graflo/architecture/contract/ingestion/sync.py 
88
89
90
def owned_types(resources: Iterable[ResourceConfig]) -> set[str]:
    """Every vertex type the *resources* own."""
    return {r.sync.owns for r in resources if r.sync is not None}

sync_errors(resources, produced, core_schema)

Why the resources' sync blocks cannot be honoured; empty when they can.

Parameters:

Name Type Description Default
resources Sequence[ResourceConfig]

Every resource of the ingestion model.

 required
produced Mapping[str, set[str]]

The vertex types each resource can produce, by resource name.

 required
core_schema CoreSchema

The schema the blocks refer to.

 required
Source code in graflo/architecture/contract/ingestion/sync.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
def sync_errors(
    resources: Sequence[ResourceConfig],
    produced: Mapping[str, set[str]],
    core_schema: CoreSchema,
) -> list[str]:
    """Why the resources' ``sync`` blocks cannot be honoured; empty when they can.

    Args:
        resources: Every resource of the ingestion model.
        produced: The vertex types each resource can produce, by resource name.
        core_schema: The schema the blocks refer to.
    """
    from graflo.architecture.contract.ingestion.emits import may_emit, step_may_write
    from graflo.architecture.contract.ingestion.steps.ref import iter_edge_steps

    errors: list[str] = []
    owners: dict[str, str] = {}
    edge_owners: dict[EdgeId, str] = {}
    vertices = core_schema.vertex_config.vertex_set
    declared = dict(core_schema.edge_config.items())
    for resource in resources:
        sync = resource.sync
        if sync is None:
            continue
        name = resource.name
        if sync.owns not in vertices:
            errors.append(
                f"{name}: owns {sync.owns!r}, which the schema does not declare"
            )
        elif sync.owns not in produced[name]:
            errors.append(f"{name}: owns {sync.owns!r}, which it does not produce")
        if sync.owns in owners:
            errors.append(
                f"{name}: owns {sync.owns!r}, already owned by {owners[sync.owns]!r}"
            )
        owners.setdefault(sync.owns, name)
        for entry in sync.edges:
            edge_id = entry.edge_id
            if edge_id in edge_owners:
                errors.append(
                    f"{name}: owns edge {edge_id}, already owned by "
                    f"{edge_owners[edge_id]!r}"
                )
                continue
            edge_owners[edge_id] = name
            edge = declared.get(edge_id)
            if edge is None:
                errors.append(
                    f"{name}: owns edge {edge_id}, which the schema does not declare"
                )
                continue
            if not may_emit(resource, edge_id, produced[name]):
                errors.append(f"{name}: owns edge {edge_id}, which it does not write")
            if any(
                view.relation_is_data_driven and step_may_write(view, edge_id)
                for view in iter_edge_steps(resource.pipeline)
            ):
                # Its owned edges are deleted by type, which must be known here.
                errors.append(
                    f"{name}: owns edge {edge_id}, whose relation a step reads "
                    "from the data"
                )
            side = sync.owner_side(entry)
            identity = edge.identities[0] if edge.identities else []
            if side != "field":
                if entry.owner_field is not None:
                    errors.append(
                        f"{name}: edge {edge_id} is owned by its {side}; "
                        "it takes no owner_field"
                    )
            elif entry.owner_field is None:
                errors.append(
                    f"{name}: edge {edge_id} touches no {sync.owns!r}; name the "
                    "identity field that holds its owner as owner_field"
                )
            elif entry.owner_field in ("source", "target", "relation") or (
                entry.owner_field not in identity
            ):
                errors.append(
                    f"{name}: owner_field {entry.owner_field!r} of edge {edge_id} "
                    f"is not an identity field of the edge ({identity})"
                )
            elif (
                sync.owns in vertices
                and len(core_schema.vertex_config.identity_fields(sync.owns)) != 1
            ):
                # The field holds one value, so it names an owner by one field.
                errors.append(
                    f"{name}: edge {edge_id} names its owner by owner_field, which "
                    f"needs {sync.owns!r} to have a single identity field"
                )
    for edge_id, owner in edge_owners.items():
        others = [
            resource.name
            for resource in resources
            if resource.name != owner
            and may_emit(resource, edge_id, produced[resource.name])
        ]
        if others:
            errors.append(
                f"edge {edge_id} is owned by {owner!r} but also written by {others}"
            )
    return errors