Record provenance in a graph with the PROV-O terms and the secoro prov extension.

Functions:

• add_activity –

  Add a prov:Activity of the given kinds: what it used, who ran it and when.

• add_agent –

  Add a prov:Agent of the given kinds.

• add_entity –

  Type a node as prov:Entity.

• add_file_entity –

  Add the file at PATH as a prov:Entity named by its content, the get_ni_iri of its bytes.

• add_relative_location –

  Add a prov-ext:RelativeLocation: a path under a root directory, the only machine path.

• create_bundle –

  A named graph for one tool's provenance: a prov:Bundle named urn:uuid:, attributed to it.

• get_git_info –

  Read the revision and repository of a checkout, for a source that is not a package.

• get_ni_iri –

  The IRI of a file's content, named by its hash: the same bytes have the same name.

• get_pkg_info –

  Read off an installed package what describes it, from its metadata and its source.

• load_execution_prov –

  Add a prov-ext:Execution of the used entities by a software package.

• load_pkg_prov –

  Add a software package as a prov:SoftwareAgent described with schema.org terms.

• load_sampling_prov –

  Add a prov-ext:Generalization that sampled a quantity into generated entities.

• load_transformation_prov –

  Add a prov-ext:Transformation of source entities into target entities.

• resolve_location –

  The IRI a node is prov:atLocation, or None when it has no location.

add_activity

add_activity(graph, activity_id, additional_types, used, agent_id, started, ended=None)

Add a prov:Activity of the given kinds: what it used, who ran it and when.

The used entities are the caller's own nodes and are not typed here.

Parameters:

• graph (Graph) –

  RDF graph to add the activity to

• activity_id (URIRef) –

  URI of the activity

• additional_types (Iterable[URIRef]) –

  its kinds besides prov:Activity, e.g. URI_PROV_EXT_TYPE_EXECUTION

• used (Iterable[URIRef]) –

  URIs of the entities the activity used

• agent_id (URIRef) –

  URI of the agent the activity is associated with

• started (datetime) –

  start time, stored as prov:startedAtTime

• ended (datetime | None, default: None ) –

  end time, stored as prov:endedAtTime; None for an activity still running

Source code in src/rdf_utils/models/prov.py

def add_activity(
    graph: Graph,
    activity_id: URIRef,
    additional_types: Iterable[URIRef],
    used: Iterable[URIRef],
    agent_id: URIRef,
    started: datetime,
    ended: datetime | None = None,
) -> None:
    """Add a `prov:Activity` of the given kinds: what it used, who ran it and when.

    The used entities are the caller's own nodes and are not typed here.

    Parameters:
        graph: RDF graph to add the activity to
        activity_id: URI of the activity
        additional_types: its kinds besides `prov:Activity`, e.g. `URI_PROV_EXT_TYPE_EXECUTION`
        used: URIs of the entities the activity used
        agent_id: URI of the agent the activity is associated with
        started: start time, stored as `prov:startedAtTime`
        ended: end time, stored as `prov:endedAtTime`; None for an activity still running
    """
    graph.add((activity_id, RDF.type, PROV.Activity))
    for type_id in additional_types:
        graph.add((activity_id, RDF.type, type_id))
    for entity_id in used:
        graph.add((activity_id, PROV.used, entity_id))
    graph.add((activity_id, PROV.wasAssociatedWith, agent_id))
    graph.add((activity_id, PROV.startedAtTime, Literal(started)))
    if ended is not None:
        graph.add((activity_id, PROV.endedAtTime, Literal(ended)))

add_agent

add_agent(graph, agent_id, additional_types, name=None, acted_on_behalf_of=None)

Add a prov:Agent of the given kinds.

Parameters:

• graph (Graph) –

  RDF graph to add the agent to

• agent_id (URIRef) –

  URI of the agent

• additional_types (Iterable[URIRef]) –

  its kinds besides prov:Agent, e.g. prov:SoftwareAgent or prov:Person

• name (str | None, default: None ) –

  stored as schema:name, which a software agent must carry

• acted_on_behalf_of (URIRef | None, default: None ) –

  URI of the agent this one acted for, stored as prov:actedOnBehalfOf

