-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_client.py
47 lines (40 loc) · 1.65 KB
/
test_client.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
from datetime import datetime, timezone
from provenance.client import ProvenanceClient
from provenance.client import ProvType, ProvVerb
def main():
# Initialize client - adjust URL to match your server
with ProvenanceClient("http://localhost:8000/api") as client:
# Create test nodes
def apply_label_prefix(label):
return f"test1-{label}"
dataset = client.create_node(
label=apply_label_prefix("test-dataset-001"),
node_type=ProvType.ENTITY,
description="Test dataset",
metadata={"format": "csv", "size": 1024}
)
print(f"Created dataset node: {dataset.label}")
process = client.create_node(
label=apply_label_prefix("test-process-001"),
node_type=ProvType.ACTIVITY,
description="Test analysis process",
metadata={"version": "1.0"}
)
print(f"Created process node: {process.label}")
# Create relation
run_id = apply_label_prefix("test-run-001")
relation = client.create_relation(
subject_label=dataset.label,
verb=ProvVerb.WAS_GENERATED_BY,
object_label=process.label,
run_id=run_id,
start_time=datetime.now(timezone.utc)
)
print(f"Created relation: {relation.subject_label} {relation.verb} {relation.object_label}")
# Fetch relations
relations = client.get_relations(run_id)
print(f"\nFound {len(relations)} relations for run ${run_id}:")
for r in relations:
print(f"- {r.subject_label} {r.verb} {r.object_label}")
if __name__ == "__main__":
main()