Source code for prov.model.bundle

"""PROV bundles and documents: containers of PROV records."""

from __future__ import annotations  # needed for | type annotations in Python < 3.10

import io
import itertools
import logging
import os
import shutil
import tempfile
import warnings
from collections import defaultdict
from collections.abc import Iterable
from typing import IO, Any, cast
from urllib.parse import urlparse

from prov import serializers
from prov.constants import (
    PROV,
    PROV_ACTIVITY,
    PROV_AGENT,
    PROV_ALTERNATE,
    PROV_ASSOCIATION,
    PROV_ATTR_ACTIVITY,
    PROV_ATTR_AGENT,
    PROV_ATTR_ALTERNATE1,
    PROV_ATTR_ALTERNATE2,
    PROV_ATTR_BUNDLE,
    PROV_ATTR_COLLECTION,
    PROV_ATTR_DELEGATE,
    PROV_ATTR_ENDER,
    PROV_ATTR_ENDTIME,
    PROV_ATTR_ENTITY,
    PROV_ATTR_GENERAL_ENTITY,
    PROV_ATTR_GENERATED_ENTITY,
    PROV_ATTR_GENERATION,
    PROV_ATTR_INFLUENCEE,
    PROV_ATTR_INFLUENCER,
    PROV_ATTR_INFORMANT,
    PROV_ATTR_INFORMED,
    PROV_ATTR_PLAN,
    PROV_ATTR_RESPONSIBLE,
    PROV_ATTR_SPECIFIC_ENTITY,
    PROV_ATTR_STARTER,
    PROV_ATTR_STARTTIME,
    PROV_ATTR_TIME,
    PROV_ATTR_TRIGGER,
    PROV_ATTR_USAGE,
    PROV_ATTR_USED_ENTITY,
    PROV_ATTRIBUTION,
    PROV_COMMUNICATION,
    PROV_DELEGATION,
    PROV_DERIVATION,
    PROV_END,
    PROV_ENTITY,
    PROV_GENERATION,
    PROV_INFLUENCE,
    PROV_INVALIDATION,
    PROV_LABEL,
    PROV_LOCATION,
    PROV_MEMBERSHIP,
    PROV_MENTION,
    PROV_ROLE,
    PROV_SPECIALIZATION,
    PROV_START,
    PROV_TYPE,
    PROV_USAGE,
    PROV_VALUE,
)
from prov.identifier import Namespace, QualifiedName
from prov.model.namespaces import NamespaceManager
from prov.model.records import (
    PROV_REC_CLS,
    ActivityRef,
    AgentRef,
    DatetimeOrStr,
    EntityRef,
    GenrationRef,
    NameValuePair,
    NSCollection,
    OptionalID,
    PathLike,
    ProvActivity,
    ProvAgent,
    ProvAlternate,
    ProvAssociation,
    ProvAttribution,
    ProvCommunication,
    ProvDelegation,
    ProvDerivation,
    ProvEnd,
    ProvEntity,
    ProvException,
    ProvExceptionInvalidQualifiedName,
    ProvInfluence,
    ProvInvalidation,
    ProvMembership,
    ProvMention,
    ProvRecord,
    ProvSpecialization,
    ProvStart,
    ProvUsage,
    QualifiedNameCandidate,
    RecordAttributesArg,
    UsageRef,
    _ensure_datetime,
)

logger = logging.getLogger(__name__)