Source code in src/rdf_utils/models/prov.py

def add_agent(
    graph: Graph,
    agent_id: URIRef,
    additional_types: Iterable[URIRef],
    name: str | None = None,
    acted_on_behalf_of: URIRef | None = None,
) -> None:
    """Add a `prov:Agent` of the given kinds.

    Parameters:
        graph: RDF graph to add the agent to
        agent_id: URI of the agent
        additional_types: its kinds besides `prov:Agent`, e.g. `prov:SoftwareAgent` or `prov:Person`
        name: stored as `schema:name`, which a software agent must carry
        acted_on_behalf_of: URI of the agent this one acted for, stored as `prov:actedOnBehalfOf`
    """
    graph.add((agent_id, RDF.type, PROV.Agent))
    for type_id in additional_types:
        graph.add((agent_id, RDF.type, type_id))
    if name is not None:
        graph.add((agent_id, SDO.name, Literal(name)))
    if acted_on_behalf_of is not None:
        graph.add((agent_id, PROV.actedOnBehalfOf, acted_on_behalf_of))

add_entity

add_entity(graph, entity_id)

Type a node as prov:Entity.

Parameters:

• graph (Graph) –

  RDF graph to add the entity to

• entity_id (URIRef) –

  URI of the entity

Source code in src/rdf_utils/models/prov.py

def add_entity(graph: Graph, entity_id: URIRef) -> None:
    """Type a node as `prov:Entity`.

    Parameters:
        graph: RDF graph to add the entity to
        entity_id: URI of the entity
    """
    graph.add((entity_id, RDF.type, PROV.Entity))

add_file_entity

add_file_entity(graph, root, path, generated_by=None, generated_at=None, sha256=None, modified_at=None, fmt=None, additional_types=())

Add the file at PATH as a prov:Entity named by its content, the get_ni_iri of its bytes.

It is located by its path in a prov-ext:RelativeLocation under ROOT, named by the UUIDv5 of its name and that path: a location is a prov:Entity, which the PROV shapes require to be an IRI, and one file at one place is one location however often it is added.

Parameters:

• graph (Graph) –

  RDF graph to add the file to

• root (str | Path | None) –

  directory the file is located under; None locates it at PATH as given, a path or an IRI

• path (str | Path) –

  the file, relative to ROOT or absolute

• generated_by (URIRef | None, default: None ) –

  URI of the activity that generated it, stored as prov:wasGeneratedBy

• generated_at (datetime | None, default: None ) –

  time the file was complete, stored as prov:generatedAtTime; with GENERATED_BY and none given, the file's modification time when its bytes are read

• sha256 (str | None, default: None ) –

  hex SHA-256 digest of its bytes, for a file that is not read here

• modified_at (datetime | None, default: None ) –

  time the file last changed, stored as dcterms:modified

• fmt (str | None, default: None ) –

  media type of the file, stored as dcterms:format

• additional_types (Iterable[URIRef], default: () ) –

  its kinds besides prov:Entity

Returns:

• URIRef –

  URI of the file

Source code in src/rdf_utils/models/prov.py

