"""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 logging
import os
import re
import typing # noqa: F401 -- used by `# type: typing.TypeAlias` comments below
from collections import defaultdict
from collections.abc import Callable, Iterable, Iterator, MutableSet
from typing import TYPE_CHECKING, Any, Union
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 = QualifiedName | str | Identifier # type: typing.TypeAlias
OptionalID = QualifiedNameCandidate | None # type: typing.TypeAlias
EntityRef = Union["ProvEntity", QualifiedNameCandidate] # type: typing.TypeAlias
ActivityRef = Union["ProvActivity", QualifiedNameCandidate] # type: typing.TypeAlias
AgentRef = Union["ProvAgent", "ProvEntity", "ProvActivity", QualifiedNameCandidate] # type: typing.TypeAlias
GenrationRef = Union["ProvGeneration", QualifiedNameCandidate] # type: typing.TypeAlias
UsageRef = Union["ProvUsage", QualifiedNameCandidate] # type: typing.TypeAlias
RecordAttributesArg = (
dict[QualifiedNameCandidate, Any] | Iterable[tuple[QualifiedNameCandidate, Any]]
) # type: typing.TypeAlias
NameValuePair = tuple[QualifiedName, Any] # type: typing.TypeAlias
DatetimeOrStr = datetime.datetime | str # type: typing.TypeAlias
NSCollection = dict[str, str] | Iterable[Namespace] # type: typing.TypeAlias
PathLike = str | bytes | os.PathLike[str] # type: typing.TypeAlias
# 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 = (
str | datetime.datetime | float | int | bool | Identifier | None
) # type: typing.TypeAlias
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}"
# 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 = () # type: 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
) -> QualifiedName | datetime.datetime | Literal | SupportedXSDParsedTypes:
# 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: QualifiedName | datetime.datetime | Literal | SupportedXSDParsedTypes
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: QualifiedName | datetime.datetime | Literal | SupportedXSDParsedTypes,
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.
if not is_collection and attr in PROV_ATTRIBUTES and self._attributes[attr]:
existing_value = first(self._attributes[attr])
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
self._attributes[attr].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 = 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: GenrationRef | 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: GenrationRef | 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: GenrationRef | 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: GenrationRef | 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: EntityRef | ActivityRef | AgentRef,
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: EntityRef | ActivityRef | AgentRef,
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: EntityRef | ActivityRef | AgentRef,
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 = (
PROV_ATTR_SPECIFIC_ENTITY,
PROV_ATTR_GENERAL_ENTITY,
) # type: tuple[QualifiedName, ...]
_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,
}