Skip to content

graflo.architecture.schema.key_contract

A vertex type's key contract, in terms any schema language can state.

A producer that only knows some properties of an entity -- an extractor reading text, a partial record -- needs to know whether what it holds identifies the entity. :func:key_contract states that for a :class:Vertex:

source graflo declaration Primary key comes from
stated identity the identity fields themselves
derived hash_identity_properties, identity_funnel a digest of source fields
minted assigned, blank the writer; never from the data

keys are the field sets that make a complete primary key (one, or one per funnel branch); alternates are the secondary identities, which find a vertex but never create one. :func:key_status says which a set of values completes.

Attributes

KeySource = Literal['stated', 'derived', 'minted'] module-attribute

KeyState = Literal['primary', 'alternate', 'partial', 'none', 'keyless'] module-attribute

Classes

KeyContract

Bases: ConfigBaseModel

How a vertex type is identified.

Source code in graflo/architecture/schema/key_contract.py 
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
class KeyContract(ConfigBaseModel):
    """How a vertex type is identified."""

    vertex: str
    source: KeySource
    identity_fields: tuple[str, ...] = Field(
        description="Stored primary identity fields (``id`` for a derived key)."
    )
    keys: tuple[KeySet, ...] = Field(
        default=(),
        description="Field sets that complete the primary key; empty when minted.",
    )
    alternates: tuple[KeySet, ...] = Field(
        default=(), description="Secondary identities: lookup keys, never upserted on."
    )

    @property
    def key_fields(self) -> frozenset[str]:
        """Every field that takes part in a primary or alternate key."""
        return frozenset(
            f for key in (*self.keys, *self.alternates) for f in key.fields
        )

Attributes

alternates = Field(default=(), description='Secondary identities: lookup keys, never upserted on.') class-attribute instance-attribute
identity_fields = Field(description='Stored primary identity fields (``id`` for a derived key).') class-attribute instance-attribute
key_fields property

Every field that takes part in a primary or alternate key.

keys = Field(default=(), description='Field sets that complete the primary key; empty when minted.') class-attribute instance-attribute
source instance-attribute
vertex instance-attribute

KeySet

Bases: ConfigBaseModel

Fields that identify a vertex when all of them hold a value.

Source code in graflo/architecture/schema/key_contract.py 
34
35
36
37
38
39
40
class KeySet(ConfigBaseModel):
    """Fields that identify a vertex when all of them hold a value."""

    name: str = Field(
        description="Funnel branch id, secondary identity name, or ``identity``."
    )
    fields: tuple[str, ...] = Field(description="Fields that must all be non-empty.")

Attributes

fields = Field(description='Fields that must all be non-empty.') class-attribute instance-attribute
name = Field(description='Funnel branch id, secondary identity name, or ``identity``.') class-attribute instance-attribute

KeyStatus

Bases: ConfigBaseModel

What a set of values completes under a :class:KeyContract.

Source code in graflo/architecture/schema/key_contract.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
class KeyStatus(ConfigBaseModel):
    """What a set of values completes under a :class:`KeyContract`."""

    state: KeyState = Field(
        description=(
            "``primary``: a key set is complete. ``alternate``: only a secondary "
            "identity is. ``partial``: some key fields hold values, no set is "
            "complete. ``none``: no key field holds a value. ``keyless``: the "
            "type takes no key from the data."
        )
    )
    key: KeySet | None = Field(
        default=None, description="The complete primary key set."
    )
    alternates: tuple[KeySet, ...] = Field(
        default=(), description="Complete secondary identities, in declaration order."
    )
    missing: tuple[str, ...] = Field(
        default=(), description="Fields still needed for the nearest primary key set."
    )

Attributes

alternates = Field(default=(), description='Complete secondary identities, in declaration order.') class-attribute instance-attribute
key = Field(default=None, description='The complete primary key set.') class-attribute instance-attribute
missing = Field(default=(), description='Fields still needed for the nearest primary key set.') class-attribute instance-attribute
state = Field(description='``primary``: a key set is complete. ``alternate``: only a secondary identity is. ``partial``: some key fields hold values, no set is complete. ``none``: no key field holds a value. ``keyless``: the type takes no key from the data.') class-attribute instance-attribute

Functions:

key_contract(vertex)

The key contract of vertex.

Source code in graflo/architecture/schema/key_contract.py 
 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
def key_contract(vertex: Vertex) -> KeyContract:
    """The key contract of *vertex*."""
    alternates = tuple(
        KeySet(name=s.name or f"secondary_{i}", fields=tuple(s.fields))
        for i, s in enumerate(vertex.secondary_identities)
    )
    mode = vertex.identity_mode
    identity = tuple(vertex.identity) if vertex.identity else ("id",)
    if mode in ("assigned", "blank"):
        return KeyContract(
            vertex=vertex.name,
            source="minted",
            identity_fields=identity,
            alternates=alternates,
        )
    if mode == "hash":
        if vertex.identity_funnel is not None:
            keys = tuple(
                KeySet(name=b.id, fields=tuple(b.required_fields))
                for b in vertex.identity_funnel.branches
            )
        else:
            keys = (
                KeySet(name="identity", fields=tuple(vertex.hash_identity_properties)),
            )
        return KeyContract(
            vertex=vertex.name,
            source="derived",
            identity_fields=identity,
            keys=keys,
            alternates=alternates,
        )
    return KeyContract(
        vertex=vertex.name,
        source="stated",
        identity_fields=identity,
        keys=(KeySet(name="identity", fields=identity),),
        alternates=alternates,
    )

key_status(contract, values)

Which key of contract the non-empty values complete.

Source code in graflo/architecture/schema/key_contract.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def key_status(contract: KeyContract, values: Mapping[str, Any]) -> KeyStatus:
    """Which key of *contract* the non-empty *values* complete."""
    alternates = tuple(a for a in contract.alternates if _complete(a, values))
    if contract.source == "minted":
        return KeyStatus(state="keyless", alternates=alternates)
    for key in contract.keys:
        if _complete(key, values):
            return KeyStatus(state="primary", key=key, alternates=alternates)
    nearest = min(
        contract.keys,
        key=lambda k: sum(_empty(values.get(f)) for f in k.fields),
    )
    missing = tuple(f for f in nearest.fields if _empty(values.get(f)))
    if alternates:
        return KeyStatus(state="alternate", alternates=alternates, missing=missing)
    stated = any(not _empty(values.get(f)) for f in contract.key_fields)
    return KeyStatus(state="partial" if stated else "none", missing=missing)

primary_identity(vertex, contract, values)

The primary identity values complete, or None.

A derived key is digested only once a key set is complete: graflo's flat digest accepts some source fields missing, which would give a different key for the same entity.

Source code in graflo/architecture/schema/key_contract.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def primary_identity(
    vertex: Vertex, contract: KeyContract, values: Mapping[str, Any]
) -> dict[str, Any] | None:
    """The primary identity *values* complete, or ``None``.

    A derived key is digested only once a key set is complete: graflo's flat
    digest accepts some source fields missing, which would give a different key
    for the same entity.
    """
    status = key_status(contract, values)
    if status.state != "primary":
        return None
    if contract.source == "derived":
        digest = compute_vertex_identity(values, vertex)
        return None if digest is None else {contract.identity_fields[0]: digest}
    return {f: values[f] for f in contract.identity_fields}