def add_file_entity(
    graph: Graph,
    root: str | Path | None,
    path: str | Path,
    generated_by: URIRef | None = None,
    generated_at: datetime | None = None,
    sha256: str | None = None,
    modified_at: datetime | None = None,
    fmt: str | None = None,
    additional_types: Iterable[URIRef] = (),
) -> URIRef:
    """Add the file at PATH as a `prov:Entity` named by its content, the `get_ni_iri` of its bytes.

    It is located by its path in a `prov-ext:RelativeLocation` under ROOT, named by the UUIDv5 of
    its name and that path: a location is a `prov:Entity`, which the PROV shapes require to be an
    IRI, and one file at one place is one location however often it is added.

    Parameters:
        graph: RDF graph to add the file to
        root: directory the file is located under; None locates it at PATH as given, a path or
            an IRI
        path: the file, relative to ROOT or absolute
        generated_by: URI of the activity that generated it, stored as `prov:wasGeneratedBy`
        generated_at: time the file was complete, stored as `prov:generatedAtTime`; with
            GENERATED_BY and none given, the file's modification time when its bytes are read
        sha256: hex SHA-256 digest of its bytes, for a file that is not read here
        modified_at: time the file last changed, stored as `dcterms:modified`
        fmt: media type of the file, stored as `dcterms:format`
        additional_types: its kinds besides `prov:Entity`

    Returns:
        URI of the file
    """
    if root is not None:
        path = Path(root, path).resolve()
    file_id = get_ni_iri(path, sha256)
    if root is None:
        location = _location_iri(path)
    else:
        rel_path = PurePath(os.path.relpath(path, Path(root).resolve())).as_posix()
        location = URIRef(uuid5(uuid5(NAMESPACE_URL, str(file_id)), rel_path).urn)
        add_relative_location(graph, location, rel_path, root)
    add_entity(graph, file_id)
    for type_id in additional_types:
        graph.add((file_id, RDF.type, type_id))
    graph.add((file_id, PROV.atLocation, location))
    if generated_by is not None:
        graph.add((file_id, PROV.wasGeneratedBy, generated_by))
        if generated_at is None and sha256 is None:
            generated_at = datetime.fromtimestamp(Path(path).stat().st_mtime, UTC)
    if generated_at is not None:
        graph.add((file_id, PROV.generatedAtTime, Literal(generated_at)))
    if modified_at is not None:
        graph.add((file_id, DCTERMS.modified, Literal(modified_at)))
    if fmt is not None:
        graph.add((file_id, DCTERMS.format, Literal(fmt)))
    return file_id

add_relative_location

add_relative_location(graph, location_id, rel_path, root)

Add a prov-ext:RelativeLocation: a path under a root directory, the only machine path.

Parameters:

• graph (Graph) –

  RDF graph to add the location to

• location_id (URIRef) –

  URI of the relative location, for a file entity's prov:atLocation

• rel_path (str | PurePath) –

  file path from the root; stored percent-encoded as an RFC 3986 relative-path reference in prov-ext:rel-path, so spaces and % survive resolution

• root (str | URIRef) –

  directory path, or the IRI of the location the path is relative to: a directory IRI ending in / or another relative location; stored as a prov:Location

Raises:

• ValueError –

  when rel_path is absolute

Source code in src/rdf_utils/models/prov.py

def add_relative_location(
    graph: Graph, location_id: URIRef, rel_path: str | PurePath, root: str | URIRef
) -> None:
    """Add a `prov-ext:RelativeLocation`: a path under a root directory, the only machine path.

    Parameters:
        graph: RDF graph to add the location to
        location_id: URI of the relative location, for a file entity's `prov:atLocation`
        rel_path: file path from the root; stored percent-encoded as an RFC 3986 relative-path
            reference in `prov-ext:rel-path`, so spaces and `%` survive resolution
        root: directory path, or the IRI of the location the path is relative to: a directory
            IRI ending in `/` or another relative location; stored as a `prov:Location`

    Raises:
        ValueError: when `rel_path` is absolute
    """
    if PurePath(rel_path).is_absolute():
        raise ValueError(f"'{rel_path}' is not a relative path -- it is absolute")
    # RFC 3986 drops a base's last segment when resolving, so a directory's IRI ends in '/'.
    root_iri = root if isinstance(root, URIRef) else URIRef(f"{Path(root).resolve().as_uri()}/")
    graph.add((root_iri, RDF.type, PROV.Location))
    for type_id in (URI_PROV_EXT_TYPE_RELATIVE_LOCATION, PROV.Location, PROV.Entity):
        graph.add((location_id, RDF.type, type_id))
    reference = quote(PurePath(rel_path).as_posix(), safe="/")
    graph.add((location_id, URI_PROV_EXT_PRED_REL_PATH, Literal(reference, datatype=XSD.anyURI)))
    graph.add((location_id, PROV.atLocation, root_iri))

create_bundle

create_bundle(agent_id)

A named graph for one tool's provenance: a prov:Bundle named urn:uuid:, attributed to it.

Parameters:

• agent_id (URIRef) –

  URI of the agent that writes the bundle, stored as prov:wasAttributedTo

