generated from coulomb/repo-seed
109 lines
3.5 KiB
Python
109 lines
3.5 KiB
Python
"""Context entity and typed relationship primitives."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from enum import Enum
|
|
from typing import Any
|
|
|
|
from .primitives import compact_dict, new_id, utc_now
|
|
|
|
|
|
class ContextEntityType(str, Enum):
|
|
PERSON = "person"
|
|
TEAM = "team"
|
|
PROJECT = "project"
|
|
CASE = "case"
|
|
CUSTOMER = "customer"
|
|
PRODUCT = "product"
|
|
PROCESS = "process"
|
|
SOURCE_SYSTEM = "source_system"
|
|
TOPIC = "topic"
|
|
BUSINESS_OBJECT = "business_object"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ContextEntity:
|
|
entity_type: ContextEntityType
|
|
name: str
|
|
external_ref: str | None = None
|
|
metadata: dict[str, Any] = field(default_factory=dict)
|
|
entity_id: str = field(default_factory=lambda: new_id("entity"))
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return compact_dict(
|
|
{
|
|
"entity_id": self.entity_id,
|
|
"entity_type": self.entity_type.value,
|
|
"name": self.name,
|
|
"external_ref": self.external_ref,
|
|
"metadata": dict(self.metadata),
|
|
}
|
|
)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "ContextEntity":
|
|
return cls(
|
|
entity_id=data["entity_id"],
|
|
entity_type=ContextEntityType(data["entity_type"]),
|
|
name=data["name"],
|
|
external_ref=data.get("external_ref"),
|
|
metadata=dict(data.get("metadata", {})),
|
|
)
|
|
|
|
|
|
class RelationshipTargetKind(str, Enum):
|
|
ASSET = "asset"
|
|
CONTEXT_ENTITY = "context_entity"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class CoreRelationship:
|
|
source_id: str
|
|
target_id: str
|
|
predicate: str
|
|
target_kind: RelationshipTargetKind = RelationshipTargetKind.ASSET
|
|
direction: str = "outbound"
|
|
confidence: float | None = None
|
|
valid_from: str | None = None
|
|
valid_to: str | None = None
|
|
actor_id: str | None = None
|
|
provenance: dict[str, Any] = field(default_factory=dict)
|
|
relationship_id: str = field(default_factory=lambda: new_id("rel"))
|
|
created_at: str = field(default_factory=lambda: utc_now().isoformat())
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return compact_dict(
|
|
{
|
|
"relationship_id": self.relationship_id,
|
|
"source_id": self.source_id,
|
|
"target_id": self.target_id,
|
|
"predicate": self.predicate,
|
|
"target_kind": self.target_kind.value,
|
|
"direction": self.direction,
|
|
"confidence": self.confidence,
|
|
"valid_from": self.valid_from,
|
|
"valid_to": self.valid_to,
|
|
"actor_id": self.actor_id,
|
|
"provenance": dict(self.provenance),
|
|
"created_at": self.created_at,
|
|
}
|
|
)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict[str, Any]) -> "CoreRelationship":
|
|
return cls(
|
|
relationship_id=data["relationship_id"],
|
|
source_id=data["source_id"],
|
|
target_id=data["target_id"],
|
|
predicate=data["predicate"],
|
|
target_kind=RelationshipTargetKind(data.get("target_kind", RelationshipTargetKind.ASSET.value)),
|
|
direction=data.get("direction", "outbound"),
|
|
confidence=data.get("confidence"),
|
|
valid_from=data.get("valid_from"),
|
|
valid_to=data.get("valid_to"),
|
|
actor_id=data.get("actor_id"),
|
|
provenance=dict(data.get("provenance", {})),
|
|
created_at=data["created_at"],
|
|
)
|