heaven_base.registry.simple_registry_service

  1import os
  2import json
  3import re
  4from typing import Dict, Any, Optional, List
  5from ..utils.get_env_value import EnvConfigUtil
  6
  7
  8# Registry pointer patterns  
  9REF_KEY_PATTERN = re.compile(r'^registry_key_ref=([^:]+):(.+)$')
 10REF_OBJ_PATTERN = re.compile(r'^registry_object_ref=([^:]+):(.+)$')
 11REF_ALL_PATTERN = re.compile(r'^registry_all_ref=([^:]+)$')
 12MAX_REF_DEPTH = 99
 13
 14
 15class SimpleRegistryService:
 16    """Simplified registry service - just JSON CRUD operations with pointer resolution."""
 17    
 18    def __init__(self, registry_dir: str = None):
 19        """Initialize the SimpleRegistryService.
 20        
 21        Args:
 22            registry_dir: Directory containing registry JSON files (defaults to HEAVEN_DATA_DIR/registry)
 23        """
 24        if registry_dir is None:
 25            registry_dir = os.path.join(EnvConfigUtil.get_heaven_data_dir(), 'registry')
 26        self.registry_dir = registry_dir
 27        os.makedirs(self.registry_dir, exist_ok=True)
 28    
 29    def _get_registry_path(self, registry_name: str) -> str:
 30        """Get the file path for a registry with dual-lookup (user dir first, then library)."""
 31        # Check if name ends with _registry, if not add it
 32        if registry_name.endswith('_registry'):
 33            filename = f"{registry_name}.json"
 34        else:
 35            filename = f"{registry_name}_registry.json"
 36        
 37        # First check user's HEAVEN_DATA_DIR/registry/
 38        user_path = os.path.join(self.registry_dir, filename)
 39        if os.path.exists(user_path):
 40            return user_path
 41            
 42        # Then check library-level heaven_base/registry/
 43        try:
 44            import heaven_base
 45            library_registry_dir = os.path.join(os.path.dirname(heaven_base.__file__), 'registry')
 46            library_path = os.path.join(library_registry_dir, filename)
 47            if os.path.exists(library_path):
 48                return library_path
 49        except ImportError:
 50            pass
 51            
 52        # Default to user path (for creation/writing)
 53        return user_path
 54    
 55    def _load_registry_data(self, registry_name: str) -> Dict[str, Any]:
 56        """Load registry data from JSON file."""
 57        registry_path = self._get_registry_path(registry_name)
 58        if not os.path.exists(registry_path):
 59            return {}
 60        
 61        try:
 62            with open(registry_path, 'r') as f:
 63                return json.load(f)
 64        except (json.JSONDecodeError, IOError):
 65            return {}
 66    
 67    def _save_registry_data(self, registry_name: str, data: Dict[str, Any]) -> None:
 68        """Save registry data to JSON file."""
 69        registry_path = self._get_registry_path(registry_name)
 70        with open(registry_path, 'w') as f:
 71            json.dump(data, f, indent=2)
 72    
 73    def _resolve_if_pointer(self, value: Any, *, _seen: set = None) -> Any:
 74        """Resolve registry-pointer strings.
 75        
 76        • registry_key_ref=<registry>:<key> → returns the locator string "@<registry>/<key>"
 77        • registry_object_ref=<registry>:<key> → opens that registry, fetches the value, then continues resolving
 78        
 79        Cycles are detected; depth is capped at MAX_REF_DEPTH.
 80        """
 81        if not isinstance(value, str):
 82            return value
 83        
 84        _seen = _seen or set()
 85        ref = value.strip()
 86        
 87        if ref in _seen:
 88            raise ValueError(f"Cyclic registry reference detected: {ref}")
 89        if len(_seen) >= MAX_REF_DEPTH:
 90            raise ValueError("Max registry-reference depth exceeded")
 91        
 92        # registry_key_ref → locator string only
 93        m = REF_KEY_PATTERN.match(ref)
 94        if m:
 95            registry_name, key = m.groups()
 96            return f"@{registry_name}/{key}"
 97        
 98        # registry_object_ref → fetch value, then recurse
 99        m = REF_OBJ_PATTERN.match(ref)
