42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
from typing import Dict, List, Any, Optional
|
|
|
|
|
|
class ContractRegistry:
|
|
"""
|
|
Lightweight, versioned contract registry for CatOpt-Graph MVP.
|
|
Stores schema definitions per contract name and version.
|
|
Provides a simple conformance check against adapter-provided data.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
# Structure: { name: { version: schema_dict } }
|
|
self._registry: Dict[str, Dict[str, Dict[str, Any]]] = {}
|
|
|
|
def register_contract(self, name: str, version: str, schema: Dict[str, Any]) -> None:
|
|
if name not in self._registry:
|
|
self._registry[name] = {}
|
|
self._registry[name][version] = schema
|
|
|
|
def get_contract(self, name: str, version: str) -> Optional[Dict[str, Any]]:
|
|
return self._registry.get(name, {}).get(version)
|
|
|
|
def list_versions(self, name: str) -> List[str]:
|
|
return list(self._registry.get(name, {}).keys())
|
|
|
|
def conformance_check(self, name: str, version: str, adapter_data: Dict[str, Any]) -> bool:
|
|
"""
|
|
Very lightweight conformance check:
|
|
- The contract schema defines required_fields.
|
|
- adapter_data must contain all required fields at top level.
|
|
This is a stub to be extended by real conformance tests.
|
|
"""
|
|
contract = self.get_contract(name, version)
|
|
if contract is None:
|
|
return False
|
|
required = contract.get("required_fields", [])
|
|
# If required_fields not provided, assume no conformance requirement
|
|
for field in required:
|
|
if field not in adapter_data:
|
|
return False
|
|
return True
|