69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
from typing import Dict, Any, List, Optional
|
|
from datetime import datetime
|
|
|
|
# Lightweight fallback if pydantic is not installed in the test environment.
|
|
try:
|
|
from pydantic import BaseModel, Field
|
|
except Exception:
|
|
class Field:
|
|
def __init__(self, default=None, default_factory=None):
|
|
self.default = default
|
|
self.default_factory = default_factory
|
|
|
|
class BaseModel:
|
|
def __init__(self, **data):
|
|
# populate annotated fields with provided values or defaults
|
|
for name, _typ in getattr(self, "__annotations__", {}).items():
|
|
if name in data:
|
|
setattr(self, name, data[name])
|
|
continue
|
|
|
|
val = getattr(self.__class__, name, None)
|
|
if isinstance(val, Field):
|
|
if val.default_factory is not None:
|
|
setattr(self, name, val.default_factory())
|
|
else:
|
|
setattr(self, name, val.default)
|
|
else:
|
|
setattr(self, name, val)
|
|
|
|
def dict(self):
|
|
return {k: getattr(self, k) for k in getattr(self, "__annotations__", {})}
|
|
|
|
|
|
class LocalPlan(BaseModel):
|
|
id: str
|
|
service: str
|
|
pipeline: Dict[str, Any]
|
|
canary: Optional[Dict[str, Any]] = None
|
|
metadata: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class SharedSignals(BaseModel):
|
|
timestamp: datetime = Field(default_factory=datetime.utcnow)
|
|
service: str
|
|
metrics: Dict[str, float] = Field(default_factory=dict)
|
|
# exposure controls: per-signal metadata (allowlist, retention seconds)
|
|
exposure: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
|
|
|
|
|
|
class PlanDelta(BaseModel):
|
|
id: str
|
|
plan_id: str
|
|
delta: Dict[str, Any]
|
|
author_id: str
|
|
timestamp: datetime = Field(default_factory=datetime.utcnow)
|
|
signature: Optional[str] = None # placeholder for signed deltas
|
|
|
|
|
|
class AuditEntry(BaseModel):
|
|
id: str
|
|
event: str
|
|
actor: str
|
|
detail: Dict[str, Any] = Field(default_factory=dict)
|
|
timestamp: datetime = Field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class AuditLog(BaseModel):
|
|
entries: List[AuditEntry] = Field(default_factory=list)
|