-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathgraph.py
More file actions
135 lines (105 loc) · 4.11 KB
/
Copy pathgraph.py
File metadata and controls
135 lines (105 loc) · 4.11 KB
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
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
"""Portable hawk-eco graph wire models.
The SDK consumes these data-only projections. Source repositories retain
ownership of graph facts, storage, and runtime behavior.
"""
from __future__ import annotations
from datetime import datetime # noqa: TC003
from enum import Enum
from pydantic import BaseModel, ConfigDict, Field, model_validator
class GraphNodeKind(str, Enum):
SYSTEM = "system"
KNOWLEDGE = "knowledge"
EXECUTION = "execution"
POLICY = "policy"
QUALITY = "quality"
OPERATIONS = "operations"
class GraphEdgeKind(str, Enum):
CONTAINS = "contains"
DEPENDS_ON = "depends_on"
REFERENCES = "references"
PRODUCED = "produced"
GOVERNED_BY = "governed_by"
VALIDATED_BY = "validated_by"
class GraphEventType(str, Enum):
CREATED = "created"
UPDATED = "updated"
TRANSITIONED = "transitioned"
OBSERVED = "observed"
DELETED = "deleted"
class GraphModel(BaseModel):
model_config = ConfigDict(extra="forbid")
class GraphScope(GraphModel):
tenant_id: str | None = None
project_id: str | None = None
repository_id: str | None = None
class GraphRef(GraphModel):
kind: GraphNodeKind
id: str = Field(min_length=1, max_length=256)
class GraphArtifactRef(GraphModel):
uri: str = Field(min_length=1, max_length=2048)
digest: str | None = Field(None, max_length=256)
media_type: str | None = Field(None, max_length=128)
class GraphProvenance(GraphModel):
producer: str = Field(min_length=1, max_length=100)
version: str | None = Field(None, max_length=100)
source_id: str | None = Field(None, max_length=256)
evidence: list[GraphArtifactRef] = Field(default_factory=list, max_length=16)
class GraphNode(GraphModel):
id: str = Field(min_length=1, max_length=256)
kind: GraphNodeKind
scope: GraphScope | None = None
created_at: datetime
effective_at: datetime | None = None
provenance: GraphProvenance
attributes: dict[str, str] = Field(default_factory=dict)
class GraphEdge(GraphModel):
id: str = Field(min_length=1, max_length=256)
kind: GraphEdgeKind
from_: GraphRef = Field(alias="from")
to: GraphRef
scope: GraphScope | None = None
created_at: datetime
effective_at: datetime | None = None
provenance: GraphProvenance
attributes: dict[str, str] = Field(default_factory=dict)
model_config = ConfigDict(extra="forbid", populate_by_name=True)
class GraphEvent(GraphModel):
id: str = Field(min_length=1, max_length=256)
type: GraphEventType
subject: GraphRef
scope: GraphScope | None = None
occurred_at: datetime
correlation_id: str | None = Field(None, max_length=256)
causation_id: str | None = Field(None, max_length=256)
idempotency_key: str | None = Field(None, max_length=256)
provenance: GraphProvenance
class GraphExport(GraphModel):
schema_version: str = Field(pattern=r"^[a-z0-9-]+\.graph/v1$")
generated_at: datetime
query_sha256: str | None = Field(None, pattern=r"^[a-f0-9]{64}$")
scope: GraphScope | None = None
nodes: list[GraphNode]
edges: list[GraphEdge]
events: list[GraphEvent]
@model_validator(mode="after")
def validate_topology(self) -> GraphExport:
nodes: dict[str, GraphNodeKind] = {}
for node in self.nodes:
if node.id in nodes:
raise ValueError(f"duplicate graph node {node.id}")
nodes[node.id] = node.kind
edge_ids: set[str] = set()
for edge in self.edges:
if edge.id in edge_ids:
raise ValueError(f"duplicate graph edge {edge.id}")
edge_ids.add(edge.id)
if nodes.get(edge.from_.id) != edge.from_.kind or nodes.get(edge.to.id) != edge.to.kind:
raise ValueError(f"dangling graph edge {edge.id}")
event_ids: set[str] = set()
for event in self.events:
if event.id in event_ids:
raise ValueError(f"duplicate graph event {event.id}")
event_ids.add(event.id)
if nodes.get(event.subject.id) != event.subject.kind:
raise ValueError(f"dangling graph event {event.id}")
return self