graflo.link.records

What a producer states about entities, before anything is known about which they are.

A producer that holds only some properties of an entity states a :class:PartialRecord: a type and the properties it holds. It rarely holds a full key, so a record is not a vertex yet. A :class:RecordGraph holds one source's records and the relations stated between them, keyed by local refs.

Each record, stated property and relation may carry evidence: whatever the producer keeps about where it read it. graflo stores and returns it, and never interprets it.

Resolution (:mod:graflo.link.resolve) maps records to entities: it gives each record it can identify an identity (:class:Resolved) and stages the rest with the reason (:class:StagedRecord). A relation waits for its staged endpoints (:class:HeldRelation). Nothing staged is written to the graph. A relation between two records states two entities, so one whose endpoints resolve to the same entity is never written (:class:DroppedRelation). Candidate gathering (:mod:graflo.link.candidates) lists the entities a staged record may be; something else -- a pick, a rule, a person -- decides.

Attributes

CandidateLane = Literal['alternate', 'exact', 'key', 'normalized', 'alias', 'compatible', 'relation', 'metadata', 'context'] module-attribute

DropReason = Literal['same_entity'] module-attribute

ResolvedBy = Literal['key', 'alternate', 'minted'] module-attribute

StageReason = Literal['partial_key', 'no_key', 'unmatched', 'ambiguous', 'owned_unmatched', 'keyless_reference', 'hidden'] module-attribute

Classes

AttributeCheck

Bases: ConfigBaseModel

A stated property of the record against the candidate's stored value.

The stored value itself is never carried.

Source code in graflo/link/records.py

class AttributeCheck(ConfigBaseModel):
    """A stated property of the record against the candidate's stored value.

    The stored value itself is never carried.
    """

    field: str
    level: MatchLevel = Field(
        description=(
            "``exact``, ``compatible`` (one is a less specific form of the other), "
            "``differs`` (not evidence against), ``conflict`` (evidence against: "
            "an exclusive field) or ``unknown`` (nothing stored to compare with). "
            "See :mod:`graflo.architecture.schema.compare`."
        )
    )
    exclusive: bool = Field(
        description="Whether a difference on the field is a ``conflict``."
    )

Attributes

exclusive = Field(description='Whether a difference on the field is a ``conflict``.') class-attribute instance-attribute

field instance-attribute

level = Field(description='``exact``, ``compatible`` (one is a less specific form of the other), ``differs`` (not evidence against), ``conflict`` (evidence against: an exclusive field) or ``unknown`` (nothing stored to compare with). See :mod:`graflo.architecture.schema.compare`.') class-attribute instance-attribute

CandidateHit

Bases: ConfigBaseModel

One way an entity was found for a staged record.

