Skip to content

graflo.architecture.evolution.aspects.provenance.vocabulary

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

SourceRecord is the record a fact was read from -- a document, a message, an API response, a table row. It carries the address of that record and a hash of its content, never the content: the text or the row stays where it was landed, and the graph points at it. Agent is who or what produced the record. Both are prov: types; wasDerivedFrom and wasAttributedTo are the PROV-O relations. mentions says a record names an entity, which holds whether or not the entity's facts were read from that record.

Attributes

AGENT = 'Agent' module-attribute

CONTENT_HASH = 'content_hash' module-attribute

MEDIA_TYPE = 'media_type' module-attribute

MENTIONS = 'mentions' module-attribute

RETRIEVED_AT = 'retrieved_at' module-attribute

ROLE = 'role' module-attribute

SOURCE_RECORD = 'SourceRecord' module-attribute

SOURCE_RECORD_ID = 'source_record_id' module-attribute

SOURCE_URI = 'source_uri' module-attribute

__all__ = ['AGENT', 'CONTENT_HASH', 'MEDIA_TYPE', 'MENTIONS', 'RETRIEVED_AT', 'ROLE', 'SOURCE_RECORD', 'SOURCE_RECORD_ID', 'SOURCE_URI', 'agent_vertex', 'attributed_to_edge', 'derived_from_edge', 'grounded_edge', 'mentions_edge', 'source_record_vertex'] module-attribute

Classes

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.",
    )

grounded_edge(source, target, relation, iri, description, *, properties=None)

A directed edge grounded in one IRI, as every aspect edge is.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def grounded_edge(
    source: str,
    target: str,
    relation: str,
    iri: str,
    description: str,
    *,
    properties: list[Field] | None = None,
) -> Edge:
    """A directed edge grounded in one IRI, as every aspect edge is."""
    return Edge(
        source=source,
        target=target,
        relation=relation,
        directed=True,
        description=description,
        semantics=Semantics(iri=iri, exact_match=[iri]),
        properties=list(properties or []),
    )

mentions_edge(target)

SourceRecord -mentions-> <target>.

Source code in graflo/architecture/evolution/aspects/provenance/vocabulary.py 
118
119
120
121
122
123
124
125
126
def mentions_edge(target: str) -> Edge:
    """``SourceRecord -mentions-> <target>``."""
    return grounded_edge(
        SOURCE_RECORD,
        target,
        MENTIONS,
        SCHEMA_MENTIONS,
        "An entity this record names.",
    )

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],
    )