100        if m:
101            registry_name, key = m.groups()
102            _seen.add(ref)
103            obj = self.get(registry_name, key)  # may itself be a pointer
104            return self._resolve_if_pointer(obj, _seen=_seen)
105        
106        # registry_all_ref → fetch entire registry contents
107        m = REF_ALL_PATTERN.match(ref)
108        if m:
109            registry_name = m.group(1)
110            _seen.add(ref)
111            all_data = self.get_all(registry_name)  # may contain pointers that get resolved
112            return all_data
113        
114        # not a recognised pointer
115        return value
116    
117    def get(self, registry_name: str, key: str) -> Optional[Any]:
118        """Get an item from a registry with pointer resolution."""
119        data = self._load_registry_data(registry_name)
120        raw_value = data.get(key)
121        return self._resolve_if_pointer(raw_value)
122    
123    def get_all(self, registry_name: str) -> Dict[str, Any]:
124        """Get all items in a registry with pointer resolution."""
125        data = self._load_registry_data(registry_name)
126        return {k: self._resolve_if_pointer(v) for k, v in data.items()}
127    
128    def get_raw(self, registry_name: str, key: str) -> Optional[Any]:
129        """Get raw value without pointer resolution (useful for editing pointers)."""
130        data = self._load_registry_data(registry_name)
131        return data.get(key)
132    
133    def get_all_raw(self, registry_name: str) -> Dict[str, Any]:
134        """Get all raw values without pointer resolution."""
135        return self._load_registry_data(registry_name)
136    
137    def set(self, registry_name: str, key: str, value: Any) -> None:
138        """Set an item in a registry."""
139        data = self._load_registry_data(registry_name)
140        data[key] = value
141        self._save_registry_data(registry_name, data)
142    
143    def delete(self, registry_name: str, key: str) -> bool:
144        """Delete an item from a registry."""
145        data = self._load_registry_data(registry_name)
146        if key not in data:
147            return False
148        
149        del data[key]
150        self._save_registry_data(registry_name, data)
151        return True
152    
153    def update(self, registry_name: str, key: str, value: Any) -> bool:
154        """Update an item with pointer validation."""
155        data = self._load_registry_data(registry_name)
156        current_raw = data.get(key)
157        
158        # Prevent updating pointers (same logic as original)
159        if isinstance(current_raw, str) and current_raw.startswith(
160            ("registry_key_ref=", "registry_object_ref=")
161        ):
162            if current_raw.startswith("registry_key_ref="):
163                _, payload = current_raw.split("=", 1)
164                target_locator = f"@{payload.replace(':', '/')}"
165            else:  # registry_object_ref
166                _, payload = current_raw.split("=", 1)
167                target_registry, target_key = payload.split(":", 1)
168                target_locator = f"@{target_registry}/{target_key}"
169            
170            raise ValueError(
171                f"Cannot update '{registry_name}:{key}'—it is a pointer to "
172                f"'{target_locator}'.\n"
173                "Update the referenced registry/key instead, or replace the "
174                "pointer string explicitly if you intend to repoint it."
175            )
176        
177        if key not in data:
178            return False
179        
180        data[key] = value
181        self._save_registry_data(registry_name, data)
182        return True
183    
184    def add(self, registry_name: str, key: str, value: Any) -> bool:
185        """Add an item to a registry (fails if key already exists)."""
186        data = self._load_registry_data(registry_name)
187        if key in data:
188            return False
189        
190        data[key] = value
191        self._save_registry_data(registry_name, data)
192        return True
193    
194    def list_keys(self, registry_name: str) -> List[str]:
195        """List all keys in a registry."""
196        data = self._load_registry_data(registry_name)
197        return list(data.keys())
198    
199    def list_registries(self) -> List[str]:
200        """List all available registries."""
201        registries = []
202        if not os.path.exists(self.registry_dir):
203            return registries
204        
205        for filename in os.listdir(self.registry_dir):
206            if filename.endswith('_registry.json'):
207                # brain_personas_registry.json -> brain_personas
208                registries.append(filename[:-14])  # Remove _registry.json extension
209            elif filename.endswith('.json'):
210                # Handle files that don't have _registry suffix
211                base_name = filename[:-5]  # Remove .json
212                if base_name.endswith('_registry'):
213                    # This is already a _registry file, extract base name
214                    registries.append(base_name[:-9])  # Remove _registry suffix
215                else:
216                    # This shouldn't happen with our naming convention, but handle it
217                    registries.append(base_name)
218        return sorted(registries)
219    
220    def create_registry(self, registry_name: str) -> bool:
221        """Create a new empty registry."""
222        registry_path = self._get_registry_path(registry_name)
223        if os.path.exists(registry_path):
224            return False
225        
226        self._save_registry_data(registry_name, {})
227        return True
228    
229    def registry_exists(self, registry_name: str) -> bool:
230        """Check if a registry exists."""
231        registry_path = self._get_registry_path(registry_name)
232        return os.path.exists(registry_path)