import glob from abc import ABC, abstractmethod from datetime import datetime, timedelta from typing import Dict, Generic, List, Optional, Tuple, TypeVar component_type = TypeVar("component_type") class ComponentRegistry(ABC, Generic[component_type]): def __init__(self, file_glob: str, ttl: Optional[timedelta] = None) -> None: super().__init__() self._registry: Optional[Dict[str, component_type]] = None self._registry_update: datetime = datetime.fromtimestamp(0) self._file_glob: str = file_glob self._ttl: Optional[timedelta] = ttl @abstractmethod def _load_component(self, path: str) -> Tuple[str, component_type]: raise NotImplementedError def is_available(self, freshness: Optional[timedelta] = None) -> bool: if self._registry is None: return False staleness = datetime.now() - self._registry_update if self._ttl is not None and staleness > self._ttl: return False if freshness is not None and staleness > freshness: return False return True def get_registry( self, force_reload: bool = False, freshness: Optional[timedelta] = None, show_error: bool = False, ) -> Dict[str, component_type]: if not force_reload and self.is_available(freshness): assert self._registry is not None return self._registry registry: Dict[str, component_type] = {} for path in glob.glob(self._file_glob): try: name, component = self._load_component(path) except Exception as e: if show_error: print(f"failed to loading component from {path}, skipping: {e}") continue if component is None: if show_error: print(f"failed to loading component from {path}, skipping") continue registry[name] = component self._registry_update = datetime.now() self._registry = registry return registry @property def registry(self) -> Dict[str, component_type]: return self.get_registry() def get_list(self, force_reload: bool = False, freshness: Optional[timedelta] = None) -> List[component_type]: registry = self.get_registry(force_reload, freshness) keys = sorted(registry.keys()) return [registry[k] for k in keys] @property def list(self) -> List[component_type]: return self.get_list() def get(self, name: str) -> Optional[component_type]: return self.registry.get(name, None) def __getitem__(self, name: str) -> Optional[component_type]: return self.get(name) @property def file_glob(self) -> str: return self._file_glob @file_glob.setter def file_glob(self, file_glob: str) -> None: if self._file_glob == file_glob: return self._file_glob = file_glob self._registry = None