Source code in src/rdf_utils/models/prov.py

def create_bundle(agent_id: URIRef) -> Graph:
    """A named graph for one tool's provenance: a `prov:Bundle` named `urn:uuid:`, attributed to it.

    Parameters:
        agent_id: URI of the agent that writes the bundle, stored as `prov:wasAttributedTo`
    """
    bundle = Graph(identifier=URIRef(uuid4().urn))
    bundle.add((bundle.identifier, RDF.type, PROV.Bundle))
    bundle.add((bundle.identifier, PROV.wasAttributedTo, agent_id))
    return bundle

get_git_info

get_git_info(path)

Read the revision and repository of a checkout, for a source that is not a package.

Parameters:

• path (Path) –

  any path inside the worktree; the repository owning it is the one read

Returns:

• tuple[str | None, str | None] –

  Revision, suffixed -dirty when the worktree has changes, and repository

Source code in src/rdf_utils/models/prov.py

def get_git_info(path: Path) -> tuple[str | None, str | None]:
    """Read the revision and repository of a checkout, for a source that is not a package.

    Parameters:
        path: any path inside the worktree; the repository owning it is the one read

    Returns:
        Revision, suffixed `-dirty` when the worktree has changes, and repository
    """
    try:
        repo = Repo(path, search_parent_directories=True)
        # A repository without a commit has no HEAD to read.
        commit = repo.head.commit.hexsha
    except (InvalidGitRepositoryError, NoSuchPathError, ValueError):
        return None, None
    if repo.is_dirty(untracked_files=True):
        commit = f"{commit}-dirty"
    remote = next((url for r in repo.remotes if r.name == "origin" for url in r.urls), None)
    return commit, _repository_iri(remote) if remote else None

get_ni_iri

get_ni_iri(path, sha256=None)

The IRI of a file's content, named by its hash: the same bytes have the same name.

Parameters:

• path (str | Path) –

  the file

• sha256 (str | None, default: None ) –

  hex SHA-256 digest of its bytes, when they are not read from PATH

Returns:

• URIRef –

  An RFC 6920 named-information IRI, ni:///sha-256;<digest>, digest base64url unpadded

Source code in src/rdf_utils/models/prov.py

def get_ni_iri(path: str | Path, sha256: str | None = None) -> URIRef:
    """The IRI of a file's content, named by its hash: the same bytes have the same name.

    Parameters:
        path: the file
        sha256: hex SHA-256 digest of its bytes, when they are not read from PATH

    Returns:
        An RFC 6920 named-information IRI, `ni:///sha-256;<digest>`, digest base64url unpadded
    """
    if sha256:
        digest = bytes.fromhex(sha256)
    else:
        # Streamed: a run's log or bag can be larger than memory.
        with Path(path).open("rb") as file:
            digest = hashlib.file_digest(file, "sha256").digest()
    return URIRef(f"ni:///sha-256;{base64.urlsafe_b64encode(digest).rstrip(b'=').decode()}")

get_pkg_info

get_pkg_info(name)

Read off an installed package what describes it, from its metadata and its source.

Parameters:

• name (str) –

  distribution name of the package, e.g. rdf_utils

Returns:

• tuple[str, str | None, str | None, str | None] –

  Name, version, revision and repository, in the order load_pkg_prov takes them

Source code in src/rdf_utils/models/prov.py

def get_pkg_info(name: str) -> tuple[str, str | None, str | None, str | None]:
    """Read off an installed package what describes it, from its metadata and its source.

    Parameters:
        name: distribution name of the package, e.g. `rdf_utils`

    Returns:
        Name, version, revision and repository, in the order `load_pkg_prov` takes them
    """
    try:
        package = distribution(name)
    except PackageNotFoundError:
        return name, None, None, None

    # PEP 610: pip records where a package was installed from, if not from an index.
    origin = json.loads(package.read_text("direct_url.json") or "{}")

    # dir_info, editable: `pip install -e <path>` or `-e git+<url>`; the checkout is the code.
    if origin.get("dir_info", {}).get("editable"):
        return package.name, package.version, *get_git_info(_path_from_file_url(origin["url"]))

    # vcs_info: installed, not editable, from a VCS URL, `pip install git+<url>[@<ref>]`.
    if "vcs_info" in origin:
        return (
            package.name,
            package.version,
            origin["vcs_info"]["commit_id"],
            _repository_iri(origin["url"]),
        )

    # An index install, or a copy of a directory whose checkout may have moved on since.
    return package.name, package.version, None, None

