"""PROV-DM records: elements, relations, literals, and datatype helpers."""
from __future__ import annotations # defer eval: TYPE_CHECKING names in signatures
import datetime
import decimal
import io
import logging
import os
import re
import typing
from collections import defaultdict
from collections.abc import Callable, Iterable, Iterator, MutableSet
from typing import IO, TYPE_CHECKING, Any, Union, cast
from prov import Error
from prov.constants import (
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_ATTRIBUTE_LITERALS,
PROV_ATTRIBUTE_QNAMES,
PROV_ATTRIBUTES,
PROV_ATTRIBUTION,
PROV_COMMUNICATION,
PROV_DELEGATION,
PROV_DERIVATION,
PROV_END,
PROV_ENTITY,
PROV_GENERATION,
PROV_INFLUENCE,
PROV_INTERNATIONALIZEDSTRING,
PROV_INVALIDATION,
PROV_LABEL,
PROV_MEMBERSHIP,
PROV_MENTION,
PROV_N_MAP,
PROV_QUALIFIEDNAME,
PROV_SPECIALIZATION,
PROV_START,
PROV_TYPE,
PROV_USAGE,
PROV_VALUE,
XSD_ANYURI,
XSD_BOOLEAN,
XSD_DATETIME,
XSD_DECIMAL,
XSD_DOUBLE,
XSD_INT,
XSD_INTEGER,
XSD_LONG,
XSD_STRING,
)
from prov.identifier import Identifier, Namespace, QualifiedName
if TYPE_CHECKING:
from prov.model.bundle import ProvBundle
logger = logging.getLogger(__name__)
# Type aliases for convenience
QualifiedNameCandidate: typing.TypeAlias = QualifiedName | str | Identifier
OptionalID: typing.TypeAlias = QualifiedNameCandidate | None
EntityRef: typing.TypeAlias = Union["ProvEntity", QualifiedNameCandidate]
ActivityRef: typing.TypeAlias = Union["ProvActivity", QualifiedNameCandidate]
AgentRef: typing.TypeAlias = Union[
"ProvAgent", "ProvEntity", "ProvActivity", QualifiedNameCandidate
]
InfluencerRef: typing.TypeAlias = EntityRef | ActivityRef | AgentRef
GenerationRef: typing.TypeAlias = Union["ProvGeneration", QualifiedNameCandidate]
UsageRef: typing.TypeAlias = Union["ProvUsage", QualifiedNameCandidate]
NameValuePair: typing.TypeAlias = tuple[QualifiedName, Any]
AttributePair: typing.TypeAlias = tuple[QualifiedNameCandidate, Any]
RecordAttributesArg: typing.TypeAlias = (
dict[QualifiedNameCandidate, Any] | Iterable[AttributePair]
)
DatetimeOrStr: typing.TypeAlias = datetime.datetime | str
NSCollection: typing.TypeAlias = dict[str, str] | Iterable[Namespace]
PathLike: typing.TypeAlias = str | bytes | os.PathLike[str]
StreamOrPath: typing.TypeAlias = io.IOBase | IO[Any] | PathLike
# Data Types
_XSD_HOUR24_RE = re.compile(r"T24:00:00(\.0+)?(?=$|[Z+-])")
_XSD_ZULU_RE = re.compile(r"Z$")
_XSD_FRACTION_RE = re.compile(r"\.(\d+)")
def _ensure_datetime(value: DatetimeOrStr | None) -> datetime.datetime | None:
"""Coerce a value to a :class:`datetime.datetime`.
A string is parsed with :func:`parse_xsd_datetime`; a
:class:`~datetime.datetime` or ``None`` is returned unchanged.
Raises:
ProvException: If a string value is not a valid ``xsd:dateTime``.
"""
if isinstance(value, str):
parsed = parse_xsd_datetime(value)
if parsed is None:
raise ProvException(f"Invalid xsd:dateTime value: {value!r}")
return parsed
return value
def parse_xsd_datetime(value: str) -> datetime.datetime | None:
"""Parse an ``xsd:dateTime`` string into a :class:`datetime.datetime`.
Accepts the ``xsd:dateTime`` lexical space: ISO 8601 date-time with
optional fractional seconds and timezone, ``Z`` for UTC, and the
hour-24 end-of-day form (which maps to 00:00:00 of the following day,
per the XSD value space).
Args:
value: The date/time string to parse.
Returns:
The parsed :class:`~datetime.datetime`, or ``None`` if ``value``
could not be parsed.
"""
text = value.strip()
if "T" not in text:
# xsd:dateTime requires a literal "T" date/time separator; bare
# xsd:date strings (e.g. "2011-11-16") are a distinct, narrower
# datatype and are intentionally not accepted here (3.0 narrowing).
return None
end_of_day = _XSD_HOUR24_RE.search(text) is not None
if end_of_day:
text = _XSD_HOUR24_RE.sub("T00:00:00", text)
# datetime.fromisoformat on Python 3.10 accepts neither the "Z" suffix
# nor fractional seconds that are not exactly 3 or 6 digits long;
# normalize both before parsing.
text = _XSD_ZULU_RE.sub("+00:00", text)
text = _XSD_FRACTION_RE.sub(
lambda m: "." + m.group(1)[:6].ljust(6, "0"), text, count=1
)
try:
parsed = datetime.datetime.fromisoformat(text)
if end_of_day:
parsed += datetime.timedelta(days=1)
except (ValueError, OverflowError):
return None
return parsed
def parse_boolean(value: str) -> bool | None:
"""Parse an ``xsd:boolean`` string into a Python :class:`bool`.
Args:
value: The string to interpret; ``"false"``/``"0"`` map to ``False``
and ``"true"``/``"1"`` map to ``True`` (case-insensitively).
Returns:
The parsed boolean, or ``None`` if ``value`` is not a recognised
boolean literal.
"""
if value.lower() in ("false", "0"):
return False
elif value.lower() in ("true", "1"):
return True
else:
return None
DATATYPE_PARSERS = {
datetime.datetime: parse_xsd_datetime,
}
# Mappings for XSD datatypes to Python standard types
SupportedXSDParsedTypes: typing.TypeAlias = (
str | datetime.datetime | float | int | bool | Identifier | None
)
XSD_DATATYPE_PARSERS: dict[QualifiedName, Callable[[str], SupportedXSDParsedTypes]] = {
XSD_STRING: str,
XSD_DOUBLE: float,
XSD_LONG: int,
XSD_INT: int,
XSD_INTEGER: int,
XSD_BOOLEAN: parse_boolean,
XSD_DATETIME: parse_xsd_datetime,
XSD_ANYURI: Identifier,
}
_INT32_MAX = 2**31 - 1
_INT64_MAX = 2**63 - 1
def canonical_xsd_datatype(value: object) -> QualifiedName | None:
"""Return the XSD datatype `prov` asserts for a plain Python value.
This is the single source of truth for the serializers' reverse maps:
a typed ``Literal`` may be collapsed to a plain Python value only when
the value's canonical datatype equals the asserted one (a lossless
collapse), and serializers emit exactly this datatype for plain values.
Returns ``None`` for values that serialize natively (str, bool,
datetime — each format handles those itself).
"""
if isinstance(value, bool): # before int: bool is an int subtype
return None
if isinstance(value, int):
if -_INT32_MAX - 1 <= value <= _INT32_MAX:
return XSD_INT
if -_INT64_MAX - 1 <= value <= _INT64_MAX:
return XSD_LONG
return XSD_INTEGER
if isinstance(value, float):
return XSD_DOUBLE
return None
def parse_xsd_types(value: str, datatype: QualifiedName) -> SupportedXSDParsedTypes:
"""Parse a string into a Python value according to its XSD datatype.
Args:
value: The lexical string value to parse.
datatype: The qualified name of the XSD datatype (e.g. ``xsd:int``).
Returns:
The parsed value in the corresponding Python type, or ``None`` if
``datatype`` has no registered parser in :data:`XSD_DATATYPE_PARSERS`.
"""
return (
XSD_DATATYPE_PARSERS[datatype](value)
if datatype in XSD_DATATYPE_PARSERS
else None
)
def first(a_set: Iterable[Any]) -> Any | None:
"""Return the first element of an iterable, or ``None`` if it is empty.
Every in-package caller passes a per-attribute :class:`TypedValueSet`
(dict-backed, insertion-ordered), so this deterministically returns the
first-inserted value -- e.g. for ``args``/``formal_attributes``,
:meth:`ProvActivity.get_startTime`/``get_endTime``, the single-value
cardinality guard in :meth:`ProvRecord.add_attributes`, and the
PROV-JSON encoder's single-value case. This is a change from 2.x, where
the equivalent plain-``set`` storage made "first" an arbitrary
hash-bucket-order element rather than an insertion-order one (#34).
"""
return next(iter(a_set), None)
class TypedValueSet(MutableSet[Any]):
"""A set-like container that deduplicates by ``(type(value), value)``.
Backs each attribute's value collection on :class:`ProvRecord` (#34), as
an internal storage detail -- it is not part of the public API and is
not exported from :mod:`prov.model`. A plain :class:`set` cannot retain
both ``2`` and ``2.0``, or both ``1`` and ``True``, because its
membership test is value-based: ``2.0 in {2}`` is ``True`` (equal hash,
equal value), so ``{2}.add(2.0)`` silently does nothing -- whichever
value was inserted first wins and the other is lost. Keying on
``(type(value), value)`` instead keeps values distinct across Python
types while leaving same-type dedup semantics (including
:class:`Literal`'s ``xsd:decimal`` value-space equality, #77) unchanged,
since two values of the same type still collide on the same key.
The retained values are observable through :attr:`ProvRecord.attributes`
/ :attr:`ProvRecord.extra_attributes` (plain ``(name, value)`` tuples,
one per retained value), record equality/hashing, and serialization.
:meth:`ProvRecord.get_attribute`, :meth:`ProvRecord.get_asserted_types`
and :attr:`ProvRecord.value` deliberately keep their 2.x return type
(a plain ``set``, built fresh from this container) instead of exposing
this class, so those three accessors re-collapse a Python-equal-but-
differently-typed pair in their *returned copy* even though the record's
own storage does not -- see ``docs/upgrading-3.0.md``. Compares equal to
a plain ``set``/``frozenset`` holding the same elements (via the
``collections.abc.Set`` mixin), remains unhashable like a plain ``set``,
and iterates in insertion order.
"""
__slots__ = ("_index",)
def __init__(self, iterable: Iterable[Any] = ()) -> None:
self._index: dict[tuple[type, Any], Any] = {}
for value in iterable:
self.add(value)
def add(self, value: Any) -> None:
# setdefault, not assignment: matches plain set.add's first-wins
# semantics when an equal-typed, equal-valued item is re-added --
# the incumbent object is retained, the new one discarded. This
# matters for values that are __eq__-equal but not identical, such
# as two xsd:decimal Literals with different lexical forms (#77) or
# two langtag Literals differing only in tag case (#259); it also
# keeps this container's semantics consistent with
# ProvRecord.add_attributes()'s own cardinality guard, which keeps
# the first value seen and ignores a later "same value" one.
self._index.setdefault((type(value), value), value)
def discard(self, value: Any) -> None:
self._index.pop((type(value), value), None)
def __contains__(self, value: object) -> bool:
return (type(value), value) in self._index
def __iter__(self) -> Iterator[Any]:
return iter(self._index.values())
def __len__(self) -> int:
return len(self._index)
def __repr__(self) -> str:
return f"{type(self).__name__}({list(self._index.values())!r})"
def _ensure_multiline_string_triple_quoted(value: str) -> str:
# converting the value to a string
s = str(value)
# Escape backslashes first, so quote-escaping below doesn't re-escape
# the backslashes it just introduced.
s = s.replace("\\", "\\\\")
# Escaping any double quote
s = s.replace('"', '\\"')
if "\n" in s:
return f'"""{s}"""'
else:
return f'"{s}"'
def encoding_provn_value(
value: str | datetime.datetime | float | bool | int | QualifiedName,
) -> str:
"""Return the PROV-N literal representation of a Python value.
Strings are quoted (triple-quoted when they span multiple lines); dates
and booleans are rendered with their XSD datatype suffix. Floats are
rendered as full-precision ``xsd:double`` (#251). Plain ints are typed by
magnitude: within +/-(2**31-1) they render as a bare ``INT_LITERAL`` (PROV-N
[60] sugar for ``xsd:int``); beyond that they carry an explicit
``xsd:long``/``xsd:integer`` suffix (#249). Any other value is rendered
via :func:`str`.
"""
if isinstance(value, str):
return _ensure_multiline_string_triple_quoted(value)
elif isinstance(value, datetime.datetime):
return f'"{value.isoformat()}" %% xsd:dateTime'
elif isinstance(value, float):
return f'"{value!r}" %% xsd:double'
elif isinstance(value, bool):
# bool is an int subtype, so :d renders "1"/"0" (not "True"/"False")
return f'"{value:d}" %% xsd:boolean'
elif isinstance(value, int):
datatype = canonical_xsd_datatype(value)
if datatype == XSD_INT:
return str(value) # bare INT_LITERAL is xsd:int sugar (PROV-N [60])
return f'"{value}" %% {datatype}'
else:
# TODO: QName export
return str(value)
[docs]
class Literal:
"""A typed (and optionally language-tagged) PROV literal value.
A literal pairs a string value with an optional datatype and an optional
language tag. Supplying a language tag forces the datatype to
``prov:InternationalizedString``: if no datatype was given one is assumed,
and any other datatype is overridden (with a warning) to comply with the
PROV-JSON/PROV-XML rules for language-tagged strings.
"""
def __init__(
self,
value: Any,
datatype: QualifiedName | None = None,
langtag: str | None = None,
):
"""Initialise the literal.
Args:
value: The literal's value; it is stored as its string form.
datatype: The qualified name of the value's datatype (default:
``None``).
langtag: An optional language tag. When given, ``datatype`` is
coerced to ``prov:InternationalizedString`` (default: ``None``).
"""
self._value: str = str(value) # value is always a string
if langtag:
if datatype is None:
logger.debug(
"Assuming prov:InternationalizedString as the type of "
f'"{value}"@{langtag}'
)
datatype = PROV_INTERNATIONALIZEDSTRING
# PROV JSON states that the type field must not be set when
# using the lang attribute and PROV XML requires it to be an
# internationalized string.
elif datatype != PROV_INTERNATIONALIZEDSTRING:
logger.warning(
f'Invalid data type ({datatype}) for "{value}"@{langtag}, overridden as '
"prov:InternationalizedString."
)
datatype = PROV_INTERNATIONALIZEDSTRING
self._datatype: QualifiedName | None = datatype
# langtag is always a string
self._langtag: str | None = str(langtag) if langtag is not None else None
def __str__(self) -> str:
return self.provn_representation()
def __repr__(self) -> str:
return f"<Literal: {self.provn_representation()}>"
def __eq__(self, other: Any) -> bool:
return (
(
self._comparison_value() == other._comparison_value()
and self._datatype == other.datatype
and self._comparison_langtag() == other._comparison_langtag()
)
if isinstance(other, Literal)
else False
)
def __ne__(self, other: Any) -> bool:
return not (self == other)
def __hash__(self) -> int:
return hash(
(self._comparison_value(), self._datatype, self._comparison_langtag())
)
def _comparison_value(self) -> str | decimal.Decimal:
"""The value to use for equality/hash: value-space for xsd:decimal (#77).
``xsd:decimal`` denotes an arbitrary-precision decimal number, so
``10``, ``10.0`` and ``"10.00"`` are the same value; falls back to
the stored lexical string (unaffected) if it is not a valid decimal.
Safe even though this only checks ``self``: ``__eq__``/``__hash__``
separately require ``self._datatype == other.datatype``, so a
``Decimal`` from here is only ever compared against another
``Decimal`` from an ``XSD_DECIMAL`` literal, never against a
lexical-string value from a different datatype.
"""
if self._datatype == XSD_DECIMAL:
try:
return decimal.Decimal(self._value)
except decimal.InvalidOperation:
pass
return self._value
def _comparison_langtag(self) -> str | None:
"""The language tag to use for equality/hash: case-folded (#259).
RDF 1.1 language tags are case-insensitive; the stored tag itself is
left untouched so serialized output preserves its original case.
"""
return self._langtag.casefold() if self._langtag is not None else None
@property
def value(self) -> str:
"""The literal's value as a string."""
return self._value
@property
def datatype(self) -> QualifiedName | None:
"""The literal's datatype, or ``None`` if it has none."""
return self._datatype
@property
def langtag(self) -> str | None:
"""The literal's language tag, or ``None`` if it has none."""
return self._langtag
[docs]
def has_no_langtag(self) -> bool:
"""Return ``True`` if the literal has no language tag."""
return self._langtag is None
[docs]
def provn_representation(self) -> str:
"""Return the PROV-N representation of the literal."""
quoted_value = _ensure_multiline_string_triple_quoted(self._value)
if self._langtag:
# a language tag can only go with prov:InternationalizedString
return f"{quoted_value}@{self._langtag!s}"
else:
return f"{quoted_value} %% {self._datatype!s}"
# Depends on `Literal` and `SupportedXSDParsedTypes` above, so it cannot join
# the "Type aliases for convenience" block near the top of the module.
CoercedAttributeValue: typing.TypeAlias = (
QualifiedName | datetime.datetime | Literal | SupportedXSDParsedTypes
)
# Exceptions and warnings
[docs]
class ProvException(Error):
"""Base class for PROV model exceptions."""
[docs]
class ProvWarning(Warning):
"""Base class for PROV model warnings."""
[docs]
class ProvExceptionInvalidQualifiedName(ProvException):
"""Exception for an invalid qualified identifier name."""
qname = None
"""The invalid qualified name that triggered the exception."""
def __init__(self, qname: Any):
"""Initialise the exception.
Args:
qname: The invalid qualified name.
"""
self.qname = qname
def __str__(self) -> str:
return f"Invalid Qualified Name: {self.qname}"
[docs]
class ProvUnificationError(ProvException):
"""Raised by :meth:`ProvBundle.unified` when records sharing an
identifier cannot be merged under PROV-CONSTRAINTS term unification
(two different concrete values for the same formal attribute, or
records of incompatible types).
"""
[docs]
class ProvElementIdentifierRequired(ProvException):
"""Exception for a missing element identifier."""
def __str__(self) -> str:
return "An identifier is missing. All PROV elements require a valid identifier."
# PROV records
class ProvRecord:
"""Base class for PROV records."""
FORMAL_ATTRIBUTES: tuple[QualifiedName, ...] = ()
"""Formal attributes names of this record type, in the expected order."""
_prov_type: QualifiedName | None = None
"""PROV type of record."""
def __init__(
self,
bundle: ProvBundle,
identifier: QualifiedName | None,
attributes: RecordAttributesArg | None = None,
):
"""Initialise the record.
Args:
bundle: The bundle owning this PROV record.
identifier: The (unique) identifier of the record.
attributes: Attributes to associate with the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
"""
self._bundle = bundle
self._identifier = identifier
self._attributes: dict[QualifiedName, TypedValueSet] = defaultdict(
TypedValueSet
)
if attributes:
self.add_attributes(attributes)
def _typed_attributes(self) -> frozenset[tuple[QualifiedName, type, Any]]:
"""``(name, type(value), value)`` triples, for equality and hashing.
:attr:`attributes` yields plain ``(name, value)`` tuples, and a
``frozenset``/``set`` of *those* would re-collapse a Python-equal-
but-differently-typed pair like ``(attr, 2)`` and ``(attr, 2.0)`` --
tuple equality and hashing both fall through to their elements', and
``2 == 2.0`` with equal hashes. Including each value's type in the
key keeps the distinction :class:`TypedValueSet` retains in storage
(#34) intact through comparison and hashing too.
"""
return frozenset(
(attr_name, type(value), value) for attr_name, value in self.attributes
)
def __hash__(self) -> int:
return hash((self.get_type(), self._identifier, self._typed_attributes()))
def copy(self) -> ProvRecord:
"""Return an exact copy of this record."""
return PROV_REC_CLS[self.get_type()](
self._bundle, self.identifier, self.attributes
)
def get_type(self) -> QualifiedName:
"""Return the PROV type of the record.
Raises:
NotImplementedError: If the record type is undefined (i.e. on the
abstract base classes).
"""
if self._prov_type is not None:
return self._prov_type
else:
raise NotImplementedError("Type not defined for this record.")
def get_asserted_types(self) -> set[QualifiedName]:
"""Return the set of all asserted PROV types of this record.
Returns a fresh, plain ``set`` copy (2.x-compatible: mutating it does
not affect the record). Since ``prov:type`` values are always
:class:`~prov.identifier.QualifiedName`\\ s -- which never collapse
under Python equality -- this copy never loses information: unlike
:meth:`get_attribute`/:attr:`value`, there is no lossy case here.
"""
return set(self._attributes[PROV_TYPE])
def add_asserted_type(self, type_identifier: QualifiedName) -> None:
"""Add a PROV type assertion to the record.
Args:
type_identifier: The qualified name of the type to assert.
"""
self._attributes[PROV_TYPE].add(type_identifier)
def get_attribute(self, attr_name: QualifiedNameCandidate) -> set[Any]:
"""Return the values (if any) for the named attribute.
Args:
attr_name: The name of the attribute.
Returns:
A fresh, plain ``set`` copy of the values held for the attribute
(empty if none); mutating it does not affect the record. This
keeps the 2.x return type: a Python-equal-but-differently-typed
pair retained on the record (e.g. ``2`` and ``2.0``, #34) still
collapses to whichever was asserted first in *this copy*, even
though the record's own storage, :attr:`attributes`/
:attr:`extra_attributes`, equality/hashing and serialization all
retain both -- see ``docs/upgrading-3.0.md``.
Raises:
ProvExceptionInvalidQualifiedName: If ``attr_name`` cannot be
resolved to a valid qualified name.
"""
attr_name_qn = self._bundle.mandatory_valid_qname(attr_name)
return set(self._attributes[attr_name_qn])
@property
def identifier(self) -> QualifiedName | None:
"""The record's identifier, or ``None`` if it has none."""
return self._identifier
@property
def attributes(self) -> list[tuple[QualifiedName, Any]]:
"""All of the record's attributes as a list of ``(name, value)`` pairs.
Attributes with multiple values appear once per value, so the same name
may occur more than once.
"""
return [
(attr_name, value)
for attr_name, values in self._attributes.items()
for value in values
]
@property
def args(self) -> tuple[Any, ...]:
"""The values of the record's formal attributes, in declaration order.
Missing formal attributes are represented by ``None``.
"""
return tuple(
first(self._attributes[attr_name]) for attr_name in self.FORMAL_ATTRIBUTES
)
@property
def formal_attributes(self) -> tuple[tuple[QualifiedName, Any], ...]:
"""The record's formal attributes as ``(name, value)`` pairs.
Pairs are in declaration order; a missing attribute has a value of
``None``.
"""
return tuple(
(attr_name, first(self._attributes[attr_name]))
for attr_name in self.FORMAL_ATTRIBUTES
)
@property
def extra_attributes(self) -> tuple[tuple[QualifiedName, Any], ...]:
"""The record's non-formal attributes as ``(name, value)`` pairs."""
return tuple(
(attr_name, attr_value)
for attr_name, attr_value in self.attributes
if attr_name not in self.FORMAL_ATTRIBUTES
)
@property
def bundle(self) -> ProvBundle:
"""The bundle that owns this record."""
return self._bundle
@property
def label(self) -> str:
"""The record's identifying label.
This is the record's ``prov:label`` attribute if set, otherwise its
identifier.
"""
return str(
first(self._attributes[PROV_LABEL])
if self._attributes[PROV_LABEL]
else self._identifier
)
@property
def value(self) -> set[Any]:
"""The set of the record's ``prov:value`` attribute values.
Returns a fresh, plain ``set`` copy (2.x-compatible); see
:meth:`get_attribute` for what that means for a Python-equal-but-
differently-typed pair of ``prov:value`` values (#34).
"""
return set(self._attributes[PROV_VALUE])
# Handling attributes
def _auto_literal_conversion(self, literal: Any) -> Any:
# This method normalise datatype for literals
if isinstance(literal, ProvRecord):
# Use the QName of the record as the literal
literal = literal.identifier
if isinstance(literal, str):
return str(literal)
elif isinstance(literal, QualifiedName):
return self._bundle.valid_qualified_name(literal)
elif isinstance(literal, Literal) and literal.has_no_langtag():
if literal.datatype == PROV_QUALIFIEDNAME:
# #238: a prov:QUALIFIED_NAME-typed Literal (e.g. decoded from
# a legacy PROV-JSON document, or asserted directly) denotes a
# QualifiedName; resolve it against this record's bundle
# namespaces. If the prefix has no in-scope namespace, keep
# the opaque Literal rather than reject it (#257 lock). Only
# PROV_QUALIFIEDNAME is handled here: an XSD_QNAME literal
# keeps today's opaque model-side behaviour -- the PROV-JSON
# codec is the only place that treats xsd:QName as a
# QualifiedName value, per the submission (#168).
resolved = self._bundle.valid_qualified_name(literal.value)
return resolved if resolved is not None else literal
if literal.datatype:
# try to convert a generic Literal object to Python standard type
# to match the JSON decoding's literal conversion
value = parse_xsd_types(literal.value, literal.datatype)
# #235: only collapse the integer family when it is lossless,
# i.e. the asserted datatype is the one `prov` would itself
# infer for the parsed value. Otherwise, keep the Literal so
# its asserted datatype survives serialization (e.g.
# Literal("42", XSD_LONG)).
if (
value is not None
and literal.datatype in (XSD_LONG, XSD_INT, XSD_INTEGER)
and canonical_xsd_datatype(value) != literal.datatype
):
return literal
else:
# A literal with no datatype nor langtag defined
# try auto-converting the value
value = self._auto_literal_conversion(literal.value)
if value is not None:
return value
# No conversion possible, return the original value
return literal
def _coerce_attribute_value(
self, attr: QualifiedName, original_value: Any
) -> CoercedAttributeValue:
# Normalise `original_value` to the datatype expected for `attr`.
#
# Raises:
# ProvException: If the value is invalid for `attr`.
# the branches below bind `value` to different types
value: CoercedAttributeValue
if attr in PROV_ATTRIBUTE_QNAMES:
# Expecting a qualified name
if isinstance(original_value, ProvRecord):
# Use the identifier of the record, which must exist, as the value for this attribute
qname = original_value.identifier
if qname is None:
raise ProvException(
f"Invalid value for attribute {attr}: {original_value}."
f" The record has no identifier."
)
else:
qname = original_value
value = self._bundle.mandatory_valid_qname(qname)
elif attr in PROV_ATTRIBUTE_LITERALS:
# Expecting a datetime object or a string that can be parsed as a datetime
if isinstance(original_value, str):
value = parse_xsd_datetime(original_value)
else:
value = original_value
if not isinstance(value, datetime.datetime):
raise ProvException(
f"Invalid value for attribute {attr}: {original_value}. "
f"Expected a datetime object or a string that can be parsed"
f" as a datetime."
)
else:
value = self._auto_literal_conversion(original_value)
if value is None:
raise ProvException(f"Invalid value for attribute {attr}: {original_value}")
return value
def _store_attribute_value(
self,
attr: QualifiedName,
value: CoercedAttributeValue,
is_collection: bool,
) -> None:
# Add `value` for `attr`, enforcing single-valued (non-collection)
# attributes have at most one (distinct) value.
#
# Raises:
# ProvException: If a second, different value is supplied for a
# single-valued (non-collection) attribute.
#
# `_attributes` is a defaultdict(TypedValueSet): every code path below
# ends up subscripting `self._attributes[attr]` at least once (the
# early-return and raise paths via the guard/`first()`, the fall-through
# path via `.add()`), so binding it once up front auto-vivifies the
# entry no earlier than it would have been created anyway.
existing_values = self._attributes[attr]
if not is_collection and attr in PROV_ATTRIBUTES and existing_values:
existing_value = first(existing_values)
is_not_same_value = True
# This duplicate-value branch runs at scale in
# _unified_records()'s merge loop (unified()/flattened() on
# large documents), where contextlib.suppress()'s per-call
# context-manager overhead adds up — the plain try/except
# stays here.
try: # noqa: SIM105
is_not_same_value = value != existing_value
except TypeError:
# Cannot compare them
pass # consider them different values
if is_not_same_value:
raise ProvException(
f"Cannot have more than one value for attribute {attr}"
)
else:
# Same value, ignore it
return
existing_values.add(value)
def add_attributes(self, attributes: RecordAttributesArg) -> None:
"""Add attributes to the record.
Attribute names are resolved to qualified names, and values are
normalised to the datatype expected for the attribute. ``None`` values
are skipped.
Args:
attributes: The attributes to add, either as a dict keyed by
qualified-name identifiers or an iterable of ``(name, value)``
pairs whose names satisfy the same condition.
Raises:
ProvExceptionInvalidQualifiedName: If an attribute name cannot be
resolved to a valid qualified name.
ProvException: If a value is invalid for its attribute, or a
second, different value is supplied for a single-valued
(non-collection) attribute.
"""
if attributes:
if isinstance(attributes, dict):
# Converting the dictionary into a list of tuples
# (i.e. attribute-value pairs)
attributes = cast(
"dict[QualifiedNameCandidate, Any]", attributes
).items()
# Check if one of the attributes specifies that the current type
# is a collection. In that case multiple attributes of the same
# type are allowed.
is_collection = any(
attr_name == PROV_ATTR_COLLECTION for attr_name, _ in attributes
)
for attr_name, original_value in attributes:
if original_value is None:
continue
# make sure the attribute name is valid
attr = self._bundle.mandatory_valid_qname(attr_name)
value = self._coerce_attribute_value(attr, original_value)
self._store_attribute_value(attr, value, is_collection)
def __eq__(self, other: Any) -> bool:
if not isinstance(other, ProvRecord):
return False
if self.get_type() != other.get_type():
return False
if self._identifier and not (self._identifier == other._identifier):
return False
return self._typed_attributes() == other._typed_attributes()
def __str__(self) -> str:
return self.get_provn()
def get_provn(self) -> str:
"""Return the PROV-N representation of the record."""
items = []
# Generating identifier
relation_id = "" # default blank
if self._identifier:
# #223: escape PN_CHARS_ESC metacharacters in the local part
identifier = self._identifier.provn_bare_representation()
if self.is_element():
items.append(identifier)
else:
# this is a relation, which relation uses a semicolon to separate identifiers
relation_id = identifier + "; "
# Writing out the formal attributes
for attr in self.FORMAL_ATTRIBUTES:
values = self._attributes.get(attr)
if values:
# Formal attributes always have single values
value = first(values)
if isinstance(value, datetime.datetime):
items.append(value.isoformat())
elif isinstance(value, QualifiedName):
# #223: escape PN_CHARS_ESC metacharacters in the local part
items.append(value.provn_bare_representation())
else:
items.append(str(value))
else:
items.append("-")
# Writing out the remaining attributes
extra = []
for attr in self._attributes:
if attr not in self.FORMAL_ATTRIBUTES:
for value in self._attributes[attr]:
try:
# try if there is a prov-n representation defined
provn_represenation = value.provn_representation()
except AttributeError:
provn_represenation = encoding_provn_value(value)
# #223: escape PN_CHARS_ESC metacharacters in the local part
attr_name = attr.provn_bare_representation()
extra.append(f"{attr_name}={provn_represenation}")
if extra:
# .format(), not an f-string: the nested string literals reuse the
# same quote character, which f-strings only allow from py3.12 (PEP 701)
items.append("[{}]".format(", ".join(extra)))
prov_n = "{}({}{})".format(
PROV_N_MAP[self.get_type()],
relation_id,
", ".join(items),
)
return prov_n
def is_element(self) -> bool:
"""Return ``True`` if the record is an element, ``False`` otherwise."""
return False
def is_relation(self) -> bool:
"""Return ``True`` if the record is a relation, ``False`` otherwise."""
return False
# Abstract classes for elements and relations
class ProvElement(ProvRecord):
"""Provenance Element (nodes in the provenance graph)."""
def __init__(
self,
bundle: ProvBundle,
identifier: QualifiedName | None,
attributes: RecordAttributesArg | None = None,
):
if identifier is None:
# All types of PROV elements require a valid identifier
raise ProvElementIdentifierRequired()
super().__init__(bundle, identifier, attributes)
def is_element(self) -> bool:
"""Return ``True`` if the record is an element, ``False`` otherwise."""
return True
def __repr__(self) -> str:
return f"<{self.__class__.__name__}: {self._identifier}>"
class ProvRelation(ProvRecord):
"""Provenance Relationship (edge between nodes)."""
def is_relation(self) -> bool:
"""Return ``True`` if the record is a relation, ``False`` otherwise."""
return True
def __repr__(self) -> str:
identifier = f" {self._identifier}" if self._identifier else ""
element_1, element_2 = [qname for _, qname in self.formal_attributes[:2]]
return f"<{self.__class__.__name__}:{identifier} ({element_1}, {element_2})>"
# Component 1: Entities and Activities
[docs]
class ProvEntity(ProvElement):
"""Provenance Entity element"""
_prov_type = PROV_ENTITY
# Convenient assertions that take the current ProvEntity as the first
# (formal) argument
[docs]
def wasGeneratedBy(
self,
activity: ActivityRef | None = None,
time: DatetimeOrStr | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvEntity:
"""Create a new generation record to this entity.
Args:
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 an ``xsd:dateTime`` string accepted by
:func:`~prov.model.parse_xsd_datetime` (default: ``None``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This entity (to allow chaining).
"""
self._bundle.generation(self, activity, time, other_attributes=attributes)
return self
[docs]
def wasInvalidatedBy(
self,
activity: ActivityRef | None,
time: DatetimeOrStr | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvEntity:
"""Create a new invalidation record for this entity.
Args:
activity: The activity (or its string identifier) involved in the
invalidation; may be ``None``.
time: Optional time of the invalidation, as a
:class:`datetime.datetime` or an ``xsd:dateTime`` string accepted by
:func:`~prov.model.parse_xsd_datetime` (default: ``None``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This entity (to allow chaining).
"""
self._bundle.invalidation(self, activity, time, other_attributes=attributes)
return self
[docs]
def wasDerivedFrom(
self,
usedEntity: EntityRef,
activity: ActivityRef | None = None,
generation: GenerationRef | None = None,
usage: UsageRef | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvEntity:
"""Create a new derivation record for this entity from a used entity.
Args:
usedEntity: The used entity (or its string identifier).
activity: The activity (or its string identifier) involved in the
derivation (default: ``None``).
generation: Optional generation record qualifying the derivation
through an internal generation (default: ``None``).
usage: Optional usage record qualifying the derivation through an
internal usage (default: ``None``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This entity (to allow chaining).
"""
self._bundle.derivation(
self, usedEntity, activity, generation, usage, other_attributes=attributes
)
return self
[docs]
def wasAttributedTo(
self, agent: AgentRef, attributes: RecordAttributesArg | None = None
) -> ProvEntity:
"""Create a new attribution record between this entity and an agent.
Args:
agent: The agent (or its string identifier) involved in the
attribution.
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This entity (to allow chaining).
"""
self._bundle.attribution(self, agent, other_attributes=attributes)
return self
[docs]
def alternateOf(self, alternate2: EntityRef) -> ProvEntity:
"""Create a new alternate record between this and another entity.
Args:
alternate2: The other entity (or its string identifier).
Returns:
This entity (to allow chaining).
"""
self._bundle.alternate(self, alternate2)
return self
[docs]
def specializationOf(self, generalEntity: EntityRef) -> ProvEntity:
"""Create a new specialisation record for this from a general entity.
Args:
generalEntity: The general entity (or its string identifier).
Returns:
This entity (to allow chaining).
"""
self._bundle.specialization(self, generalEntity)
return self
[docs]
def hadMember(self, entity: EntityRef) -> ProvEntity:
"""Create a new membership record adding an entity to this collection.
Args:
entity: The entity (or its string identifier) to add to the
collection.
Returns:
This entity (to allow chaining).
"""
self._bundle.membership(self, entity)
return self
[docs]
def wasRevisionOf(
self,
usedEntity: EntityRef,
activity: ActivityRef | None = None,
generation: GenerationRef | None = None,
usage: UsageRef | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvEntity:
"""Create a new revision record for this entity from a used entity.
Args:
usedEntity: The original entity (or its string identifier).
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``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This entity (to allow chaining).
"""
self._bundle.revision(
self, usedEntity, activity, generation, usage, other_attributes=attributes
)
return self
[docs]
def wasQuotedFrom(
self,
usedEntity: EntityRef,
activity: ActivityRef | None = None,
generation: GenerationRef | None = None,
usage: UsageRef | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvEntity:
"""Create a new quotation record for this entity from a quoted entity.
Args:
usedEntity: The quoted entity (or its string identifier).
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``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This entity (to allow chaining).
"""
self._bundle.quotation(
self, usedEntity, activity, generation, usage, other_attributes=attributes
)
return self
[docs]
def hadPrimarySource(
self,
usedEntity: EntityRef,
activity: ActivityRef | None = None,
generation: GenerationRef | None = None,
usage: UsageRef | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvEntity:
"""Create a new primary-source record for this entity.
Args:
usedEntity: The primary-source entity (or its string identifier).
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``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This entity (to allow chaining).
"""
self._bundle.primary_source(
self, usedEntity, activity, generation, usage, other_attributes=attributes
)
return self
[docs]
def mentionOf(self, generalEntity: EntityRef, bundle: EntityRef) -> ProvEntity:
"""Create a new mention record of this entity from a general entity.
Args:
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:
This entity (to allow chaining).
"""
self._bundle.mention(self, generalEntity, bundle)
return self
[docs]
def wasInfluencedBy(
self,
influencer: InfluencerRef,
attributes: RecordAttributesArg | None = None,
) -> ProvEntity:
"""Create a new influence record on this entity by an influencer.
Args:
influencer: The influencing entity, activity or agent (or its
string identifier).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This entity (to allow chaining).
"""
self._bundle.influence(self, influencer, other_attributes=attributes)
return self
[docs]
class ProvActivity(ProvElement):
"""Provenance Activity element."""
FORMAL_ATTRIBUTES = (PROV_ATTR_STARTTIME, PROV_ATTR_ENDTIME)
_prov_type = PROV_ACTIVITY
# Convenient methods
[docs]
def set_time(
self,
startTime: datetime.datetime | None = None,
endTime: datetime.datetime | None = None,
) -> None:
"""Set the start and/or end time of this activity.
Only non-``None`` arguments are applied; the values are stored as
given (no string parsing is performed here).
Args:
startTime: The start time as a :class:`datetime.datetime`
(default: ``None``).
endTime: The end time as a :class:`datetime.datetime`
(default: ``None``).
"""
if startTime is not None:
self._attributes[PROV_ATTR_STARTTIME] = TypedValueSet([startTime])
if endTime is not None:
self._attributes[PROV_ATTR_ENDTIME] = TypedValueSet([endTime])
[docs]
def get_startTime(self) -> datetime.datetime | None:
"""Return the activity's start time, or ``None`` if unset."""
values = self._attributes[PROV_ATTR_STARTTIME]
return first(values) if values else None
[docs]
def get_endTime(self) -> datetime.datetime | None:
"""Return the activity's end time, or ``None`` if unset."""
values = self._attributes[PROV_ATTR_ENDTIME]
return first(values) if values else None
# Convenient assertions that take the current ProvActivity as the first
# (formal) argument
[docs]
def used(
self,
entity: EntityRef,
time: DatetimeOrStr | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvActivity:
"""Create a new usage record for this activity.
Args:
entity: The entity (or its string identifier) involved in the
usage relationship.
time: Optional time of the usage, as a :class:`datetime.datetime`
or an ``xsd:dateTime`` string accepted by
:func:`~prov.model.parse_xsd_datetime`
(default: ``None``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This activity (to allow chaining).
"""
self._bundle.usage(self, entity, time, other_attributes=attributes)
return self
[docs]
def wasStartedBy(
self,
trigger: EntityRef | None,
starter: ActivityRef | None = None,
time: DatetimeOrStr | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvActivity:
"""Create a new start record for this activity.
The activity did not exist before being started by the trigger.
Args:
trigger: The entity triggering the start of this activity; may be
``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 an ``xsd:dateTime`` string accepted by
:func:`~prov.model.parse_xsd_datetime`
(default: ``None``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This activity (to allow chaining).
"""
self._bundle.start(self, trigger, starter, time, other_attributes=attributes)
return self
[docs]
def wasEndedBy(
self,
trigger: EntityRef | None,
ender: ActivityRef | None = None,
time: DatetimeOrStr | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvActivity:
"""Create a new end record for this activity.
Args:
trigger: The entity triggering the end of this activity; may be
``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
an ``xsd:dateTime`` string accepted by
:func:`~prov.model.parse_xsd_datetime`
(default: ``None``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This activity (to allow chaining).
"""
self._bundle.end(self, trigger, ender, time, other_attributes=attributes)
return self
[docs]
def wasAssociatedWith(
self,
agent: AgentRef,
plan: EntityRef | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvActivity:
"""Create a new association record for this activity.
Args:
agent: The agent (or its string identifier) involved in the
association.
plan: Optional entity qualifying the association through an
internal plan (default: ``None``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This activity (to allow chaining).
"""
self._bundle.association(self, agent, plan, other_attributes=attributes)
return self
[docs]
def wasInfluencedBy(
self,
influencer: InfluencerRef,
attributes: RecordAttributesArg | None = None,
) -> ProvActivity:
"""Create a new influence record on this activity by an influencer.
Args:
influencer: The influencing entity, activity or agent (or its
string identifier).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This activity (to allow chaining).
"""
self._bundle.influence(self, influencer, other_attributes=attributes)
return self
[docs]
class ProvGeneration(ProvRelation):
"""Provenance Generation relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_ENTITY, PROV_ATTR_ACTIVITY, PROV_ATTR_TIME)
_prov_type = PROV_GENERATION
[docs]
class ProvUsage(ProvRelation):
"""Provenance Usage relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_ACTIVITY, PROV_ATTR_ENTITY, PROV_ATTR_TIME)
_prov_type = PROV_USAGE
[docs]
class ProvCommunication(ProvRelation):
"""Provenance Communication relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_INFORMED, PROV_ATTR_INFORMANT)
_prov_type = PROV_COMMUNICATION
[docs]
class ProvStart(ProvRelation):
"""Provenance Start relationship."""
FORMAL_ATTRIBUTES = (
PROV_ATTR_ACTIVITY,
PROV_ATTR_TRIGGER,
PROV_ATTR_STARTER,
PROV_ATTR_TIME,
)
_prov_type = PROV_START
[docs]
class ProvEnd(ProvRelation):
"""Provenance End relationship."""
FORMAL_ATTRIBUTES = (
PROV_ATTR_ACTIVITY,
PROV_ATTR_TRIGGER,
PROV_ATTR_ENDER,
PROV_ATTR_TIME,
)
_prov_type = PROV_END
[docs]
class ProvInvalidation(ProvRelation):
"""Provenance Invalidation relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_ENTITY, PROV_ATTR_ACTIVITY, PROV_ATTR_TIME)
_prov_type = PROV_INVALIDATION
# Component 2: Derivations
[docs]
class ProvDerivation(ProvRelation):
"""Provenance Derivation relationship."""
FORMAL_ATTRIBUTES = (
PROV_ATTR_GENERATED_ENTITY,
PROV_ATTR_USED_ENTITY,
PROV_ATTR_ACTIVITY,
PROV_ATTR_GENERATION,
PROV_ATTR_USAGE,
)
_prov_type = PROV_DERIVATION
# Component 3: Agents, Responsibility, and Influence
[docs]
class ProvAgent(ProvElement):
"""Provenance Agent element."""
_prov_type = PROV_AGENT
# Convenient assertions that take the current ProvAgent as the first
# (formal) argument
[docs]
def actedOnBehalfOf(
self,
responsible: AgentRef,
activity: ActivityRef | None = None,
attributes: RecordAttributesArg | None = None,
) -> ProvAgent:
"""Create a new delegation record on behalf of this agent.
Args:
responsible: The agent (or its string identifier) that the
responsibility is delegated to.
activity: Optional activity qualifying the delegation
(default: ``None``).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This agent (to allow chaining).
"""
self._bundle.delegation(
self, responsible, activity, other_attributes=attributes
)
return self
[docs]
def wasInfluencedBy(
self,
influencer: InfluencerRef,
attributes: RecordAttributesArg | None = None,
) -> ProvAgent:
"""Create a new influence record on this agent by an influencer.
Args:
influencer: The influencing entity, activity or agent (or its
string identifier).
attributes: Optional extra attributes for the record, as a dict or
an iterable of ``(name, value)`` pairs (default: ``None``).
Returns:
This agent (to allow chaining).
"""
self._bundle.influence(self, influencer, other_attributes=attributes)
return self
[docs]
class ProvAttribution(ProvRelation):
"""Provenance Attribution relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_ENTITY, PROV_ATTR_AGENT)
_prov_type = PROV_ATTRIBUTION
[docs]
class ProvAssociation(ProvRelation):
"""Provenance Association relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_ACTIVITY, PROV_ATTR_AGENT, PROV_ATTR_PLAN)
_prov_type = PROV_ASSOCIATION
[docs]
class ProvDelegation(ProvRelation):
"""Provenance Delegation relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_DELEGATE, PROV_ATTR_RESPONSIBLE, PROV_ATTR_ACTIVITY)
_prov_type = PROV_DELEGATION
[docs]
class ProvInfluence(ProvRelation):
"""Provenance Influence relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_INFLUENCEE, PROV_ATTR_INFLUENCER)
_prov_type = PROV_INFLUENCE
# Component 5: Alternate Entities
[docs]
class ProvSpecialization(ProvRelation):
"""Provenance Specialization relationship."""
FORMAL_ATTRIBUTES: tuple[QualifiedName, ...] = (
PROV_ATTR_SPECIFIC_ENTITY,
PROV_ATTR_GENERAL_ENTITY,
)
_prov_type = PROV_SPECIALIZATION
[docs]
class ProvAlternate(ProvRelation):
"""Provenance Alternate relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_ALTERNATE1, PROV_ATTR_ALTERNATE2)
_prov_type = PROV_ALTERNATE
[docs]
class ProvMention(ProvSpecialization):
"""Provenance Mention relationship (specific Specialization)."""
FORMAL_ATTRIBUTES = (
PROV_ATTR_SPECIFIC_ENTITY,
PROV_ATTR_GENERAL_ENTITY,
PROV_ATTR_BUNDLE,
)
_prov_type = PROV_MENTION
# Component 6: Collections
[docs]
class ProvMembership(ProvRelation):
"""Provenance Membership relationship."""
FORMAL_ATTRIBUTES = (PROV_ATTR_COLLECTION, PROV_ATTR_ENTITY)
_prov_type = PROV_MEMBERSHIP
# Class mappings from PROV record type
PROV_REC_CLS = {
PROV_ENTITY: ProvEntity,
PROV_ACTIVITY: ProvActivity,
PROV_GENERATION: ProvGeneration,
PROV_USAGE: ProvUsage,
PROV_COMMUNICATION: ProvCommunication,
PROV_START: ProvStart,
PROV_END: ProvEnd,
PROV_INVALIDATION: ProvInvalidation,
PROV_DERIVATION: ProvDerivation,
PROV_AGENT: ProvAgent,
PROV_ATTRIBUTION: ProvAttribution,
PROV_ASSOCIATION: ProvAssociation,
PROV_DELEGATION: ProvDelegation,
PROV_INFLUENCE: ProvInfluence,
PROV_SPECIALIZATION: ProvSpecialization,
PROV_ALTERNATE: ProvAlternate,
PROV_MENTION: ProvMention,
PROV_MEMBERSHIP: ProvMembership,
}