Each lane carries its own fields and no others: relation the relation and the anchor, metadata a role, context its anchor (a record's ref, or the role subject) and the level of its label, every other lane what matched, and alias and compatible also how many entities that reached.

Source code in graflo/link/records.py

class CandidateHit(ConfigBaseModel):
    """One way an entity was found for a staged record.

    Each lane carries its own fields and no others: ``relation`` the relation
    and the anchor, ``metadata`` a role, ``context`` its anchor (a record's ref,
    or the role ``subject``) and the level of its label, every other lane what
    matched, and ``alias`` and ``compatible`` also how many entities that reached.
    """

    lane: CandidateLane = Field(
        description=(
            "How it was found. ``alternate``: one of the entities a stated "
            "secondary identity matched. ``exact``, ``key``, ``normalized``, "
            "``alias``, ``compatible``: a lookup over the record's stated labels "
            "and the caller's lookup text. ``relation``: linked by a stated "
            "relation to a resolved record of the same source. ``metadata``: "
            "named by the source's metadata, added by the caller. ``context``: a "
            "neighbour of the record's type, one hop from an entity the source "
            "resolved by key or a subject it is known to concern."
        )
    )
    matched: str | None = Field(
        default=None,
        description="The words or values of the record it was found by.",
    )
    relation: str | None = Field(
        default=None, description="``relation`` lane: the relation followed."
    )
    anchor: str | None = Field(
        default=None,
        description="``relation`` lane: ref of the resolved record it is linked to.",
    )
    role: str | None = Field(
        default=None,
        description="``metadata`` lane: a role the caller assigns; ``context``: ``subject``.",
    )
    targets: int | None = Field(
        default=None,
        description="``alias`` lane: how many entities the alias reached in the "
        "reader's view; ``compatible`` lane: how many entities of the type hold the "
        "matched form.",
    )
    level: MatchLevel | None = Field(
        default=None,
        description="``context`` lane: the record's stated label against the entity's.",
    )

    @model_validator(mode="after")
    def _fields_of_its_lane(self) -> CandidateHit:
        if self.lane == "relation":
            allowed = [{"relation", "anchor"}]
        elif self.lane == "metadata":
            allowed = [{"role"}]
        elif self.lane == "context":
            allowed = [{"anchor", "level"}, {"role", "level"}]
        elif self.lane in ("alias", "compatible"):
            allowed = [{"matched", "targets"}, {"matched"}]
        else:
            allowed = [{"matched"}]
        stated = {
            name
            for name in ("matched", "relation", "anchor", "role", "targets", "level")
            if getattr(self, name) is not None
        }
        if stated not in allowed:
            raise ValueError(
                f"a {self.lane} hit carries {' or '.join(str(sorted(a)) for a in allowed)}, "
                f"got {sorted(stated)}"
            )
        return self

Attributes

anchor = Field(default=None, description='``relation`` lane: ref of the resolved record it is linked to.') class-attribute instance-attribute

lane = Field(description="How it was found. ``alternate``: one of the entities a stated secondary identity matched. ``exact``, ``key``, ``normalized``, ``alias``, ``compatible``: a lookup over the record's stated labels and the caller's lookup text. ``relation``: linked by a stated relation to a resolved record of the same source. ``metadata``: named by the source's metadata, added by the caller. ``context``: a neighbour of the record's type, one hop from an entity the source resolved by key or a subject it is known to concern.") class-attribute instance-attribute

level = Field(default=None, description="``context`` lane: the record's stated label against the entity's.") class-attribute instance-attribute

matched = Field(default=None, description='The words or values of the record it was found by.') class-attribute instance-attribute

relation = Field(default=None, description='``relation`` lane: the relation followed.') class-attribute instance-attribute

role = Field(default=None, description='``metadata`` lane: a role the caller assigns; ``context``: ``subject``.') class-attribute instance-attribute

targets = Field(default=None, description="``alias`` lane: how many entities the alias reached in the reader's view; ``compatible`` lane: how many entities of the type hold the matched form.") class-attribute instance-attribute

DroppedRelation

Bases: ConfigBaseModel

A relation resolution never writes, and why.

Source code in graflo/link/records.py

class DroppedRelation(ConfigBaseModel):
    """A relation resolution never writes, and why."""

    source_id: str
    relation: RecordRelation
    reason: DropReason = Field(
        description=(
            "``same_entity``: its two records resolved to one entity (same type "
            "and identity); a relation between two records states two entities, "
            "so it is not written as a self-loop."
        )
    )

Attributes

reason = Field(description='``same_entity``: its two records resolved to one entity (same type and identity); a relation between two records states two entities, so it is not written as a self-loop.') class-attribute instance-attribute

relation instance-attribute

source_id instance-attribute

EntityCandidate

Bases: ConfigBaseModel

An entity a staged record may be. A candidate never closes a record.

Source code in graflo/link/records.py

class EntityCandidate(ConfigBaseModel):
    """An entity a staged record may be. A candidate never closes a record."""

    vertex_type: str
    key: dict[str, Any] = Field(description="Its identity fields and values.")
    label: str | None = Field(default=None, description="Its label, if it has one.")
    lanes: list[CandidateHit] = Field(
        min_length=1, description="Every way it was found, best first."
    )
    attributes: list[AttributeCheck] = Field(
        default_factory=list,
        description=(
            "The record's other comparable stated properties against the "
            "candidate's stored values. A key in conflict never appears: it "
            "leaves the candidate out."
        ),
    )
    support: SupportFeatures = Field(
        default_factory=SupportFeatures,
        description="The features that identify it, read by ``support()``.",
    )

Attributes

attributes = Field(default_factory=list, description="The record's other comparable stated properties against the candidate's stored values. A key in conflict never appears: it leaves the candidate out.") class-attribute instance-attribute

key = Field(description='Its identity fields and values.') class-attribute instance-attribute

label = Field(default=None, description='Its label, if it has one.') class-attribute instance-attribute

lanes = Field(min_length=1, description='Every way it was found, best first.') class-attribute instance-attribute

support = Field(default_factory=SupportFeatures, description='The features that identify it, read by ``support()``.') class-attribute instance-attribute

vertex_type instance-attribute

HeldRelation

Bases: ConfigBaseModel

A relation waiting for staged endpoints.

Source code in graflo/link/records.py

class HeldRelation(ConfigBaseModel):
    """A relation waiting for staged endpoints."""

    source_id: str
    relation: RecordRelation
    waiting_on: list[str] = Field(description="Refs of its staged endpoints.")

Attributes

relation instance-attribute

source_id instance-attribute

waiting_on = Field(description='Refs of its staged endpoints.') class-attribute instance-attribute

KnownSubject

Bases: ConfigBaseModel

An entity a source is known to concern, from its metadata, not its text.

Source code in graflo/link/records.py

class KnownSubject(ConfigBaseModel):
    """An entity a source is known to concern, from its metadata, not its text."""

    vertex_type: str
    key: dict[str, Any] = Field(description="Its identity fields and values.")

Attributes

key = Field(description='Its identity fields and values.') class-attribute instance-attribute

vertex_type instance-attribute

PartialRecord

Bases: ConfigBaseModel

An entity a source states: its type and the properties it holds.

Source code in graflo/link/records.py

class PartialRecord(ConfigBaseModel):
    """An entity a source states: its type and the properties it holds."""

    ref: str = Field(description="Local id, unique within its graph.")
    type: str = Field(description="Vertex type.")
    may_create: bool = Field(
        description=(
            "The record may create its entity when absent, rather than only "
            "reference one assumed to exist. Decides only whether a type that "
            "takes no key from the data is minted."
        )
    )
    props: dict[str, StatedProp] = Field(default_factory=dict)
    evidence: dict[str, Any] | None = Field(default=None, description=_EVIDENCE)

    def values(self) -> dict[str, Any]:
        """Stated values by property name."""
        return {name: prop.value for name, prop in self.props.items()}

Attributes

evidence = Field(default=None, description=_EVIDENCE) class-attribute instance-attribute

may_create = Field(description='The record may create its entity when absent, rather than only reference one assumed to exist. Decides only whether a type that takes no key from the data is minted.') class-attribute instance-attribute

props = Field(default_factory=dict) class-attribute instance-attribute

ref = Field(description='Local id, unique within its graph.') class-attribute instance-attribute

type = Field(description='Vertex type.') class-attribute instance-attribute

Methods:

values()

Stated values by property name.

Source code in graflo/link/records.py

def values(self) -> dict[str, Any]:
    """Stated values by property name."""
    return {name: prop.value for name, prop in self.props.items()}

RecordGraph

Bases: ConfigBaseModel

One source's records and the relations between them.

Source code in graflo/link/records.py

class RecordGraph(ConfigBaseModel):
    """One source's records and the relations between them."""

    source_id: str
    schema_ref: str = Field(description="``<schema name>@<version>`` read against.")
    reference_time: datetime | None = None
    records: list[PartialRecord] = Field(default_factory=list)
    relations: list[RecordRelation] = Field(default_factory=list)

    @model_validator(mode="after")
    def _refs_resolve(self) -> RecordGraph:
        refs = [r.ref for r in self.records]
        if len(set(refs)) != len(refs):
            raise ValueError(f"{self.source_id}: record refs are not unique")
        known = set(refs)
        for relation in self.relations:
            for ref in (relation.source, relation.target):
                if ref not in known:
                    raise ValueError(
                        f"{self.source_id}: relation {relation.edge_id} names "
                        f"unknown record {ref!r}"
                    )
        return self

Attributes

records = Field(default_factory=list) class-attribute instance-attribute

reference_time = None class-attribute instance-attribute

relations = Field(default_factory=list) class-attribute instance-attribute

schema_ref = Field(description='``<schema name>@<version>`` read against.') class-attribute instance-attribute

source_id instance-attribute

RecordRelation

Bases: ConfigBaseModel

An edge a source states between two records.

Source code in graflo/link/records.py

class RecordRelation(ConfigBaseModel):
    """An edge a source states between two records."""

    edge_id: EdgeId
    source: str = Field(description="Ref of the source record.")
    target: str = Field(description="Ref of the target record.")
    props: dict[str, StatedProp] = Field(default_factory=dict)
    evidence: dict[str, Any] | None = Field(default=None, description=_EVIDENCE)

    def values(self) -> dict[str, Any]:
        """Stated values by property name."""
        return {name: prop.value for name, prop in self.props.items()}

Attributes

edge_id instance-attribute

evidence = Field(default=None, description=_EVIDENCE) class-attribute instance-attribute

props = Field(default_factory=dict) class-attribute instance-attribute

source = Field(description='Ref of the source record.') class-attribute instance-attribute

target = Field(description='Ref of the target record.') class-attribute instance-attribute

Methods:

values()

Stated values by property name.

Source code in graflo/link/records.py

def values(self) -> dict[str, Any]:
    """Stated values by property name."""
    return {name: prop.value for name, prop in self.props.items()}

Resolution

Bases: ConfigBaseModel

The outcome of resolving record graphs.

Source code in graflo/link/records.py

class Resolution(ConfigBaseModel):
    """The outcome of resolving record graphs."""

    graphs: list[RecordGraph] = Field(default_factory=list)
    resolved: list[Resolved] = Field(default_factory=list)
    staged: Staged = Field(default_factory=Staged)
    dropped: list[DroppedRelation] = Field(
        default_factory=list,
        description="Relations between resolved records that are never written.",
    )

    def identity_of(self, source_id: str, ref: str) -> dict[str, Any] | None:
        """Identity of record *ref* of *source_id*, when resolved."""
        for item in self.resolved:
            if item.source_id == source_id and item.ref == ref:
                return item.identity
        return None

Attributes

dropped = Field(default_factory=list, description='Relations between resolved records that are never written.') class-attribute instance-attribute

graphs = Field(default_factory=list) class-attribute instance-attribute

resolved = Field(default_factory=list) class-attribute instance-attribute

staged = Field(default_factory=Staged) class-attribute instance-attribute

Methods:

identity_of(source_id, ref)

Identity of record ref of source_id, when resolved.

Source code in graflo/link/records.py

def identity_of(self, source_id: str, ref: str) -> dict[str, Any] | None:
    """Identity of record *ref* of *source_id*, when resolved."""
    for item in self.resolved:
        if item.source_id == source_id and item.ref == ref:
            return item.identity
    return None

Resolved

Bases: ConfigBaseModel

A record identified as one entity.

Source code in graflo/link/records.py

class Resolved(ConfigBaseModel):
    """A record identified as one entity."""

    source_id: str
    ref: str
    type: str
    identity: dict[str, Any]
    by: ResolvedBy = Field(
        description=(
            "``key``: the record states a complete primary key. ``alternate``: a "
            "complete secondary identity found exactly one entity. ``minted``: a "
            "keyless entity the record may create."
        )
    )
    alternate: str | None = Field(
        default=None, description="Secondary identity the lookup used."
    )
    found_in: Literal["batch", "graph"] | None = Field(
        default=None,
        description="Where a lookup found the entity: another record or the graph.",
    )
    owned: bool = Field(
        default=False,
        description="The type is ingested by a connector: the record links, never writes.",
    )
    link_only: bool = Field(
        default=False,
        description=(
            "Found in the graph outside what the write may change (the "
            "resolution's ``writable`` filter): the record links, never writes."
        ),
    )

Attributes

alternate = Field(default=None, description='Secondary identity the lookup used.') class-attribute instance-attribute

by = Field(description='``key``: the record states a complete primary key. ``alternate``: a complete secondary identity found exactly one entity. ``minted``: a keyless entity the record may create.') class-attribute instance-attribute

found_in = Field(default=None, description='Where a lookup found the entity: another record or the graph.') class-attribute instance-attribute

identity instance-attribute

link_only = Field(default=False, description="Found in the graph outside what the write may change (the resolution's ``writable`` filter): the record links, never writes.") class-attribute instance-attribute

owned = Field(default=False, description='The type is ingested by a connector: the record links, never writes.') class-attribute instance-attribute

ref instance-attribute

source_id instance-attribute

type instance-attribute

Staged

Bases: ConfigBaseModel

Everything resolution could not write yet.

Source code in graflo/link/records.py

class Staged(ConfigBaseModel):
    """Everything resolution could not write yet."""

    records: list[StagedRecord] = Field(default_factory=list)
    relations: list[HeldRelation] = Field(default_factory=list)

Attributes

records = Field(default_factory=list) class-attribute instance-attribute

relations = Field(default_factory=list) class-attribute instance-attribute

StagedCandidates

Bases: ConfigBaseModel

The candidates of one staged record, best first.

Source code in graflo/link/records.py

class StagedCandidates(ConfigBaseModel):
    """The candidates of one staged record, best first."""

    source_id: str
    record: PartialRecord
    reason: StageReason
    candidates: list[EntityCandidate] = Field(default_factory=list)
    truncated: bool = Field(
        default=False, description="More were found than the cap let through."
    )

Attributes

candidates = Field(default_factory=list) class-attribute instance-attribute

reason instance-attribute

record instance-attribute

source_id instance-attribute

truncated = Field(default=False, description='More were found than the cap let through.') class-attribute instance-attribute

StagedRecord

Bases: ConfigBaseModel

A record not identified yet, and why.

Source code in graflo/link/records.py

class StagedRecord(ConfigBaseModel):
    """A record not identified yet, and why."""

    source_id: str
    record: PartialRecord
    reason: StageReason = Field(
        description=(
            "``partial_key``: some key fields stated, no key complete. ``no_key``: "
            "no key field stated. ``unmatched``: a secondary identity found "
            "nothing. ``ambiguous``: it found several entities. "
            "``owned_unmatched``: a connector-owned entity the graph does not "
            "hold yet. ``keyless_reference``: a keyless type the record only "
            "references. ``hidden``: its key names an entity the resolution's "
            "guard hides, and its write would not add to the guarded fields, so "
            "it is neither written nor linked."
        )
    )
    matches: int = Field(default=0, description="Entities a lookup found.")

Attributes

matches = Field(default=0, description='Entities a lookup found.') class-attribute instance-attribute

reason = Field(description="``partial_key``: some key fields stated, no key complete. ``no_key``: no key field stated. ``unmatched``: a secondary identity found nothing. ``ambiguous``: it found several entities. ``owned_unmatched``: a connector-owned entity the graph does not hold yet. ``keyless_reference``: a keyless type the record only references. ``hidden``: its key names an entity the resolution's guard hides, and its write would not add to the guarded fields, so it is neither written nor linked.") class-attribute instance-attribute

record instance-attribute

source_id instance-attribute

StatedProp

Bases: ConfigBaseModel

A property value a record states.

Source code in graflo/link/records.py

class StatedProp(ConfigBaseModel):
    """A property value a record states."""

    value: Any
    evidence: dict[str, Any] | None = Field(default=None, description=_EVIDENCE)

Attributes

evidence = Field(default=None, description=_EVIDENCE) class-attribute instance-attribute

value instance-attribute

SupportFeatures

Bases: ConfigBaseModel

What identifies a candidate, as :func:graflo.link.candidates.support reads it.

Source code in graflo/link/records.py

class SupportFeatures(ConfigBaseModel):
    """What identifies a candidate, as :func:`graflo.link.candidates.support` reads it."""

    identifying_lane: bool = Field(
        default=False,
        description="Found by ``key``, ``alternate``, ``exact`` or ``normalized``.",
    )
    exclusive_agreement: bool = Field(
        default=False, description="An exclusive field stated and stored alike."
    )
    unique_alias: bool = Field(
        default=False,
        description="Found by an alias that names only it in the reader's view.",
    )
    unique_compatible: bool = Field(
        default=False,
        description="Found by a less specific form of one of its values that no "
        "other entity of its type holds in the reader's view.",
    )
    conflict: bool = Field(
        default=False, description="A stated field is evidence against it."
    )

Attributes

conflict = Field(default=False, description='A stated field is evidence against it.') class-attribute instance-attribute

exclusive_agreement = Field(default=False, description='An exclusive field stated and stored alike.') class-attribute instance-attribute

identifying_lane = Field(default=False, description='Found by ``key``, ``alternate``, ``exact`` or ``normalized``.') class-attribute instance-attribute

unique_alias = Field(default=False, description="Found by an alias that names only it in the reader's view.") class-attribute instance-attribute

unique_compatible = Field(default=False, description="Found by a less specific form of one of its values that no other entity of its type holds in the reader's view.") class-attribute instance-attribute