load_execution_prov

load_execution_prov(graph, activity_id, used, pkg_id, started, ended=None)

Add a prov-ext:Execution of the used entities by a software package.

Parameters:

• graph (Graph) –

  RDF graph to add the execution to

• activity_id (URIRef) –

  URI of the execution

• used (Iterable[URIRef]) –

  URIs of the entities executed, e.g. an executable and its configuration

• pkg_id (URIRef) –

  URI of the software package that executed them

• started (datetime) –

  start time

• ended (datetime | None, default: None ) –

  end time, None while still running

Source code in src/rdf_utils/models/prov.py

def load_execution_prov(
    graph: Graph,
    activity_id: URIRef,
    used: Iterable[URIRef],
    pkg_id: URIRef,
    started: datetime,
    ended: datetime | None = None,
) -> None:
    """Add a `prov-ext:Execution` of the used entities by a software package.

    Parameters:
        graph: RDF graph to add the execution to
        activity_id: URI of the execution
        used: URIs of the entities executed, e.g. an executable and its configuration
        pkg_id: URI of the software package that executed them
        started: start time
        ended: end time, None while still running
    """
    add_activity(
        graph,
        activity_id,
        (URI_PROV_EXT_TYPE_EXECUTION,),
        used,
        pkg_id,
        started,
        ended,
    )

load_pkg_prov

load_pkg_prov(graph, pkg_id, name, version=None, commit=None, repository=None)

Add a software package as a prov:SoftwareAgent described with schema.org terms.

Parameters:

• graph (Graph) –

  RDF graph to add the package to

• pkg_id (URIRef) –

  URI of the package

• name (str) –

  package name, stored as schema:name

• version (str | None, default: None ) –

  package version, stored as schema:softwareVersion

• commit (str | None, default: None ) –

  revision identifier, stored as schema:identifier

• repository (str | None, default: None ) –

  URL of the source repository, stored as schema:codeRepository

Source code in src/rdf_utils/models/prov.py

def load_pkg_prov(
    graph: Graph,
    pkg_id: URIRef,
    name: str,
    version: str | None = None,
    commit: str | None = None,
    repository: str | None = None,
) -> None:
    """Add a software package as a `prov:SoftwareAgent` described with schema.org terms.

    Parameters:
        graph: RDF graph to add the package to
        pkg_id: URI of the package
        name: package name, stored as `schema:name`
        version: package version, stored as `schema:softwareVersion`
        commit: revision identifier, stored as `schema:identifier`
        repository: URL of the source repository, stored as `schema:codeRepository`
    """
    add_agent(graph, pkg_id, (PROV.SoftwareAgent,), name)
    if version is not None:
        graph.add((pkg_id, SDO.softwareVersion, Literal(version)))
    if commit is not None:
        graph.add((pkg_id, SDO.identifier, Literal(commit)))
    if repository is not None:
        graph.add((pkg_id, SDO.codeRepository, URIRef(repository)))

load_sampling_prov

load_sampling_prov(graph, activity_id, used, generated, quantity_id, agent_id, started, ended=None)

Add a prov-ext:Generalization that sampled a quantity into generated entities.

Parameters:

• graph (Graph) –

  RDF graph to add the sampling to

• activity_id (URIRef) –

  URI of the sampling

• used (Iterable[URIRef]) –

  URIs of the entities used besides the quantity, e.g. the model

• generated (Iterable[URIRef]) –

  URIs of the sampled entities; typed as prov:Entity here

• quantity_id (URIRef) –

  URI of the sampled quantity

• agent_id (URIRef) –

  URI of the agent that ran the sampling

• started (datetime) –

  start time

• ended (datetime | None, default: None ) –

  end time, None while still running

Source code in src/rdf_utils/models/prov.py

