from __future__ import annotations
from typing import TYPE_CHECKING, Callable
if TYPE_CHECKING:
from .base import Store
from .memory import MemoryStore
__all__ = ("StoreRegistry",)
def default_default_factory(name: str) -> Store:
return MemoryStore()
[docs]
class StoreRegistry:
"""Registry for :class:`Store <.base.Store>` instances."""
__slots__ = ("_default_factory", "_stores")
[docs]
def __init__(
self, stores: dict[str, Store] | None = None, default_factory: Callable[[str], Store] = default_default_factory
) -> None:
"""Initialize ``StoreRegistry``.
Args:
stores: A dictionary mapping store names to stores, used to initialize the registry
default_factory: A callable used by :meth:`StoreRegistry.get` to provide a store, if the requested name hasn't
been registered yet. This callable receives the requested name and should return a
:class:`Store <.base.Store>` instance.
"""
self._stores = stores or {}
self._default_factory = default_factory
[docs]
def register(self, name: str, store: Store, allow_override: bool = False) -> None:
"""Register a new :class:`Store <.base.Store>`.
Args:
name: Name to register the store under
store: The store to register
allow_override: Whether to allow overriding an existing store of the same name
Raises:
ValueError: If a store is already registered under this name and ``override`` is not ``True``
"""
if not allow_override and name in self._stores:
raise ValueError(f"Store with the name {name!r} already exists")
self._stores[name] = store
[docs]
def get(self, name: str) -> Store:
"""Get a store registered under ``name``. If no such store is registered, create a store using the default
factory with ``name`` and register the returned store under ``name``.
Args:
name: Name of the store
Returns:
A :class:`Store <.base.Store>`
"""
if not self._stores.get(name):
self._stores[name] = self._default_factory(name)
return self._stores[name]