56 lines
2.2 KiB
Python
56 lines
2.2 KiB
Python
"""In-memory registry and minimal API for CatOpt Studio MVP.
|
|
|
|
This module provides a tiny, process-local registry suitable for tests and a
|
|
conformance harness. It exposes the minimal operations requested in the idea
|
|
description: register_contract, push_signal, propose_delta, get_provenance.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime
|
|
from typing import Dict, Any, List, Optional
|
|
|
|
|
|
@dataclass
|
|
class RegistryEntry:
|
|
contract_id: str
|
|
adapter: str
|
|
schema: Dict[str, Any]
|
|
created: datetime = field(default_factory=datetime.utcnow)
|
|
|
|
|
|
class InMemoryRegistry:
|
|
def __init__(self):
|
|
self._contracts: Dict[str, RegistryEntry] = {}
|
|
self._signals: List[Dict[str, Any]] = []
|
|
self._deltas: List[Dict[str, Any]] = []
|
|
|
|
def register_contract(self, contract_id: str, adapter: str, schema: Dict[str, Any]) -> RegistryEntry:
|
|
entry = RegistryEntry(contract_id=contract_id, adapter=adapter, schema=schema)
|
|
self._contracts[contract_id] = entry
|
|
return entry
|
|
|
|
def push_signal(self, contract_id: str, signal: Dict[str, Any]) -> None:
|
|
record = {"contract_id": contract_id, "signal": signal, "timestamp": datetime.utcnow()}
|
|
self._signals.append(record)
|
|
|
|
def propose_delta(self, contract_id: str, delta: Dict[str, Any], author: Optional[str] = None) -> None:
|
|
record = {"contract_id": contract_id, "delta": delta, "author": author or "anonymous", "timestamp": datetime.utcnow()}
|
|
self._deltas.append(record)
|
|
|
|
def get_provenance(self, contract_id: str) -> Dict[str, Any]:
|
|
# Return a tiny provenance summary for the given contract
|
|
signals = [s for s in self._signals if s["contract_id"] == contract_id]
|
|
deltas = [d for d in self._deltas if d["contract_id"] == contract_id]
|
|
entry = self._contracts.get(contract_id)
|
|
return {
|
|
"contract_id": contract_id,
|
|
"registered": entry.created if entry else None,
|
|
"signals_count": len(signals),
|
|
"deltas_count": len(deltas),
|
|
}
|
|
|
|
|
|
# Module-level default registry for convenience in tests and examples
|
|
default_registry = InMemoryRegistry()
|