import threading
from abc import ABC, abstractmethod
from typing import Generic, TypeVar
from collections.abc import Callable
T = TypeVar('T')
[docs]
class AbstractRegistry(ABC, Generic[T]):
"""
Base class for instance registries. This class stores instances in a dictionary under given keys. When an instance
is first requested, it will be created using the :py:meth:`_create()` method and stored in the internal dictionary
with the given key.
On subsequent requests with the same key, the same instance will be returned.
"""
def __init__(self):
self._instances: dict[str, T] = {}
def get(self, key: str) -> T:
if key not in self._instances:
self._instances[key] = self._create(key)
return self._instances[key]
def delete(self, key: str):
if key in self._instances:
del self._instances[key]
@abstractmethod
def _create(self, key: str) -> T:
raise NotImplementedError('The _create method must be implemented in subclasses.')
[docs]
def create_registry_getter(type_: type[T], registry_factory: Callable[[], AbstractRegistry[T]]) -> Callable[[str], T]:
"""
Registry factory. Creates a separate registry for each thread.
:param type_: Type of the registry.
:param registry_factory: Factory function that will be called to create a registry for each thread.
:return: Registry of given type.
"""
registry_tls = threading.local()
def get(name: str) -> T:
nonlocal registry_tls
if hasattr(registry_tls, 'registry'):
registry = getattr(registry_tls, 'registry')
else:
registry = registry_factory()
setattr(registry_tls, 'registry', registry)
return registry.get(name)
return get