A library for W3C Provenance Data Model supporting PROV-JSON, PROV-XML and PROV-O (RDF)
—
PROV relationship classes that connect elements together, representing the provenance graph structure. Relationships capture how entities, activities, and agents are related through generation, usage, derivation, attribution, and influence connections.
class ProvRelation(ProvRecord):
def __init__(self, bundle, identifier, attributes):
"""
Base class for all PROV relationships.
Args:
bundle (ProvBundle): Containing bundle
identifier (QualifiedName, optional): Relation identifier
attributes (dict): Relation attributes including formal arguments
"""
def is_relation(self):
"""
Check if this is a PROV relation.
Returns:
bool: Always True for relations
"""
def is_element(self):
"""
Check if this is a PROV element.
Returns:
bool: Always False for relations
"""Relationships between activities and entities they generate or use.
class ProvGeneration(ProvRelation):
"""
Generation relationship: wasGeneratedBy(entity, activity, time, attributes).
Formal attributes:
- PROV_ATTR_ENTITY: Generated entity
- PROV_ATTR_ACTIVITY: Generating activity
- PROV_ATTR_TIME: Generation time (optional)
"""
class ProvUsage(ProvRelation):
"""
Usage relationship: used(activity, entity, time, attributes).
Formal attributes:
- PROV_ATTR_ACTIVITY: Using activity
- PROV_ATTR_ENTITY: Used entity
- PROV_ATTR_TIME: Usage time (optional)
"""
class ProvInvalidation(ProvRelation):
"""
Invalidation relationship: wasInvalidatedBy(entity, activity, time, attributes).
Formal attributes:
- PROV_ATTR_ENTITY: Invalidated entity
- PROV_ATTR_ACTIVITY: Invalidating activity
- PROV_ATTR_TIME: Invalidation time (optional)
"""Relationships between activities.
class ProvCommunication(ProvRelation):
"""
Communication relationship: wasInformedBy(informed, informant, attributes).
Formal attributes:
- PROV_ATTR_INFORMED: Informed activity
- PROV_ATTR_INFORMANT: Informing activity
"""
class ProvStart(ProvRelation):
"""
Start relationship: wasStartedBy(activity, trigger, starter, time, attributes).
Formal attributes:
- PROV_ATTR_ACTIVITY: Started activity
- PROV_ATTR_TRIGGER: Starting entity
- PROV_ATTR_STARTER: Starting activity (optional)
- PROV_ATTR_TIME: Start time (optional)
"""
class ProvEnd(ProvRelation):
"""
End relationship: wasEndedBy(activity, trigger, ender, time, attributes).
Formal attributes:
- PROV_ATTR_ACTIVITY: Ended activity
- PROV_ATTR_TRIGGER: Ending entity
- PROV_ATTR_ENDER: Ending activity (optional)
- PROV_ATTR_TIME: End time (optional)
"""Relationships showing how entities are derived from other entities.
class ProvDerivation(ProvRelation):
"""
Derivation relationship: wasDerivedFrom(generatedEntity, usedEntity, activity, generation, usage, attributes).
Formal attributes:
- PROV_ATTR_GENERATED_ENTITY: Derived entity
- PROV_ATTR_USED_ENTITY: Source entity
- PROV_ATTR_ACTIVITY: Deriving activity (optional)
- PROV_ATTR_GENERATION: Generation record (optional)
- PROV_ATTR_USAGE: Usage record (optional)
"""Relationships connecting entities and activities to responsible agents.
class ProvAttribution(ProvRelation):
"""
Attribution relationship: wasAttributedTo(entity, agent, attributes).
Formal attributes:
- PROV_ATTR_ENTITY: Attributed entity
- PROV_ATTR_AGENT: Responsible agent
"""
class ProvAssociation(ProvRelation):
"""
Association relationship: wasAssociatedWith(activity, agent, plan, attributes).
Formal attributes:
- PROV_ATTR_ACTIVITY: Associated activity
- PROV_ATTR_AGENT: Associated agent
- PROV_ATTR_PLAN: Plan entity (optional)
"""
class ProvDelegation(ProvRelation):
"""
Delegation relationship: actedOnBehalfOf(delegate, responsible, activity, attributes).
Formal attributes:
- PROV_ATTR_DELEGATE: Delegated agent
- PROV_ATTR_RESPONSIBLE: Responsible agent
- PROV_ATTR_ACTIVITY: Delegating activity (optional)
"""General influence and specialized entity relationships.
class ProvInfluence(ProvRelation):
"""
Influence relationship: wasInfluencedBy(influencee, influencer, attributes).
Formal attributes:
- PROV_ATTR_INFLUENCEE: Influenced element
- PROV_ATTR_INFLUENCER: Influencing element
"""
class ProvSpecialization(ProvRelation):
"""
Specialization relationship: specializationOf(specificEntity, generalEntity, attributes).
Formal attributes:
- PROV_ATTR_SPECIFIC_ENTITY: Specific entity
- PROV_ATTR_GENERAL_ENTITY: General entity
"""
class ProvAlternate(ProvRelation):
"""
Alternate relationship: alternateOf(alternate1, alternate2, attributes).
Formal attributes:
- PROV_ATTR_ALTERNATE1: First alternate entity
- PROV_ATTR_ALTERNATE2: Second alternate entity
"""
class ProvMention(ProvSpecialization):
"""
Mention relationship: mentionOf(specificEntity, generalEntity, bundle, attributes).
Formal attributes:
- PROV_ATTR_SPECIFIC_ENTITY: Specific entity
- PROV_ATTR_GENERAL_ENTITY: General entity
- PROV_ATTR_BUNDLE: Contextual bundle
"""Relationships for collection entities and their members.
class ProvMembership(ProvRelation):
"""
Membership relationship: hadMember(collection, entity, attributes).
Formal attributes:
- PROV_ATTR_COLLECTION: Collection entity
- PROV_ATTR_ENTITY: Member entity
"""Bundles provide convenient methods for creating relationships directly:
class ProvBundle:
# Generation relationships
def generation(self, entity, activity, time=None, identifier=None, other_attributes=None):
"""Create a generation relationship."""
def usage(self, activity, entity, time=None, identifier=None, other_attributes=None):
"""Create a usage relationship."""
def invalidation(self, entity, activity, time=None, identifier=None, other_attributes=None):
"""Create an invalidation relationship."""
# Activity relationships
def communication(self, informed, informant, identifier=None, other_attributes=None):
"""Create a communication relationship."""
def start(self, activity, trigger, starter=None, time=None, identifier=None, other_attributes=None):
"""Create a start relationship."""
def end(self, activity, trigger, ender=None, time=None, identifier=None, other_attributes=None):
"""Create an end relationship."""
# Derivation relationships
def derivation(self, generatedEntity, usedEntity, activity=None, generation=None, usage=None, identifier=None, other_attributes=None):
"""Create a derivation relationship."""
def revision(self, generatedEntity, usedEntity, activity=None, generation=None, usage=None, identifier=None, other_attributes=None):
"""Create a revision derivation relationship."""
def quotation(self, generatedEntity, usedEntity, activity=None, generation=None, usage=None, identifier=None, other_attributes=None):
"""Create a quotation derivation relationship."""
def primary_source(self, generatedEntity, usedEntity, activity=None, generation=None, usage=None, identifier=None, other_attributes=None):
"""Create a primary source derivation relationship."""
# Attribution relationships
def attribution(self, entity, agent, identifier=None, other_attributes=None):
"""Create an attribution relationship."""
def association(self, activity, agent, plan=None, identifier=None, other_attributes=None):
"""Create an association relationship."""
def delegation(self, delegate, responsible, activity=None, identifier=None, other_attributes=None):
"""Create a delegation relationship."""
# Influence and entity relationships
def influence(self, influencee, influencer, identifier=None, other_attributes=None):
"""Create an influence relationship."""
def specialization(self, specificEntity, generalEntity, identifier=None, other_attributes=None):
"""Create a specialization relationship."""
def alternate(self, alternate1, alternate2, identifier=None, other_attributes=None):
"""Create an alternate relationship."""
def mention(self, specificEntity, generalEntity, bundle, identifier=None, other_attributes=None):
"""Create a mention relationship."""
# Collection relationships
def membership(self, collection, entity, identifier=None, other_attributes=None):
"""Create a membership relationship."""from prov.model import ProvDocument
doc = ProvDocument()
# Create elements
dataset = doc.entity('ex:dataset')
analysis = doc.activity('ex:analysis')
result = doc.entity('ex:result')
researcher = doc.agent('ex:researcher')
# Create relationships through element methods
analysis.used(dataset, time='2023-01-01T10:00:00')
result.wasGeneratedBy(analysis, time='2023-01-01T11:00:00')
result.wasAttributedTo(researcher)
analysis.wasAssociatedWith(researcher)# Same relationships using bundle methods
doc.usage(analysis, dataset, time='2023-01-01T10:00:00')
doc.generation(result, analysis, time='2023-01-01T11:00:00')
doc.attribution(result, researcher)
doc.association(analysis, researcher)# Simple derivation
processed_data = doc.entity('ex:processedData')
processed_data.wasDerivedFrom(dataset)
# Detailed derivation with activity
doc.derivation(
generatedEntity=processed_data,
usedEntity=dataset,
activity=analysis,
other_attributes={'ex:method': 'statistical_analysis'}
)
# Specialized derivations
revised_data = doc.entity('ex:revisedData')
doc.revision(revised_data, processed_data, analysis)
quoted_data = doc.entity('ex:quotedData')
doc.quotation(quoted_data, dataset)# Specialization and alternates
specific_dataset = doc.entity('ex:specificDataset')
general_dataset = doc.entity('ex:generalDataset')
alternate_dataset = doc.entity('ex:alternateDataset')
doc.specialization(specific_dataset, general_dataset)
doc.alternate(dataset, alternate_dataset)
# Collection membership
collection = doc.entity('ex:dataCollection', {'prov:type': 'prov:Collection'})
doc.membership(collection, dataset)
doc.membership(collection, processed_data)# Delegation
supervisor = doc.agent('ex:supervisor')
student = doc.agent('ex:student')
doc.delegation(student, supervisor, analysis)
# Multiple associations
tool = doc.agent('ex:analysisTool', {'prov:type': 'prov:SoftwareAgent'})
plan = doc.entity('ex:analysisPlan')
doc.association(analysis, researcher)
doc.association(analysis, tool, plan)# Activity communication
data_collection = doc.activity('ex:dataCollection')
data_analysis = doc.activity('ex:dataAnalysis')
doc.communication(data_analysis, data_collection)
# Activity start/end
trigger_event = doc.entity('ex:triggerEvent')
doc.start(analysis, trigger_event, data_collection)
doc.end(analysis, result, data_analysis)# Add custom attributes to relationships
usage_rel = doc.usage(analysis, dataset,
other_attributes={
'ex:accessMethod': 'database_query',
'ex:queryTime': '0.5s'
})
# Access relationship attributes
attrs = usage_rel.attributes
formal_attrs = usage_rel.formal_attributes
extra_attrs = usage_rel.extra_attributes
print(f"Used entity: {usage_rel.get_attribute('prov:entity')}")
print(f"Using activity: {usage_rel.get_attribute('prov:activity')}")Install with Tessl CLI
npx tessl i tessl/pypi-prov