Skip to content

graflo.architecture.evolution.aspects.provenance

The provenance aspect: where a fact came from, and who produced it (PROV-O).

Modules:

Name Description
plan

Planning the provenance aspect: SourceRecord, Agent and their edges.
vocabulary

The provenance aspect's types: where a fact came from, and who produced it.

Attributes

AGENT = 'Agent' module-attribute

SOURCE_RECORD = 'SourceRecord' module-attribute

__all__ = ['AGENT', 'SOURCE_RECORD', 'agent_vertex', 'attributed_to_edge', 'derived_from_edge', 'plan_provenance', 'source_record_vertex'] module-attribute

Functions:

agent_vertex()

Who or what acted or measured: a person, a team, a sensor, a system.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def agent_vertex() -> Vertex:
    """Who or what acted or measured: a person, a team, a sensor, a system."""
    return Vertex(
        name=AGENT,
        description="Who or what acted or measured.",
        semantics=Semantics(iri=PROV_AGENT, exact_match=[PROV_AGENT, SOSA_SENSOR]),
        properties=[
            Field(name="agent_id", type=FieldType.STRING),
            Field(
                name="label",
                type=FieldType.STRING,
                semantics=FieldSemantics(iri=RDFS_LABEL),
            ),
            Field(name="agent_kind", type=FieldType.STRING),
        ],
        identity=["agent_id"],
    )

attributed_to_edge()

SourceRecord -wasAttributedTo-> Agent, with the agent's role on the edge.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
129
130
131
132
133
134
135
136
137
138
def attributed_to_edge() -> Edge:
    """``SourceRecord -wasAttributedTo-> Agent``, with the agent's role on the edge."""
    return grounded_edge(
        SOURCE_RECORD,
        AGENT,
        "wasAttributedTo",
        PROV_WAS_ATTRIBUTED_TO,
        "Who or what produced this record, and in which role.",
        properties=[Field(name=ROLE, type=FieldType.STRING)],
    )

derived_from_edge(source)

<source> -wasDerivedFrom-> SourceRecord.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
107
108
109
110
111
112
113
114
115
def derived_from_edge(source: str) -> Edge:
    """``<source> -wasDerivedFrom-> SourceRecord``."""
    return grounded_edge(
        source,
        SOURCE_RECORD,
        "wasDerivedFrom",
        PROV_WAS_DERIVED_FROM,
        "What this fact was read from.",
    )

plan_provenance(manifest, *, attach=(), mentions=())

The ops that make lineage expressible in manifest.

SourceRecord -wasAttributedTo-> Agent is emitted even when there is nothing else to attach: it is what makes the schema able to say where a fact came from, which is the question the world-model profile asks.

The vocabulary is joined with what the manifest has (:func:~graflo.architecture.evolution.aspects.join.join): a SourceRecord or Agent it already declares keeps its own properties and gains the aspect's missing ones, so manifests that carry the vocabulary, such as source kits, compose and lift in any order.

Parameters:

Name Type Description Default
manifest GraphManifest

the manifest to lift, already finish_init()-ed.

 required
attach Iterable[str]

vertex type names that gain a wasDerivedFrom edge to SourceRecord, in this order. The caller vouches that each one exists or is added by the same change set.

 ()
mentions Iterable[str]

vertex type names SourceRecord gains a mentions edge to, in this order, on the same terms.

 ()

Raises:

Type Description
LiftError

the manifest has no schema, or declares part of the vocabulary in a way that contradicts it.
Source code in graflo/architecture/evolution/aspects/provenance/plan.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
def plan_provenance(
    manifest: GraphManifest,
    *,
    attach: Iterable[str] = (),
    mentions: Iterable[str] = (),
) -> list[ManifestOp]:
    """The ops that make lineage expressible in *manifest*.

    ``SourceRecord -wasAttributedTo-> Agent`` is emitted even when there is
    nothing else to attach: it is what makes the schema *able* to say where a
    fact came from, which is the question the world-model profile asks.

    The vocabulary is joined with what the manifest has
    (:func:`~graflo.architecture.evolution.aspects.join.join`): a
    ``SourceRecord`` or ``Agent`` it already declares keeps its own properties
    and gains the aspect's missing ones, so manifests that carry the
    vocabulary, such as source kits, compose and lift in any order.

    Args:
        manifest: the manifest to lift, already ``finish_init()``-ed.
        attach: vertex type names that gain a ``wasDerivedFrom`` edge to
            ``SourceRecord``, in this order. The caller vouches that each one
            exists or is added by the same change set.
        mentions: vertex type names ``SourceRecord`` gains a ``mentions`` edge to,
            in this order, on the same terms.

    Raises:
        LiftError: the manifest has no schema, or declares part of the
            vocabulary in a way that contradicts it.
    """
    schema = manifest.graph_schema
    if schema is None:
        raise LiftError(
            "cannot lift a manifest with no schema block: there are no types to "
            "attach provenance to"
        )
    return join(
        schema.core_schema,
        vertices=[vocab.source_record_vertex(), vocab.agent_vertex()],
        edges=[
            *(vocab.derived_from_edge(name) for name in attach),
            *(vocab.mentions_edge(name) for name in mentions),
            vocab.attributed_to_edge(),
        ],
    )

source_record_vertex()

The record a fact was read from: its address and hash, never its content.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def source_record_vertex() -> Vertex:
    """The record a fact was read from: its address and hash, never its content."""
    return Vertex(
        name=SOURCE_RECORD,
        description=(
            "The record a fact was read from: a document, a message, an API "
            "response, a table row. Carries where the record is and a hash of "
            "what it held, not the content itself."
        ),
        semantics=Semantics(iri=PROV_ENTITY, exact_match=[PROV_ENTITY]),
        properties=[
            Field(name=SOURCE_RECORD_ID, type=FieldType.STRING),
            Field(name=SOURCE_URI, type=FieldType.STRING),
            Field(name=MEDIA_TYPE, type=FieldType.STRING),
            Field(name=CONTENT_HASH, type=FieldType.STRING),
            Field(
                name=RETRIEVED_AT,
                type=FieldType.DATETIME,
                semantics=FieldSemantics(iri=PROV_GENERATED_AT),
            ),
        ],
        identity=[SOURCE_RECORD_ID],
    )