feat(registry): añade FileSystemPolicyStore con carga YAML versionada
Incluye 'yaml' en los overrides de mypy (PyYAML no trae type stubs). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,64 @@
|
||||
"""PolicyStore basado en sistema de ficheros (YAML por versión + index.yaml)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
|
||||
from agentforge_core.domain.policy import PolicyDefinition, PolicyVersionMeta
|
||||
|
||||
|
||||
class FileSystemPolicyStore:
|
||||
"""Lee políticas de policies/<name>/{index.yaml,versions/vN.yaml}."""
|
||||
|
||||
def __init__(self, root: Path) -> None:
|
||||
self._root = Path(root)
|
||||
|
||||
def list_policies(self) -> list[PolicyDefinition]:
|
||||
if not self._root.exists():
|
||||
return []
|
||||
result: list[PolicyDefinition] = []
|
||||
for d in sorted(self._root.iterdir()):
|
||||
if d.is_dir() and (d / "index.yaml").exists():
|
||||
result.append(self.get_policy(d.name))
|
||||
return result
|
||||
|
||||
def get_policy(self, name: str, version: str | None = None) -> PolicyDefinition:
|
||||
index = self._load_index(name)
|
||||
v = version or index["active_version"]
|
||||
path = self._root / name / "versions" / f"{v}.yaml"
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"policy version not found: {name}@{v}")
|
||||
with path.open(encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f)
|
||||
return PolicyDefinition.model_validate(data)
|
||||
|
||||
def list_versions(self, name: str) -> list[PolicyVersionMeta]:
|
||||
index = self._load_index(name)
|
||||
return [
|
||||
PolicyVersionMeta(
|
||||
id=v["id"],
|
||||
hash=v["hash"],
|
||||
author=v["author"],
|
||||
message=v["message"],
|
||||
created_at=self._parse_dt(v["created_at"]),
|
||||
)
|
||||
for v in index["versions"]
|
||||
]
|
||||
|
||||
def _load_index(self, name: str) -> dict[str, Any]:
|
||||
index_path = self._root / name / "index.yaml"
|
||||
if not index_path.exists():
|
||||
raise FileNotFoundError(f"policy not found: {name}")
|
||||
with index_path.open(encoding="utf-8") as f:
|
||||
data: dict[str, Any] = yaml.safe_load(f)
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
def _parse_dt(value: str | datetime) -> datetime:
|
||||
if isinstance(value, datetime):
|
||||
return value
|
||||
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
Reference in New Issue
Block a user