Skip to content

graflo.hq.sync.refusals

What a sync run cannot do: write an element no sync block owns.

A sync run replaces what owner records wrote, so everything it writes must be replaceable: every edge is owned by a resource in the run, and every owned type carries a hash field (the sync aspect). Checked before anything is staged. A vertex type the run does not own must be written key-only; that is checked on the folded rows (:func:graflo.hq.sync.run.check_key_only).

Classes

Functions:

hash_fields(schema, owned)

Each type of owned that has one, mapped to its set_by: sync field.

Source code in graflo/hq/sync/refusals.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
def hash_fields(schema: Schema, owned: set[str]) -> dict[str, str]:
    """Each type of *owned* that has one, mapped to its ``set_by: sync`` field."""
    vc = schema.core_schema.vertex_config
    out: dict[str, str] = {}
    for vertex in sorted(owned):
        fields = [f.name for f in vc.properties(vertex) if f.set_by == "sync"]
        if len(fields) > 1:
            raise ValueError(
                f"vertex {vertex!r} has several fields set by sync: {fields}"
            )
        if fields:
            out[vertex] = fields[0]
    return out

sync_refusals(schema, ingestion_model, resources=None)

Why a sync run of resources (all when None) cannot start; empty when it can.

Source code in graflo/hq/sync/refusals.py 
18
19
20
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
def sync_refusals(
    schema: Schema, ingestion_model: IngestionModel, resources: list[str] | None = None
) -> list[str]:
    """Why a sync run of *resources* (all when ``None``) cannot start; empty when it can."""
    in_scope = [
        r for r in ingestion_model.resources if resources is None or r.name in resources
    ]
    owners = [r for r in in_scope if r.sync is not None]
    if not owners:
        return ["no resource in the run declares a sync block"]
    reasons: list[str] = []
    owned = owned_edges(in_scope)
    core = schema.core_schema
    vc = core.vertex_config
    for resource in in_scope:
        produced = ingestion_model.fetch_resource(resource.name).collect_vertex_names()
        for edge_id in dict(core.edge_config.items()):
            if edge_id not in owned and may_emit(resource, edge_id, produced):
                reasons.append(
                    f"{resource.name}: writes edge {edge_id}, which no sync block owns"
                )
    for resource in owners:
        assert resource.sync is not None
        owns = resource.sync.owns
        if not [f for f in vc.properties(owns) if f.set_by == "sync"]:
            reasons.append(
                f"{resource.name}: {owns!r} has no field set by sync; lift the "
                "sync aspect (graflo lift --aspect sync)"
            )
    return reasons