| | 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 |
| |
|