[docs] class ProvBundle: """PROV Bundle""" def __init__( self, records: Iterable[ProvRecord] | None = None, identifier: QualifiedName | None = None, namespaces: NSCollection | None = None, document: ProvDocument | None = None, ): """Initialise the bundle. Args: records: Optional iterable of records to add to the bundle (default: ``None``). identifier: Optional identifier of the bundle (default: ``None``). namespaces: Optional namespaces to register, as a ``{prefix: uri}`` dict or an iterable of :class:`~prov.identifier.Namespace` (default: ``None``). document: Optional parent document for the bundle (default: ``None``). """ # Initializing bundle-specific attributes self._identifier = identifier self._records = [] # type: list[ProvRecord] self._id_map = defaultdict(list) # type: dict[QualifiedName, list[ProvRecord]] self._document = document self._namespaces = NamespaceManager( namespaces, parent=(document._namespaces if document is not None else None) ) # type: NamespaceManager if records: for record in records: self.add_record(record) def __repr__(self) -> str: return f"<{self.__class__.__name__}: {self._identifier}>" @property def namespaces(self) -> set[Namespace]: """The set of the bundle's registered namespaces.""" return set(self._namespaces.get_registered_namespaces()) @property def default_ns_uri(self) -> str | None: """The bundle's default namespace URI, or ``None`` if none is set.""" default_ns = self._namespaces.get_default_namespace() return default_ns.uri if default_ns else None @property def document(self) -> ProvDocument | None: """The parent document of this bundle, or ``None`` if it has none.""" return self._document @property def identifier(self) -> QualifiedName | None: """The bundle's identifier, or ``None`` if it has none.""" return self._identifier @property def records(self) -> list[ProvRecord]: """A copy of the list of all records in this bundle.""" return list(self._records) # Bundle configurations
[docs] def set_default_namespace(self, uri: str) -> None: """Set the bundle's default namespace to one with the given URI. Args: uri: The URI of the default namespace. """ self._namespaces.set_default_namespace(uri)
[docs] def get_default_namespace(self) -> Namespace | None: """Return the default namespace, or ``None`` if none is set.""" return self._namespaces.get_default_namespace()
[docs] def add_namespace( self, namespace_or_prefix: Namespace | str, uri: str | None = None ) -> Namespace: """Add a namespace to the bundle, unless an equivalent one exists. Args: namespace_or_prefix: A :class:`~prov.identifier.Namespace` to add, or a prefix string (in which case ``uri`` is required). uri: The namespace URI; required when ``namespace_or_prefix`` is a prefix string (default: ``None``). Returns: The registered namespace (which may be an existing or renamed one). Raises: ProvException: If a prefix string is given without a ``uri``. """ if isinstance(namespace_or_prefix, Namespace): return self._namespaces.add_namespace(namespace_or_prefix) else: if uri is not None: return self._namespaces.add_namespace( Namespace(namespace_or_prefix, uri) ) else: raise ProvException("Cannot add a namespace without a URI")
[docs] def get_registered_namespaces(self) -> Iterable[Namespace]: """Return all namespaces registered on the bundle. Returns: An iterable of :class:`~prov.identifier.Namespace`. """ return self._namespaces.get_registered_namespaces()
[docs] def valid_qualified_name( self, identifier: QualifiedNameCandidate ) -> QualifiedName | None: """Resolve an identifier to a qualified name using this bundle. Args: identifier: The candidate to resolve. Returns: The resolved :class:`~prov.identifier.QualifiedName`, or ``None`` if it could not be resolved. """ return self._namespaces.valid_qualified_name(identifier)
[docs] def mandatory_valid_qname( self, identifier: QualifiedNameCandidate ) -> QualifiedName: """Resolve an identifier to a qualified name, requiring success. Args: identifier: The candidate to resolve. Returns: The resolved :class:`~prov.identifier.QualifiedName`. Raises: ProvExceptionInvalidQualifiedName: If the identifier cannot be resolved to a valid qualified name. """ valid_qname = self.valid_qualified_name(identifier) if valid_qname is not None: return valid_qname else: raise ProvExceptionInvalidQualifiedName(identifier)
[docs] def get_records( self, class_or_type_or_tuple: type | tuple[type] | None = None ) -> Iterable[ProvRecord]: """Return the bundle's records, optionally filtered by type. Args: class_or_type_or_tuple: An optional class or tuple of classes; only records passing ``isinstance()`` against it are returned (default: ``None``, meaning all records). Returns: The matching :class:`ProvRecord` objects (a list when unfiltered, otherwise a filter iterator). """ results = list(self._records) # make a (shallow) copy of the record list if class_or_type_or_tuple: return filter(lambda rec: isinstance(rec, class_or_type_or_tuple), results) else: return results
[docs] def get_record(self, identifier: QualifiedNameCandidate) -> list[ProvRecord]: """Return all records matching a given identifier. Args: identifier: The record identifier to look up. Returns: The matching :class:`ProvRecord` objects, or an empty list if the identifier is invalid or unknown. """ valid_id = self.valid_qualified_name(identifier) return list(self._id_map[valid_id]) if valid_id is not None else []
# Miscellaneous functions
[docs] def is_document(self) -> bool: """Return ``True`` if this is a document, ``False`` otherwise.""" return False
[docs] def is_bundle(self) -> bool: """Return ``True`` if this is a (named) bundle, ``False`` otherwise.""" return True
[docs] def has_bundles(self) -> bool: """Return ``True`` if this object contains bundles, ``False`` otherwise.""" return False
@property def bundles(self) -> Iterable[ProvBundle]: """The bundles contained in this object. Raises: ProvException: Always, since a plain bundle cannot contain sub-bundles. Only :class:`ProvDocument` overrides this. """ raise ProvException("A PROV bundle does not contain sub-bundles")
[docs] def get_provn(self, _indent_level: int = 0) -> str: """Return the PROV-N representation of the bundle.""" indentation = "" + (" " * _indent_level) newline = "\n" + (" " * (_indent_level + 1)) # if this is the document, start the document; # otherwise, start the bundle lines = ["document"] if self.is_document() else [f"bundle {self._identifier}"] default_namespace = self._namespaces.get_default_namespace() if default_namespace: lines.append(f"default <{default_namespace.uri}>") registered_namespaces = self._namespaces.get_registered_namespaces() if registered_namespaces: lines.extend( [ f"prefix {namespace.prefix} <{namespace.uri}>" for namespace in registered_namespaces ] ) if default_namespace or registered_namespaces: # a blank line between the prefixes and the assertions lines.append("") # adding all the records lines.extend([record.get_provn() for record in self._records]) if self.is_document(): # Print out bundles lines.extend(bundle.get_provn(_indent_level + 1) for bundle in self.bundles) provn_str = newline.join(lines) + "\n" # closing the structure provn_str += indentation + ( "endDocument" if self.is_document() else "endBundle" ) return provn_str
def __eq__(self, other: Any) -> bool: if not isinstance(other, ProvBundle): return False other_records = set(other.get_records()) this_records = set(self.get_records()) if len(this_records) != len(other_records): return False # check if all records for equality for record_a in this_records: # Manually look for the record found = False for record_b in other_records: if record_a == record_b: other_records.remove(record_b) found = True break if not found: logger.debug( "Equality (ProvBundle): Could not find this record: %s", str(record_a), ) return False return True def __ne__(self, other: Any) -> bool: return not (self == other) __hash__ = None # type: ignore[assignment] # Transformations def _unified_records(self) -> list[ProvRecord]: """Returns a list of unified records.""" # TODO: Check unification rules in the PROV-CONSTRAINTS document # This method simply merges the records having the same name merged_records = {} for _identifier, records in self._id_map.items(): if len(records) > 1: # more than one record having the same identifier # merge the records merged = records[0].copy() for record in records[1:]: merged.add_attributes(record.attributes) # map all of them to the merged record for record in records: merged_records[record] = merged if not merged_records: # No merging done, just return the list of original records return list(self._records) added_merged_records = set() unified_records = [] for record in self._records: if record in merged_records: merged = merged_records[record] if merged not in added_merged_records: unified_records.append(merged) added_merged_records.add(merged) else: # add the original record unified_records.append(record) return unified_records
[docs] def unified(self) -> ProvBundle: """Return a new bundle with records sharing an identifier merged. For each identifier carried by more than one record, a single merged record is produced by unioning the attributes of all records with that identifier onto a copy of the first. Records with a unique identifier, or no identifier, pass through unchanged. This is a simple identifier-keyed attribute union, not the full PROV-CONSTRAINTS unification: no type/attribute conflicts are detected and no inference is performed. The original bundle is left untouched. Returns: The new, unified :class:`ProvBundle`. """ warnings.warn( "prov 3.0 will change unified() to merge records per the W3C " "PROV-CONSTRAINTS rules; records sharing an identifier but having " "conflicting formal attributes will then raise an error instead of " "having their attributes silently unioned. See " "https://github.com/trungdong/prov/blob/master/ROADMAP.md", FutureWarning, stacklevel=2, ) unified_records = self._unified_records() bundle = ProvBundle(records=unified_records, identifier=self.identifier) return bundle
[docs] def update(self, other: ProvBundle) -> None: """Append all records of another bundle into this bundle. Args: other: The :class:`ProvBundle` whose records are appended. Raises: ProvException: If ``other`` is not a :class:`ProvBundle`, or is a document that itself contains sub-bundles. """ if isinstance(other, ProvBundle): if other.is_document() and other.has_bundles(): # Cannot add bundles to a bundle raise ProvException( "ProvBundle.update(): The other bundle is a document with " "sub-bundle(s)." ) for record in other.get_records(): self.add_record(record) else: raise ProvException( "ProvBundle.update(): The other bundle is not a ProvBundle " f"instance ({type(other)})" )
# Provenance statements def _add_record(self, record: ProvRecord) -> None: # IMPORTANT: All records need to be added to a bundle/document via this # method. Otherwise, the _id_map dict will not be correctly updated identifier = record.identifier if identifier is not None: self._id_map[identifier].append(record) self._records.append(record)
[docs] def new_record( self, record_type: QualifiedName, identifier: OptionalID, attributes: RecordAttributesArg | None = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvRecord: """Create a new record, add it to the bundle, and return it. Args: record_type: The record type, one of the keys of :data:`PROV_REC_CLS`. identifier: The identifier for the new record (may be ``None`` for relations). attributes: Formal attributes of the record, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). other_attributes: Additional (non-formal) attributes, in the same forms (default: ``None``). Returns: The newly created and added :class:`ProvRecord`. """ attr_list = [] # type: list[tuple[QualifiedNameCandidate, Any]] if attributes: if isinstance(attributes, dict): attr_list.extend((attr, value) for attr, value in attributes.items()) else: # expecting a list of attributes here attr_list.extend(attributes) if other_attributes: attr_list.extend( other_attributes.items() if isinstance(other_attributes, dict) else other_attributes ) record_identifier = ( self.valid_qualified_name(identifier) if identifier else None ) new_record = PROV_REC_CLS[record_type](self, record_identifier, attr_list) self._add_record(new_record) return new_record
[docs] def add_record(self, record: ProvRecord) -> ProvRecord: """Add a copy of a record to this bundle. The record is re-created within this bundle (resolving its identifier and attributes against this bundle's namespaces), so the returned record is a new object, not ``record`` itself. Args: record: The :class:`ProvRecord` to copy into the bundle. Returns: The newly created :class:`ProvRecord` belonging to this bundle. """ return self.new_record( record.get_type(), record.identifier, record.formal_attributes, record.extra_attributes, )
[docs] def entity( self, identifier: QualifiedNameCandidate, other_attributes: RecordAttributesArg | None = None, ) -> ProvEntity: """Create a new entity and add it to the bundle. Args: identifier: The identifier for the new entity. other_attributes: Optional attributes for the entity, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvEntity`. """ return self.new_record(PROV_ENTITY, identifier, None, other_attributes) # type: ignore
[docs] def activity( self, identifier: QualifiedNameCandidate, startTime: DatetimeOrStr | None = None, endTime: DatetimeOrStr | None = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvActivity: """Create a new activity and add it to the bundle. Args: identifier: The identifier for the new activity. startTime: Optional start time, as a :class:`datetime.datetime` or a string parseable by :func:`dateutil.parser.parse` (default: ``None``). endTime: Optional end time, in the same forms (default: ``None``). other_attributes: Optional attributes for the activity, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvActivity`. """ attributes = { PROV_ATTR_STARTTIME: _ensure_datetime(startTime), PROV_ATTR_ENDTIME: _ensure_datetime(endTime), } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_ACTIVITY, identifier, attributes, other_attributes, ) # type: ignore
[docs] def generation( self, entity: EntityRef, activity: ActivityRef | None = None, time: DatetimeOrStr | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvRecord: """Create a new generation record for an entity. Args: entity: The generated entity (or its string identifier). activity: The activity (or its string identifier) involved in the generation (default: ``None``). time: Optional time of the generation, as a :class:`datetime.datetime` or a string parseable by :func:`dateutil.parser.parse` (default: ``None``). identifier: Optional identifier for the generation record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new generation record. """ attributes = { PROV_ATTR_ENTITY: entity, PROV_ATTR_ACTIVITY: activity, PROV_ATTR_TIME: _ensure_datetime(time), } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_GENERATION, identifier, attributes, other_attributes, )
[docs] def usage( self, activity: ActivityRef, entity: EntityRef | None = None, time: DatetimeOrStr | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvUsage: """Create a new usage record for an activity. Args: activity: The using activity (or its string identifier). entity: The entity (or its string identifier) involved in the usage relationship (default: ``None``). time: Optional time of the usage, as a :class:`datetime.datetime` or a string parseable by :func:`dateutil.parser.parse` (default: ``None``). identifier: Optional identifier for the usage record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvUsage` record. """ attributes = { PROV_ATTR_ACTIVITY: activity, PROV_ATTR_ENTITY: entity, PROV_ATTR_TIME: _ensure_datetime(time), } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_USAGE, identifier, attributes, other_attributes, ) # type: ignore
[docs] def start( self, activity: ActivityRef, trigger: EntityRef | None = None, starter: ActivityRef | None = None, time: DatetimeOrStr | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvStart: """Create a new start record for an activity. Args: activity: The started activity (or its string identifier). trigger: The entity (or its string identifier) triggering the start (default: ``None``). starter: Optional activity qualifying the start, through which the trigger entity is generated (default: ``None``). time: Optional time of the start, as a :class:`datetime.datetime` or a string parseable by :func:`dateutil.parser.parse` (default: ``None``). identifier: Optional identifier for the start record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvStart` record. """ attributes = { PROV_ATTR_ACTIVITY: activity, PROV_ATTR_TRIGGER: trigger, PROV_ATTR_STARTER: starter, PROV_ATTR_TIME: _ensure_datetime(time), } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_START, identifier, attributes, other_attributes, ) # type: ignore
[docs] def end( self, activity: ActivityRef, trigger: EntityRef | None = None, ender: ActivityRef | None = None, time: DatetimeOrStr | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvEnd: """Create a new end record for an activity. Args: activity: The ended activity (or its string identifier). trigger: The entity (or its string identifier) triggering the end (default: ``None``). ender: Optional activity qualifying the end, through which the trigger entity is generated (default: ``None``). time: Optional time of the end, as a :class:`datetime.datetime` or a string parseable by :func:`dateutil.parser.parse` (default: ``None``). identifier: Optional identifier for the end record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvEnd` record. """ attributes = { PROV_ATTR_ACTIVITY: activity, PROV_ATTR_TRIGGER: trigger, PROV_ATTR_ENDER: ender, PROV_ATTR_TIME: _ensure_datetime(time), } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_END, identifier, attributes, other_attributes, ) # type: ignore
[docs] def invalidation( self, entity: EntityRef, activity: ActivityRef | None = None, time: DatetimeOrStr | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvInvalidation: """Create a new invalidation record for an entity. Args: entity: The invalidated entity (or its string identifier). activity: The activity (or its string identifier) involved in the invalidation (default: ``None``). time: Optional time of the invalidation, as a :class:`datetime.datetime` or a string parseable by :func:`dateutil.parser.parse` (default: ``None``). identifier: Optional identifier for the invalidation record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvInvalidation` record. """ attributes = { PROV_ATTR_ENTITY: entity, PROV_ATTR_ACTIVITY: activity, PROV_ATTR_TIME: _ensure_datetime(time), } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_INVALIDATION, identifier, attributes, other_attributes, ) # type: ignore
[docs] def communication( self, informed: ActivityRef, informant: ActivityRef, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvCommunication: """Create a new communication record between two activities. Args: informed: The informed activity (relationship destination). informant: The informing activity (relationship source). identifier: Optional identifier for the communication record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvCommunication` record. """ attributes = { PROV_ATTR_INFORMED: informed, PROV_ATTR_INFORMANT: informant, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_COMMUNICATION, identifier, attributes, other_attributes, ) # type: ignore
[docs] def agent( self, identifier: QualifiedNameCandidate, other_attributes: RecordAttributesArg | None = None, ) -> ProvAgent: """Create a new agent and add it to the bundle. Args: identifier: The identifier for the new agent. other_attributes: Optional attributes for the agent, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvAgent`. """ return self.new_record(PROV_AGENT, identifier, None, other_attributes) # type: ignore
[docs] def attribution( self, entity: EntityRef, agent: AgentRef, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvAttribution: """Create a new attribution record between an entity and an agent. Args: entity: The entity (or its string identifier) being attributed (relationship source). agent: The agent (or its string identifier) the entity is attributed to (relationship destination). identifier: Optional identifier for the attribution record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvAttribution` record. """ attributes = { PROV_ATTR_ENTITY: entity, PROV_ATTR_AGENT: agent, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_ATTRIBUTION, identifier, attributes, other_attributes, ) # type: ignore
[docs] def association( self, activity: ActivityRef, agent: AgentRef | None = None, plan: EntityRef | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvAssociation: """Create a new association record for an activity. Args: activity: The activity (or its string identifier). agent: The agent (or its string identifier) associated with the activity (default: ``None``). plan: Optional entity qualifying the association through an internal plan (default: ``None``). identifier: Optional identifier for the association record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvAssociation` record. """ attributes = { PROV_ATTR_ACTIVITY: activity, PROV_ATTR_AGENT: agent, PROV_ATTR_PLAN: plan, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_ASSOCIATION, identifier, attributes, other_attributes, ) # type: ignore
[docs] def delegation( self, delegate: AgentRef, responsible: AgentRef, activity: ActivityRef | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvDelegation: """Create a new delegation record between two agents. Args: delegate: The agent (or its string identifier) delegating the responsibility (relationship source). responsible: The agent (or its string identifier) the responsibility is delegated to (relationship destination). activity: Optional activity qualifying the delegation (default: ``None``). identifier: Optional identifier for the delegation record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvDelegation` record. """ attributes = { PROV_ATTR_DELEGATE: delegate, PROV_ATTR_RESPONSIBLE: responsible, PROV_ATTR_ACTIVITY: activity, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_DELEGATION, identifier, attributes, other_attributes, ) # type: ignore
[docs] def influence( self, influencee: EntityRef | ActivityRef | AgentRef, influencer: EntityRef | ActivityRef | AgentRef, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvInfluence: """Create a new influence record between two entities, activities or agents. Args: influencee: The influenced entity, activity or agent (relationship source). influencer: The influencing entity, activity or agent (relationship destination). identifier: Optional identifier for the influence record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvInfluence` record. """ attributes = { PROV_ATTR_INFLUENCEE: influencee, PROV_ATTR_INFLUENCER: influencer, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_INFLUENCE, identifier, attributes, other_attributes, ) # type: ignore
[docs] def derivation( self, generatedEntity: EntityRef, usedEntity: EntityRef, activity: ActivityRef | None = None, generation: GenrationRef | None = None, usage: UsageRef | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvDerivation: """Create a new derivation record for a generated entity from a used entity. Args: generatedEntity: The generated entity (or its string identifier), the relationship source. usedEntity: The used entity (or its string identifier), the relationship destination. activity: The activity (or its string identifier) involved in the derivation (default: ``None``). generation: Optional generation record qualifying the derivation through the activity's generation of the generated entity (default: ``None``). usage: Optional usage record qualifying the derivation through the activity's use of the used entity (default: ``None``). identifier: Optional identifier for the derivation record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvDerivation` record. """ attributes = { PROV_ATTR_GENERATED_ENTITY: generatedEntity, PROV_ATTR_USED_ENTITY: usedEntity, PROV_ATTR_ACTIVITY: activity, PROV_ATTR_GENERATION: generation, PROV_ATTR_USAGE: usage, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_DERIVATION, identifier, attributes, other_attributes ) # type: ignore
[docs] def revision( self, generatedEntity: EntityRef, usedEntity: EntityRef, activity: ActivityRef | None = None, generation: GenrationRef | None = None, usage: UsageRef | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvDerivation: """Create a new revision record for a generated entity from a used entity. A revision is a derivation with an additional ``prov:Revision`` type. Args: generatedEntity: The generated (revised) entity (or its string identifier), the relationship source. usedEntity: The used (original) entity (or its string identifier), the relationship destination. activity: The activity (or its string identifier) involved in the revision (default: ``None``). generation: Optional generation record qualifying the derivation (default: ``None``). usage: Optional usage record qualifying the derivation (default: ``None``). identifier: Optional identifier for the revision record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvDerivation` record, typed as a revision. """ record = self.derivation( generatedEntity, usedEntity, activity, generation, usage, identifier, other_attributes, ) record.add_asserted_type(PROV["Revision"]) return record
[docs] def quotation( self, generatedEntity: EntityRef, usedEntity: EntityRef, activity: ActivityRef | None = None, generation: GenrationRef | None = None, usage: UsageRef | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvDerivation: """Create a new quotation record for a generated entity from a used entity. A quotation is a derivation with an additional ``prov:Quotation`` type. Args: generatedEntity: The quoting entity (or its string identifier), the relationship source. usedEntity: The quoted entity (or its string identifier), the relationship destination. activity: The activity (or its string identifier) involved in the quotation (default: ``None``). generation: Optional generation record qualifying the derivation (default: ``None``). usage: Optional usage record qualifying the derivation (default: ``None``). identifier: Optional identifier for the quotation record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvDerivation` record, typed as a quotation. """ record = self.derivation( generatedEntity, usedEntity, activity, generation, usage, identifier, other_attributes, ) record.add_asserted_type(PROV["Quotation"]) return record
[docs] def primary_source( self, generatedEntity: EntityRef, usedEntity: EntityRef, activity: ActivityRef | None = None, generation: GenrationRef | None = None, usage: UsageRef | None = None, identifier: OptionalID = None, other_attributes: RecordAttributesArg | None = None, ) -> ProvDerivation: """Create a new primary-source record for a generated entity from a used entity. A primary source is a derivation with an additional ``prov:PrimarySource`` type. Args: generatedEntity: The derived entity (or its string identifier), the relationship source. usedEntity: The primary-source entity (or its string identifier), the relationship destination. activity: The activity (or its string identifier) involved (default: ``None``). generation: Optional generation record qualifying the derivation (default: ``None``). usage: Optional usage record qualifying the derivation (default: ``None``). identifier: Optional identifier for the primary-source record (default: ``None``). other_attributes: Optional extra attributes, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvDerivation` record, typed as a primary source. """ record = self.derivation( generatedEntity, usedEntity, activity, generation, usage, identifier, other_attributes, ) record.add_asserted_type(PROV["PrimarySource"]) return record
[docs] def specialization( self, specificEntity: EntityRef, generalEntity: EntityRef ) -> ProvSpecialization: """Create a new specialisation record from a general entity. Args: specificEntity: The specific entity (or its string identifier), the relationship source. generalEntity: The general entity (or its string identifier), the relationship destination. Returns: The new :class:`ProvSpecialization` record. """ attributes = { PROV_ATTR_SPECIFIC_ENTITY: specificEntity, PROV_ATTR_GENERAL_ENTITY: generalEntity, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_SPECIALIZATION, None, attributes, ) # type: ignore
[docs] def alternate(self, alternate1: EntityRef, alternate2: EntityRef) -> ProvAlternate: """Create a new alternate record between two entities. Args: alternate1: The first entity (or its string identifier), the relationship source. alternate2: The second entity (or its string identifier), the relationship destination. Returns: The new :class:`ProvAlternate` record. """ attributes = { PROV_ATTR_ALTERNATE1: alternate1, PROV_ATTR_ALTERNATE2: alternate2, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_ALTERNATE, None, attributes, ) # type: ignore
[docs] def mention( self, specificEntity: EntityRef, generalEntity: EntityRef, bundle: EntityRef ) -> ProvMention: """Create a new mention record from a general entity in a bundle. Args: specificEntity: The specific entity (or its string identifier), the relationship source. generalEntity: The general entity (or its string identifier), the relationship destination. bundle: The bundle (or its string identifier) that the general entity is described in. Returns: The new :class:`ProvMention` record. """ attributes = { PROV_ATTR_SPECIFIC_ENTITY: specificEntity, PROV_ATTR_GENERAL_ENTITY: generalEntity, PROV_ATTR_BUNDLE: bundle, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_MENTION, None, attributes, ) # type: ignore
[docs] def collection( self, identifier: QualifiedNameCandidate, other_attributes: RecordAttributesArg | None = None, ) -> ProvEntity: """Create a new collection entity and add it to the bundle. A collection is an entity with an additional ``prov:Collection`` type. Args: identifier: The identifier for the new collection. other_attributes: Optional attributes for the collection, as a dict or an iterable of ``(name, value)`` pairs (default: ``None``). Returns: The new :class:`ProvEntity`, typed as a collection. """ record = self.new_record(PROV_ENTITY, identifier, None, other_attributes) record.add_asserted_type(PROV["Collection"]) return record # type: ignore
[docs] def membership(self, collection: EntityRef, entity: EntityRef) -> ProvMembership: """Create a new membership record adding an entity to a collection. Args: collection: The collection (or its string identifier) the entity is added to. entity: The entity (or its string identifier) added to the collection. Returns: The new :class:`ProvMembership` record. """ attributes = { PROV_ATTR_COLLECTION: collection, PROV_ATTR_ENTITY: entity, } # type: dict[QualifiedNameCandidate, Any] return self.new_record( PROV_MEMBERSHIP, None, attributes, ) # type: ignore
[docs] def plot( self, filename: PathLike | None = None, show_nary: bool = True, use_labels: bool = False, show_element_attributes: bool = True, show_relation_attributes: bool = True, ) -> None: """Plot the bundle as a graph, saving to a file or displaying it. Args: filename: The path to save the plot to; the image format is derived from its extension. If not given, the plot is shown in an interactive matplotlib window (default: ``None``). show_nary: Whether to show all elements in n-ary relations (default: ``True``). use_labels: Whether to label elements by their ``prov:label`` property instead of their identifier (default: ``False``). show_element_attributes: Whether to show element attributes (default: ``True``). show_relation_attributes: Whether to show relation attributes (default: ``True``). Raises: ValueError: If the format implied by ``filename`` cannot be saved. ImportError: If no ``filename`` is given but matplotlib is not installed. """ # Lazy imports to have soft dependencies on pydot and matplotlib # (imported even later). from prov import dot if filename: format = str(os.path.splitext(filename)[-1]).lower().strip(os.path.extsep) else: format = "png" format = format.lower() d = dot.prov_to_dot( self, show_nary=show_nary, use_labels=use_labels, show_element_attributes=show_element_attributes, show_relation_attributes=show_relation_attributes, ) method = f"create_{format}" if not hasattr(d, method): raise ValueError(f"Format '{format}' cannot be saved.") with io.BytesIO() as buf: buf.write(getattr(d, method)()) buf.seek(0, 0) if filename: with open(filename, "wb") as fh: fh.write(buf.read()) else: # Use matplotlib to show the image as it likely is more # widespread than PIL and works nicely in the ipython notebook. try: import matplotlib.image as mpimg # type: ignore import matplotlib.pylab as plt # type: ignore except ImportError as e: raise ImportError( "The plot() method requires matplotlib when no filename" ' is provided. Install it with: pip install "prov[plot]"' ) from e max_size = 30 img = mpimg.imread(buf) # pydot makes a border around the image. remove it. img = img[1:-1, 1:-1] size = (img.shape[1] / 100.0, img.shape[0] / 100.0) scale = max_size / max(size) if max(size) > max_size else 1.0 size = (scale * size[0], scale * size[1]) plt.figure(figsize=size) plt.subplots_adjust(bottom=0, top=1, left=0, right=1) plt.xticks([]) plt.yticks([]) plt.imshow(img) plt.axis("off") plt.show()
# Aliases wasGeneratedBy = generation used = usage wasStartedBy = start wasEndedBy = end wasInvalidatedBy = invalidation wasInformedBy = communication wasAttributedTo = attribution wasAssociatedWith = association actedOnBehalfOf = delegation wasInfluencedBy = influence wasDerivedFrom = derivation wasRevisionOf = revision wasQuotedFrom = quotation hadPrimarySource = primary_source alternateOf = alternate specializationOf = specialization mentionOf = mention hadMember = membership
[docs] class ProvDocument(ProvBundle): """Provenance Document.""" def __init__( self, records: Iterable[ProvRecord] | None = None, namespaces: NSCollection | None = None, ): """Initialise the document. Args: records: Optional iterable of records to add to the document (default: ``None``). namespaces: Optional namespaces to register, as a ``{prefix: uri}`` dict or an iterable of :class:`~prov.identifier.Namespace` (default: ``None``). """ ProvBundle.__init__( self, records=records, identifier=None, namespaces=namespaces ) self._bundles = {} # type: dict[QualifiedName, ProvBundle] def __repr__(self) -> str: return "<ProvDocument>" def __eq__(self, other: Any) -> bool: if not isinstance(other, ProvDocument): return False # Comparing the documents' content if not super().__eq__(other): return False # Comparing the documents' bundles for b_id, bundle in self._bundles.items(): if b_id not in other._bundles: return False other_bundle = other._bundles[b_id] if bundle != other_bundle: return False # Everything is the same return True
[docs] def is_document(self) -> bool: """Return ``True`` if this is a document, ``False`` otherwise.""" return True
[docs] def is_bundle(self) -> bool: """Return ``True`` if this is a (named) bundle, ``False`` otherwise.""" return False
[docs] def has_bundles(self) -> bool: """Return ``True`` if the document contains bundles, ``False`` otherwise.""" return len(self._bundles) > 0
@property def bundles(self) -> Iterable[ProvBundle]: """The bundles contained in this document.""" return self._bundles.values() # Transformations
[docs] def flattened(self) -> ProvDocument: """Return a new document with all bundle records lifted to the top level. Every record from every bundle is moved up alongside the document's own records and the bundle structure is discarded; this is purely structural (nothing is merged or deduplicated). If the document has no bundles, it is returned unchanged. The original document is left untouched. Returns: The (new) flattened :class:`ProvDocument`, or this document itself if it has no bundles. """ if self._bundles: # Creating a new document for all the records new_doc = ProvDocument() bundled_records = itertools.chain( *[b.get_records() for b in self._bundles.values()] ) for record in itertools.chain(self._records, bundled_records): new_doc.add_record(record) return new_doc else: # returning the same document return self
[docs] def unified(self) -> ProvDocument: """Return a new document with records sharing an identifier merged. The identifier-keyed attribute union (see :meth:`ProvBundle.unified`) is applied to the document's top-level records and, recursively, to each contained bundle, preserving the bundle structure. The original document is left untouched. Returns: The new, unified :class:`ProvDocument`. """ warnings.warn( "prov 3.0 will change unified() to merge records per the W3C " "PROV-CONSTRAINTS rules; records sharing an identifier but having " "conflicting formal attributes will then raise an error instead of " "having their attributes silently unioned. See " "https://github.com/trungdong/prov/blob/master/ROADMAP.md", FutureWarning, stacklevel=2, ) document = ProvDocument(self._unified_records()) document._namespaces = self._namespaces for bundle in self.bundles: unified_bundle = bundle.unified() document.add_bundle(unified_bundle) return document
[docs] def update(self, other: ProvBundle) -> None: """Append all records of another document or bundle into this document. Any bundles of ``other`` are also merged in: a bundle whose identifier already exists in this document is updated in place, otherwise a new bundle is created. Args: other: The :class:`ProvDocument` or :class:`ProvBundle` whose records are appended. Raises: ProvException: If ``other`` is not a :class:`ProvBundle` (or subclass). """ if isinstance(other, ProvBundle): for record in other.get_records(): self.add_record(record) if other.has_bundles(): for bundle in other.bundles: bundle_id = bundle.identifier assert bundle_id is not None if bundle.identifier in self._bundles: self._bundles[bundle.identifier].update(bundle) else: new_bundle = self.bundle(bundle_id) new_bundle.update(bundle) else: raise ProvException( "ProvDocument.update(): The other is not a ProvDocument or " f"ProvBundle instance ({type(other)})" )
# Bundle operations
[docs] def add_bundle( self, bundle: ProvBundle, identifier: QualifiedName | None = None ) -> None: """Add a bundle to this document. If a document (with no nested bundles) is passed, its records are copied into a fresh bundle. The bundle's identifier is normalised against this document's namespaces. Args: bundle: The :class:`ProvBundle` to add. identifier: Optional identifier to use for the bundle; if not given, the bundle's own identifier is used (default: ``None``). Raises: ProvException: If ``bundle`` is not a :class:`ProvBundle`, is a document with nested bundles, has no usable identifier, or an identifier collides with an existing bundle. """ if not isinstance(bundle, ProvBundle): raise ProvException( "Only a ProvBundle instance can be added as a bundle in a ProvDocument." ) if bundle.is_document(): if bundle.has_bundles(): raise ProvException( "Cannot add a document with nested bundles as a bundle." ) # Make it a new ProvBundle new_bundle = ProvBundle(namespaces=bundle.namespaces) new_bundle.update(bundle) bundle = new_bundle if identifier is None: identifier = bundle.identifier if not identifier: raise ProvException("The provided bundle has no identifier") # Link the bundle namespace manager to the document's bundle._namespaces.parent = self._namespaces valid_id = bundle.mandatory_valid_qname(identifier) # IMPORTANT: Rewriting the bundle identifier for consistency bundle._identifier = valid_id if valid_id in self._bundles: raise ProvException("A bundle with that identifier already exists") self._bundles[valid_id] = bundle bundle._document = self
[docs] def bundle(self, identifier: QualifiedNameCandidate) -> ProvBundle: """Create a new, empty named bundle in this document. Args: identifier: The identifier to use for the bundle. Returns: The newly created :class:`ProvBundle`. Raises: ProvException: If ``identifier`` is ``None`` or invalid, or a bundle with that identifier already exists. """ if identifier is None: raise ProvException( "An identifier is required. Cannot create an unnamed bundle." ) valid_id = self.valid_qualified_name(identifier) if valid_id is None: raise ProvException(f'The provided identifier "{identifier}" is not valid') if valid_id in self._bundles: raise ProvException("A bundle with that identifier already exists") b = ProvBundle(identifier=valid_id, document=self) self._bundles[valid_id] = b return b
# Serializing and deserializing
[docs] def serialize( self, destination: io.IOBase | IO[Any] | PathLike | None = None, format: str = "json", **args: Any, ) -> str | None: """Serialize this document to a destination or return it as a string. The available serialization formats are ``"json"``, ``"rdf"``, ``"xml"`` and ``"provn"`` (see :func:`prov.serializers.get`). Args: destination: A writable stream (any object with a ``write`` method) or a local file path to write to. If ``None``, the serialization is returned as a string (default: ``None``). A non-local (network) location is not written and yields ``None``. format: The serialization format (default: ``"json"``, i.e. PROV-JSON). **args: Extra keyword arguments passed to the underlying serializer. Returns: The serialization as a string if no ``destination`` was given, otherwise ``None``. """ serializer = serializers.get(format)(self) if destination is None: buffer = io.StringIO() serializer.serialize(buffer, **args) return buffer.getvalue() # Duck-type on write() rather than isinstance(..., io.IOBase): common # file-like wrappers such as tempfile.NamedTemporaryFile's # _TemporaryFileWrapper proxy a stream's write() but are not # themselves io.IOBase instances (#240). if hasattr(destination, "write"): stream = cast(io.IOBase, destination) serializer.serialize(stream, **args) else: location = str(destination) _scheme, netloc, path, _params, _query, _fragment = urlparse(location) if netloc != "": print( "WARNING: not saving as location " + "is not a local file reference" ) return None fd, name = tempfile.mkstemp() stream = os.fdopen(fd, "wb") serializer.serialize(stream, **args) stream.close() if hasattr(shutil, "move"): shutil.move(name, path) else: shutil.copy(name, path) os.remove(name) return None
[docs] @staticmethod def deserialize( source: io.IOBase | IO[Any] | PathLike | None = None, content: str | bytes | None = None, format: str = "json", **args: Any, ) -> ProvDocument: """Deserialize a document from a source stream/file or a string. Exactly one of ``source`` or ``content`` should be given; ``content`` takes precedence if both are. Note that not all formats support deserialization (PROV-N is write-only). Args: source: A readable stream (any object with a ``read`` method) or a file path to read from (default: ``None``). content: The document as a ``str`` or ``bytes`` to read from (default: ``None``). format: The serialization format (default: ``"json"``, i.e. PROV-JSON). **args: Extra keyword arguments passed to the underlying deserializer. Returns: The deserialized :class:`ProvDocument`. Raises: TypeError: If neither ``source`` nor ``content`` is provided. """ serializer = serializers.get(format)() if content is not None: # io.StringIO only accepts unicode strings stream = io.StringIO( content if isinstance(content, str) else content.decode() ) return serializer.deserialize(stream, **args) if source is not None: # Duck-type on read() rather than isinstance(..., io.IOBase): see # the matching comment in serialize() (#240). if hasattr(source, "read"): return serializer.deserialize(cast(io.IOBase, source), **args) else: with open(source) as f: return serializer.deserialize(f, **args) raise TypeError("Either source or content must be provided")
def sorted_attributes( element: QualifiedName, attributes: Iterable[NameValuePair] ) -> list[NameValuePair]: """Sort attributes into the order required by PROV-XML. Args: element: The PROV record type used to derive the formal-attribute order. attributes: The ``(name, value)`` attribute pairs to sort. Returns: The attributes ordered by formal-attribute position, then the universal label/location/role/type/value attributes, then any remaining attributes alphabetically. """ attributes = list(attributes) order = list(PROV_REC_CLS[element].FORMAL_ATTRIBUTES) # Append label, location, role, type, and value attributes. This is # universal amongst all elements. order.extend([PROV_LABEL, PROV_LOCATION, PROV_ROLE, PROV_TYPE, PROV_VALUE]) # Sort function. The PROV XML specification talks about alphabetical # sorting. We now interpret it as sorting by tag including the prefix # first and then sorting by the text, also including the namespace # prefix if given. def sort_fct(x: NameValuePair) -> tuple[str, str]: return str(x[0]), str(x[1].value if hasattr(x[1], "value") else x[1]) sorted_elements = [] for item in order: this_type_list = [] for e in list(attributes): if e[0] != item: continue this_type_list.append(e) attributes.remove(e) this_type_list.sort(key=sort_fct) sorted_elements.extend(this_type_list) # Add remaining attributes. According to the spec, the other attributes # have a fixed alphabetical order. attributes.sort(key=sort_fct) sorted_elements.extend(attributes) return sorted_elements