tessl/pypi-prov
A library for W3C Provenance Data Model supporting PROV-JSON, PROV-XML and PROV-O (RDF)
—
Pending
prov-elements.mddocs/
PROV Elements
Classes representing the core PROV elements: entities (things), activities (processes), and agents (responsible parties). Each element supports relationship creation and attribute management following the W3C PROV Data Model.
Capabilities
ProvElement (Base Class)
Base class for all PROV elements providing common functionality.
class ProvElement(ProvRecord):
def __init__(self, bundle, identifier, attributes=None):
"""
Base class for PROV elements.
Args:
bundle (ProvBundle): Containing bundle
identifier (QualifiedName): Element identifier
attributes (dict, optional): Element attributes
"""
def is_element(self):
"""
Check if this is a PROV element.
Returns:
bool: Always True for elements
"""
def is_relation(self):
"""
Check if this is a PROV relation.
Returns:
bool: Always False for elements
"""ProvEntity
Represents entities - things in the world (physical, digital, conceptual, or otherwise).
class ProvEntity(ProvElement):
def wasGeneratedBy(self, activity, time=None, attributes=None):
"""
Assert that this entity was generated by an activity.
Args:
activity (ProvActivity or QualifiedName): Generating activity
time (datetime or str, optional): Generation time
attributes (dict, optional): Additional attributes
Returns:
ProvGeneration: Created generation record
"""
def wasInvalidatedBy(self, activity, time=None, attributes=None):
"""
Assert that this entity was invalidated by an activity.
Args:
activity (ProvActivity or QualifiedName): Invalidating activity
time (datetime or str, optional): Invalidation time
attributes (dict, optional): Additional attributes
Returns:
ProvInvalidation: Created invalidation record
"""
def wasDerivedFrom(self, other_entity, activity=None, generation=None, usage=None, attributes=None):
"""
Assert that this entity was derived from another entity.
Args:
other_entity (ProvEntity or QualifiedName): Source entity
activity (ProvActivity or QualifiedName, optional): Deriving activity
generation (ProvGeneration or QualifiedName, optional): Generation record
usage (ProvUsage or QualifiedName, optional): Usage record
attributes (dict, optional): Additional attributes
Returns:
ProvDerivation: Created derivation record
"""
def wasAttributedTo(self, agent, attributes=None):
"""
Assert that this entity was attributed to an agent.
Args:
agent (ProvAgent or QualifiedName): Responsible agent
attributes (dict, optional): Additional attributes
Returns:
ProvAttribution: Created attribution record
"""
def alternateOf(self, alternate2):
"""
Assert that this entity is an alternate of another entity.
Args:
alternate2 (ProvEntity or QualifiedName): Alternate entity
Returns:
ProvAlternate: Created alternate record
"""
def specializationOf(self, general_entity):
"""
Assert that this entity is a specialization of another entity.
Args:
general_entity (ProvEntity or QualifiedName): General entity
Returns:
ProvSpecialization: Created specialization record
"""
def hadMember(self, entity):
"""
Assert that this collection entity had a member.
Args:
entity (ProvEntity or QualifiedName): Member entity
Returns:
ProvMembership: Created membership record
"""ProvActivity
Represents activities - processes or functions that act upon or with entities.
class ProvActivity(ProvElement):
def __init__(self, bundle, identifier, startTime=None, endTime=None, other_attributes=None):
"""
Create a PROV activity.
Args:
bundle (ProvBundle): Containing bundle
identifier (QualifiedName): Activity identifier
startTime (datetime or str, optional): Activity start time
endTime (datetime or str, optional): Activity end time
other_attributes (dict, optional): Additional attributes
"""
def set_time(self, startTime=None, endTime=None):
"""
Set the time bounds for this activity.
Args:
startTime (datetime or str, optional): Start time
endTime (datetime or str, optional): End time
"""
def get_startTime(self):
"""
Get the start time of this activity.
Returns:
datetime: Start time or None
"""
def get_endTime(self):
"""
Get the end time of this activity.
Returns:
datetime: End time or None
"""
def used(self, entity, time=None, attributes=None):
"""
Assert that this activity used an entity.
Args:
entity (ProvEntity or QualifiedName): Used entity
time (datetime or str, optional): Usage time
attributes (dict, optional): Additional attributes
Returns:
ProvUsage: Created usage record
"""
def wasInformedBy(self, other_activity, attributes=None):
"""
Assert that this activity was informed by another activity.
Args:
other_activity (ProvActivity or QualifiedName): Informing activity
attributes (dict, optional): Additional attributes
Returns:
ProvCommunication: Created communication record
"""
def wasStartedBy(self, trigger, starter=None, time=None, attributes=None):
"""
Assert that this activity was started by a trigger entity.
Args:
trigger (ProvEntity or QualifiedName): Starting entity
starter (ProvActivity or QualifiedName, optional): Starting activity
time (datetime or str, optional): Start time
attributes (dict, optional): Additional attributes
Returns:
ProvStart: Created start record
"""
def wasEndedBy(self, trigger, ender=None, time=None, attributes=None):
"""
Assert that this activity was ended by a trigger entity.
Args:
trigger (ProvEntity or QualifiedName): Ending entity
ender (ProvActivity or QualifiedName, optional): Ending activity
time (datetime or str, optional): End time
attributes (dict, optional): Additional attributes
Returns:
ProvEnd: Created end record
"""
def wasAssociatedWith(self, agent, plan=None, attributes=None):
"""
Assert that this activity was associated with an agent.
Args:
agent (ProvAgent or QualifiedName): Associated agent
plan (ProvEntity or QualifiedName, optional): Plan entity
attributes (dict, optional): Additional attributes
Returns:
ProvAssociation: Created association record
"""ProvAgent
Represents agents - entities that bear responsibility for activities or other agents' actions.
class ProvAgent(ProvElement):
def actedOnBehalfOf(self, other_agent, activity=None, attributes=None):
"""
Assert that this agent acted on behalf of another agent.
Args:
other_agent (ProvAgent or QualifiedName): Responsible agent
activity (ProvActivity or QualifiedName, optional): Delegating activity
attributes (dict, optional): Additional attributes
Returns:
ProvDelegation: Created delegation record
"""ProvRecord (Base Class)
Base class for all PROV records providing common record functionality.
class ProvRecord:
def get_type(self):
"""
Get the PROV type of this record.
Returns:
QualifiedName: PROV type
"""
def get_asserted_types(self):
"""
Get all asserted types for this record.
Returns:
set[QualifiedName]: Asserted types
"""
def add_asserted_type(self, type_identifier):
"""
Add an asserted type to this record.
Args:
type_identifier (QualifiedName): Type to assert
"""
def get_attribute(self, attr_name):
"""
Get attribute values by name.
Args:
attr_name (QualifiedName or str): Attribute name
Returns:
set: Attribute values
"""
def add_attributes(self, attributes):
"""
Add multiple attributes to this record.
Args:
attributes (dict or iterable): Attributes to add
"""
def copy(self):
"""
Create a copy of this record.
Returns:
ProvRecord: Copied record
"""
def get_provn(self):
"""
Get PROV-N representation of this record.
Returns:
str: PROV-N string
"""
# Properties
@property
def identifier(self):
"""Record identifier (QualifiedName or None)."""
@property
def attributes(self):
"""List of (name, value) attribute pairs."""
@property
def args(self):
"""Tuple of formal arguments."""
@property
def formal_attributes(self):
"""Tuple of formal attributes."""
@property
def extra_attributes(self):
"""Tuple of extra attributes."""
@property
def bundle(self):
"""Containing bundle."""
@property
def label(self):
"""Human-readable label."""
@property
def value(self):
"""Primary value of this record."""Usage Examples
Creating Entities
from prov.model import ProvDocument
doc = ProvDocument()
# Create entities with attributes
entity1 = doc.entity('ex:dataset1', {
'prov:label': 'My Dataset',
'prov:type': 'prov:Collection',
'ex:size': 1000
})
entity2 = doc.entity('ex:derivedDataset', {
'prov:label': 'Processed Dataset'
})
# Entities can be derived from other entities
entity2.wasDerivedFrom(entity1)Creating Activities
import datetime
# Create activity with time bounds
activity = doc.activity('ex:analysis',
startTime=datetime.datetime(2023, 1, 1, 10, 0, 0),
endTime=datetime.datetime(2023, 1, 1, 12, 0, 0),
other_attributes={
'prov:label': 'Data Analysis Process',
'ex:tool': 'Python'
}
)
# Activities use entities and generate new ones
activity.used(entity1)
entity2.wasGeneratedBy(activity)Creating Agents
# Create different types of agents
person = doc.agent('ex:researcher', {
'prov:type': 'prov:Person',
'foaf:name': 'Dr. Jane Smith',
'foaf:mbox': 'jane@example.org'
})
organization = doc.agent('ex:university', {
'prov:type': 'prov:Organization',
'foaf:name': 'Example University'
})
software = doc.agent('ex:script', {
'prov:type': 'prov:SoftwareAgent',
'ex:version': '1.0'
})
# Agents can be associated with activities
activity.wasAssociatedWith(person)
activity.wasAssociatedWith(software, plan=doc.entity('ex:analysisScript'))
# Agents can act on behalf of other agents
person.actedOnBehalfOf(organization)Working with Attributes
# Add attributes after creation
entity1.add_attributes({
'dc:creator': 'Data Collection System',
'dc:created': '2023-01-01T09:00:00'
})
# Get specific attributes
labels = entity1.get_attribute('prov:label')
types = entity1.get_asserted_types()
# Check record type
if entity1.is_element():
print("This is a PROV element")
print(f"Element type: {entity1.get_type()}")Install with Tessl CLI
npx tessl i tessl/pypi-prov