def load_sampling_prov(
    graph: Graph,
    activity_id: URIRef,
    used: Iterable[URIRef],
    generated: Iterable[URIRef],
    quantity_id: URIRef,
    agent_id: URIRef,
    started: datetime,
    ended: datetime | None = None,
) -> None:
    """Add a `prov-ext:Generalization` that sampled a quantity into generated entities.

    Parameters:
        graph: RDF graph to add the sampling to
        activity_id: URI of the sampling
        used: URIs of the entities used besides the quantity, e.g. the model
        generated: URIs of the sampled entities; typed as `prov:Entity` here
        quantity_id: URI of the sampled quantity
        agent_id: URI of the agent that ran the sampling
        started: start time
        ended: end time, None while still running
    """
    for entity_id in generated:
        add_entity(graph, entity_id)
        graph.add((entity_id, PROV.wasGeneratedBy, activity_id))
    add_activity(
        graph,
        activity_id,
        (URI_PROV_EXT_TYPE_GENERALIZATION,),
        [*used, quantity_id],
        agent_id,
        started,
        ended,
    )

load_transformation_prov

load_transformation_prov(graph, activity_id, sources, targets, pkg_id, started, ended=None)

Add a prov-ext:Transformation of source entities into target entities.

Parameters:

• graph (Graph) –

  RDF graph to add the transformation to

• activity_id (URIRef) –

  URI of the transformation

• sources (Iterable[URIRef]) –

  URIs of the entities transformed

• targets (Iterable[URIRef]) –

  URIs of the entities generated; typed as prov:Entity here

• pkg_id (URIRef) –

  URI of the software package that ran the transformation

• started (datetime) –

  start time

• ended (datetime | None, default: None ) –

  end time, None while still running

Source code in src/rdf_utils/models/prov.py

def load_transformation_prov(
    graph: Graph,
    activity_id: URIRef,
    sources: Iterable[URIRef],
    targets: Iterable[URIRef],
    pkg_id: URIRef,
    started: datetime,
    ended: datetime | None = None,
) -> None:
    """Add a `prov-ext:Transformation` of source entities into target entities.

    Parameters:
        graph: RDF graph to add the transformation to
        activity_id: URI of the transformation
        sources: URIs of the entities transformed
        targets: URIs of the entities generated; typed as `prov:Entity` here
        pkg_id: URI of the software package that ran the transformation
        started: start time
        ended: end time, None while still running
    """
    for entity_id in targets:
        add_entity(graph, entity_id)
        graph.add((entity_id, PROV.wasGeneratedBy, activity_id))
    add_activity(
        graph,
        activity_id,
        (URI_PROV_EXT_TYPE_TRANSFORMATION,),
        sources,
        pkg_id,
        started,
        ended,
    )

resolve_location

resolve_location(graph, node)

The IRI a node is prov:atLocation, or None when it has no location.

An IRI location is returned as is. A prov-ext:RelativeLocation resolves its prov-ext:rel-path against its own location, as an RFC 3986 reference, so a root that is itself relative resolves too. Any other location, e.g. a blank node or literal, raises a ValueError.

Source code in src/rdf_utils/models/prov.py

def resolve_location(graph: Graph, node: URIRef) -> URIRef | None:
    """The IRI a node is `prov:atLocation`, or None when it has no location.

    An IRI location is returned as is. A `prov-ext:RelativeLocation` resolves its
    `prov-ext:rel-path` against its own location, as an RFC 3986 reference, so a root that is
    itself relative resolves too. Any other location, e.g. a blank node or literal, raises a
    ValueError.
    """
    location = graph.value(node, PROV.atLocation)
    if location is None:
        return None
    if (location, RDF.type, URI_PROV_EXT_TYPE_RELATIVE_LOCATION) in graph:
        root = str(resolve_location(graph, location))
        # A root is a directory; RFC 3986 would otherwise drop its last segment.
        root = root if root.endswith("/") else f"{root}/"
        return URIRef(urljoin(root, str(graph.value(location, URI_PROV_EXT_PRED_REL_PATH))))
    if isinstance(location, URIRef):
        return location
    raise ValueError(f"location '{location}' of '{node}' is neither an IRI nor a relative location")