Registry
coola.registry ¶
General-purpose registry primitives used across the package.
Registry is a thread-safe key-value container and TypeRegistry
adds the lookup of the most specific registered type (Method Resolution
Order). Both are generic, keep the insertion order, return snapshots
from items, keys and values, raise
AlreadyRegisteredError on a duplicate unless exist_ok=True, copy
independently with copy, survive os.fork and load the entries of
optional libraries lazily (register_lazy). See the registry user
guide for the details.
coola.registry.AlreadyRegisteredError ¶
Bases: ValueError
Raised when registering a key that is already registered and
exist_ok is False.
It is a ValueError (the value passed is not acceptable in the
current state of the registry).
Example
>>> from coola.registry import AlreadyRegisteredError, Registry
>>> registry = Registry[str, int]()
>>> registry.register("key", 1)
>>> try:
... registry.register("key", 2)
... except AlreadyRegisteredError:
... print("already registered")
...
already registered
coola.registry.KeyNotRegisteredError ¶
Bases: KeyError
Raised when no value is registered for a key.
It subclasses KeyError (a LookupError). Registries whose
keys are types raise the more specific TypeNotRegisteredError.
Example
>>> from coola.registry import KeyNotRegisteredError, Registry
>>> registry = Registry[str, int]()
>>> try:
... registry["missing"]
... except KeyNotRegisteredError:
... print("not registered")
...
not registered
coola.registry.Registry ¶
Bases: Generic[K, V]
A thread-safe generic key-value registry for storing and managing typed mappings.
The Registry class provides a type-safe container for registering and retrieving values by key. It supports all standard dictionary operations through operator overloading and provides additional methods for safe registration and querying.
The state is a dictionary that is replaced on every write and never
mutated, so the reads take no lock, items(), keys() and
values() return snapshots, and a writer never shows a partial
state to a reader. The writers are serialized by one reentrant lock.
TypeRegistry extends this class.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
initial_state
|
dict[K, V] | None
|
An optional dictionary to initialize the registry with. If provided, a copy is made to prevent external modifications. Defaults to None, which creates an empty registry. |
None
|
Attributes:
| Name | Type | Description |
|---|---|---|
_state |
dict[K, V]
|
Internal dictionary storing the key-value pairs. It is replaced, never mutated. |
_lock |
RLock
|
Threading lock for synchronizing the writers. |
Raises:
| Type | Description |
|---|---|
TypeError
|
If a key is not valid for the registry. |
Example
Basic usage with registration and retrieval:
>>> from coola.registry import Registry
>>> registry = Registry[str, int]()
>>> registry.register("key1", 42)
>>> registry.get("key1")
42
>>> registry
Registry(
(key1): 42
)
Using dictionary-style operations:
>>> from coola.registry import Registry
>>> registry = Registry[str, int]()
>>> registry["key2"] = 100
>>> "key2" in registry
True
>>> del registry["key2"]
Initializing with existing data:
>>> from coola.registry import Registry
>>> registry = Registry[str, int](initial_state={"a": 1, "b": 2})
>>> len(registry)
2
coola.registry.Registry.clear ¶
clear() -> None
Remove all entries from the registry.
This method empties the registry, leaving it in the same state as a newly created empty registry. This operation cannot be undone.
Example
>>> from coola.registry import Registry
>>> registry = Registry[str, int]({"key1": 42, "key2": 100})
>>> len(registry)
2
>>> registry.clear()
>>> len(registry)
0
>>> registry.has("key1")
False
coola.registry.Registry.copy ¶
copy() -> Self
Create a copy of the registry.
The copy has the same entries and the same pending lazy groups as the registry, and is independent from it: registering or removing an entry in one registry does not change the other. The keys and the values are not copied.
Returns:
| Type | Description |
|---|---|
Self
|
A new registry of the same class. |
Example
>>> from coola.registry import Registry
>>> registry = Registry[str, int]({"key": 42})
>>> other = registry.copy()
>>> other.register("other", 1)
>>> registry.has("other")
False
>>> other.get("key")
42
coola.registry.Registry.equal ¶
equal(other: object) -> bool
Indicate if two objects are equal or not.
Two registries are equal if they have the same class and the
same entries. The values are compared with ==, so a value
without __eq__ is equal to itself only. A registry is equal
to its copy. The pending lazy groups are loaded before the
comparison, except if the two registries have the same pending
groups.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
object
|
The object to compare with. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from coola.registry import Registry
>>> registry1 = Registry[str, int]({"key1": 42, "key2": 100})
>>> registry2 = Registry[str, int]({"key1": 42, "key2": 100})
>>> registry3 = Registry[str, int]({"key1": 42})
>>> registry1.equal(registry2)
True
>>> registry1.equal(registry3)
False
coola.registry.Registry.get ¶
get(key: K, default: V | None = None) -> V | None
Retrieve the value associated with a key.
This method performs a lookup in the registry and returns the corresponding value.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
K
|
The key whose value should be retrieved. |
required |
default
|
V | None
|
Value to return if the key does not exist. |
None
|
Returns:
| Type | Description |
|---|---|
V | None
|
The value associated with the specified key. |
Example
>>> from coola.registry import Registry
>>> registry = Registry[str, int]({"key1": 42, "key2": 100})
>>> registry.get("key1")
42
>>> registry.get("missing") is None
True
coola.registry.Registry.has ¶
has(key: K) -> bool
Check whether a key is registered in the registry.
This method provides a safe way to test for key existence without risking a KeyError exception.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
K
|
The key to check for existence. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True if the key exists in the registry, False otherwise. |
Example
>>> from coola.registry import Registry
>>> registry = Registry[str, int]({"key1": 42, "key2": 100})
>>> registry.has("key1")
True
>>> registry.has("missing")
False
coola.registry.Registry.items ¶
items() -> ItemsView[K, V]
Return key-value pairs.
Returns:
| Type | Description |
|---|---|
ItemsView[K, V]
|
The key-value pairs. |
Example
>>> from coola.registry import Registry
>>> registry = Registry[str, int](initial_state={"a": 1, "b": 2})
>>> registry.items()
dict_items([('a', 1), ('b', 2)])
coola.registry.Registry.keys ¶
keys() -> KeysView[K]
Return registered keys.
Returns:
| Type | Description |
|---|---|
KeysView[K]
|
The registered keys. |
Example
>>> from coola.registry import Registry
>>> registry = Registry[str, int](initial_state={"a": 1, "b": 2})
>>> registry.keys()
dict_keys(['a', 'b'])
coola.registry.Registry.register ¶
register(
key: K, value: V, *, exist_ok: bool = False
) -> None
Register a new key-value pair in the registry.
By default, this method raises an error if you try to register a key that already exists. This prevents accidental overwriting of values. Set exist_ok=True to allow overwriting.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
K
|
The key to register. Must be hashable. |
required |
value
|
V
|
The value to associate with the key. |
required |
exist_ok
|
bool
|
Controls behavior when the key already exists. If False (default), raises AlreadyRegisteredError for duplicate keys. If True, silently overwrites the existing value. |
False
|
Raises:
| Type | Description |
|---|---|
AlreadyRegisteredError
|
If the key is already registered and exist_ok is False. |
TypeError
|
If the key is rejected by the registry (for example a
non-type key in a |
Example
Basic registration:
>>> from coola.registry import Registry
>>> registry = Registry[str, int]()
>>> registry.register("key1", 42)
>>> registry.get("key1")
42
Attempting to register a duplicate key:
>>> registry.register("key1", 100)
Traceback (most recent call last):
...
coola.registry.exceptions.AlreadyRegisteredError: A value is already registered for 'key1'. Use a different key or set exist_ok=True to override.
Overwriting with exist_ok:
>>> registry.register("key1", 100, exist_ok=True)
>>> registry.get("key1")
100
coola.registry.Registry.register_many ¶
register_many(
mapping: Mapping[K, V], *, exist_ok: bool = False
) -> None
Register multiple key-value pairs in a single operation.
This is a convenience method for bulk registration. It iterates through
the provided mapping and registers each key-value pair. All registrations
follow the same exist_ok policy. The operation is atomic with respect to
this registry when exist_ok is False: either every key-value pair in
mapping ends up registered, or (if any key already exists) none of
them do - there is no partial registration. This says nothing about
atomicity across multiple registries: if mapping is registered into
several registries in sequence, each register_many call is atomic on
its own, but the overall multi-registry operation is not - a failure on
a later registry does not roll back an earlier one.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
mapping
|
Mapping[K, V]
|
A dictionary or mapping containing the key-value pairs to register. The keys and values must match the registry's type parameters. |
required |
exist_ok
|
bool
|
Controls behavior when any key already exists. If False (default), raises error on the first duplicate key. If True, overwrites all existing values without error. |
False
|
Raises:
| Type | Description |
|---|---|
AlreadyRegisteredError
|
If exist_ok is False and any key in the mapping is already registered. The error occurs on the first duplicate encountered, and no partial registration occurs. |
TypeError
|
If a key is rejected by the registry. No key is registered. |
Example
Registering multiple entries at once:
>>> from coola.registry import Registry
>>> registry = Registry[str, int]()
>>> registry.register_many({"key1": 42, "key2": 100, "key3": 7})
>>> registry.get("key1")
42
>>> registry.get("key2")
100
>>> len(registry)
3
Bulk update with exist_ok:
>>> registry.register_many({"key1": 1, "key4": 4}, exist_ok=True)
>>> registry.get("key1")
1
>>> registry.get("key4")
4
coola.registry.Registry.unregister ¶
unregister(key: K) -> V
Remove a key-value pair from the registry and return the value.
This method removes the specified key from the registry and returns the value that was associated with it. This allows you to retrieve the value one last time before it's removed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key
|
K
|
The key to unregister and remove from the registry. |
required |
Returns:
| Type | Description |
|---|---|
V
|
The value that was associated with the key before removal. |
Raises:
| Type | Description |
|---|---|
KeyNotRegisteredError
|
If the key is not registered. It
subclasses |
Example
>>> from coola.registry import Registry
>>> registry = Registry[str, int]({"key1": 42, "key2": 100})
>>> registry.has("key1")
True
>>> value = registry.unregister("key1")
>>> value
42
>>> registry.has("key1")
False
coola.registry.Registry.values ¶
values() -> ValuesView[V]
Return registered values.
Returns:
| Type | Description |
|---|---|
ValuesView[V]
|
The registered values. |
Example
>>> from coola.registry import Registry
>>> registry = Registry[str, int](initial_state={"a": 1, "b": 2})
>>> registry.values()
dict_values([1, 2])
coola.registry.TypeNotRegisteredError ¶
Bases: KeyNotRegisteredError
Raised when no value is registered for a type or any of its parent types.
It subclasses KeyNotRegisteredError (and so KeyError), and it lets
callers distinguish a missing type handler from an ordinary dict miss.
Example
>>> from coola.registry import TypeNotRegisteredError, TypeRegistry
>>> registry = TypeRegistry[str]()
>>> try:
... registry.resolve(int)
... except TypeNotRegisteredError:
... print("not registered")
...
not registered
coola.registry.TypeRegistry ¶
Bases: Registry[type, T], Generic[T]
A thread-safe type-based registry for storing and retrieving values.
The TypeRegistry class provides a thread-safe container for mapping Python types to values. It supports standard dictionary operations through operator overloading and provides methods for safe registration and querying.
The registry uses the Method Resolution Order (MRO) for type lookup
through the resolve() method. When resolving a type, it
automatically selects the most specific registered type, walking up
the inheritance hierarchy if needed. This makes it ideal for
type-based dispatching systems.
The registry includes an internal cache for type resolution, bounded
to _MAX_CACHE_SIZE (1024) entries. The cache is a plain
dictionary that is filled under the lock and replaced by an empty
one on any change of the registry, so cache hits in resolve()
are lock-free and do not contend under threads. When the cache is
full, it is reset rather than evicting entries. It holds strong
references to the types, so a dynamically created class stays alive
until the cache is reset.
A type that cannot be a key (every dataclass, the classes that
define a method, the virtual subclasses of an abstract base class)
is supported by registering a fallback predicate with
register_fallback. The order of precedence of resolve() is:
the registered types (MRO order, object excluded), then the
fallbacks (registration order), then object.
Note: resolve() walks dtype.__mro__, which reflects real
(static) inheritance only. Virtual subclasses registered via
abc.ABCMeta.register() do not appear in __mro__, so
registering a value for an ABC does not make resolve() match
that ABC's virtual subclasses even though isinstance would
report them as instances of the ABC. See resolve() for details.
A fallback on issubclass is deliberately not provided. Many
built-in types are virtual subclasses of Sequence (str,
bytes, range, memoryview...), so such a fallback would
send str to the handler registered for Sequence and, for
instance, iterate over a string character by character. Register the
virtual subclass explicitly instead.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
initial_state
|
dict[type, T] | None
|
An optional dictionary to initialize the registry with. If provided, a copy is made to prevent external modifications. Defaults to None, which creates an empty registry. |
None
|
Attributes: _state: Internal dictionary storing the type-value
pairs. _cache: Bounded dictionary of type resolution lookups, filled
under the lock and replaced by an empty one on any change.
_fallbacks: The (predicate, handler) pairs in registration
order. The dictionary is replaced on write, never mutated. _lock:
Threading lock for synchronizing access to both state and cache.
Raises:
| Type | Description |
|---|---|
TypeError
|
If a key is not a type. |
Example
Basic usage with registration and retrieval:
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]()
>>> registry.register(int, "I am an integer")
>>> registry.get(int)
'I am an integer'
>>> registry
TypeRegistry(
(<class 'int'>): I am an integer
)
Using dictionary-style operations:
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]()
>>> registry[str] = "I am a string"
>>> str in registry
True
>>> registry[str]
'I am a string'
>>> del registry[str]
>>> str in registry
False
Initializing with existing data:
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[int](initial_state={str: 100, float: 200})
>>> len(registry)
2
>>> registry.get(str)
100
Using resolve() with inheritance (MRO lookup):
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]()
>>> registry.register(object, "I am an object")
>>> registry.register(int, "I am an integer")
>>> # Direct match
>>> registry.resolve(int)
'I am an integer'
>>> # Falls back to parent type via MRO
>>> registry.resolve(bool) # bool inherits from int
'I am an integer'
>>> # Falls back to object
>>> registry.resolve(str)
'I am an object'
Bulk registration:
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]()
>>> registry.register_many({int: "integer", float: "float", str: "string"})
>>> len(registry)
3
>>> registry.get(float)
'float'
coola.registry.TypeRegistry.clear ¶
clear() -> None
Remove all the entries, the fallbacks and the pending lazy groups from the registry.
Example
>>> from dataclasses import is_dataclass
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]({int: "int"})
>>> registry.register_fallback(is_dataclass, "dataclass")
>>> registry.clear()
>>> len(registry), len(registry.fallbacks())
(0, 0)
coola.registry.TypeRegistry.copy ¶
copy() -> Self
Create a copy of the registry.
The copy has the same entries, the same fallbacks and the same pending lazy groups as the registry, and is independent from it. The keys, the predicates and the values are not copied.
Returns:
| Type | Description |
|---|---|
Self
|
A new registry of the same class. |
Example
>>> from dataclasses import is_dataclass
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]({object: "object"})
>>> registry.register_fallback(is_dataclass, "dataclass")
>>> other = registry.copy()
>>> other.unregister_fallback(is_dataclass)
'dataclass'
>>> len(registry.fallbacks())
1
coola.registry.TypeRegistry.equal ¶
equal(other: object) -> bool
Indicate if two objects are equal or not.
Two registries are equal if they have the same class, the same entries and the same fallbacks (predicates and handlers).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
other
|
object
|
The object to compare with. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
|
Example
>>> from dataclasses import is_dataclass
>>> from coola.registry import TypeRegistry
>>> registry1 = TypeRegistry[str]({int: "int"})
>>> registry2 = TypeRegistry[str]({int: "int"})
>>> registry1.equal(registry2)
True
>>> registry2.register_fallback(is_dataclass, "dataclass")
>>> registry1.equal(registry2)
False
coola.registry.TypeRegistry.fallbacks ¶
fallbacks() -> ItemsView[Callable[[type], bool], T]
Return a snapshot of the (predicate, value) pairs in
registration order.
A fallback registered after the call does not change the view.
Returns:
| Type | Description |
|---|---|
ItemsView[Callable[[type], bool], T]
|
A view of the |
Example
>>> from dataclasses import is_dataclass
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]()
>>> registry.register_fallback(is_dataclass, "dataclass")
>>> list(registry.fallbacks())
[(<function is_dataclass at ...>, 'dataclass')]
coola.registry.TypeRegistry.register_fallback ¶
register_fallback(
predicate: Callable[[type], bool],
value: T,
*,
exist_ok: bool = False,
) -> None
Register a value for the types accepted by a predicate.
A fallback supports the types that cannot be a key of the
registry, for example every dataclass, the classes that define a
method, or the virtual subclasses of an abstract base class.
resolve tries the fallbacks, in registration order, only if
no registered type (object excluded) matches, and before
object. The predicate must be pure and cheap: its result is
cached per type until the next change of the registry. A
predicate that raises propagates and is not cached.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
predicate
|
Callable[[type], bool]
|
A function that takes a type and returns
|
required |
value
|
T
|
The value to return for the types accepted by
|
required |
exist_ok
|
bool
|
If |
False
|
Raises:
| Type | Description |
|---|---|
AlreadyRegisteredError
|
If the predicate is already
registered and |
Example
>>> from collections.abc import Sequence
>>> from collections import deque
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]({object: "object"})
>>> registry.register_fallback(lambda t: issubclass(t, Sequence), "sequence")
>>> registry.resolve(deque)
'sequence'
coola.registry.TypeRegistry.register_lazy ¶
register_lazy(
packages: str | Sequence[str],
factory: str | Callable[[], Mapping[type, T]],
) -> None
Register entries that are built the first time they are needed.
This is used to register the handlers of an optional library
without importing the library when the registry is created. The
group is loaded the first time the registry is asked for a type
that comes from one of the packages (the top-level package of
the type or of one of its base classes), or for the whole content
of the registry (len, iteration, items, keys,
values, equal). repr, copy and register_lazy
load nothing.
The entries that are loaded never replace the entries that are
already registered, so a type that was registered by hand keeps
its value. A group whose factory raises stays pending, so the
next access raises again. Two groups with a package in common
both stay and are both loaded. clear removes the groups that
are not loaded.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
packages
|
str | Sequence[str]
|
The name or the names of the top-level packages of the group. The group is loaded when a type from any of these packages is needed. |
required |
factory
|
str | Callable[[], Mapping[type, T]]
|
A callable that returns the entries, or its path
|
required |
Example
>>> from decimal import Decimal
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]()
>>> registry.register_lazy("decimal", lambda: {Decimal: "decimal"})
>>> registry.resolve(Decimal)
'decimal'
coola.registry.TypeRegistry.resolve ¶
resolve(dtype: type) -> T
Resolve a type to its associated value using MRO lookup.
This method finds the most appropriate value for a given type by walking the Method Resolution Order (MRO). It first checks for a direct match, then searches through parent types in MRO order to find the most specific registered type.
Results are cached internally to optimize performance for repeated lookups of the same type.
Note
Resolution walks dtype.__mro__, i.e. real (static)
inheritance only. Virtual subclasses registered with
abc.ABCMeta.register() do not appear in __mro__, so
registering a value for an ABC does not make this method
resolve any of that ABC's virtual subclasses, even though
isinstance(instance_of_virtual_subclass, the_abc) is
True. To make a virtual subclass resolve to a value, either
register the subclass itself or a real (non-virtual) ancestor of
it.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dtype
|
type
|
The type to resolve. |
required |
Returns:
| Type | Description |
|---|---|
T
|
The value associated with the type or its nearest registered |
T
|
parent type in the MRO. |
Raises:
| Type | Description |
|---|---|
TypeError
|
If |
TypeNotRegisteredError
|
If no matching type is found in the registry, including parent types in the MRO. |
Example
Basic resolution with inheritance:
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]()
>>> registry.register(object, "base")
>>> registry.register(int, "integer")
>>> # Direct match
>>> registry.resolve(int)
'integer'
>>> # bool inherits from int, so resolves to int's value
>>> registry.resolve(bool)
'integer'
>>> # str inherits from object, so resolves to object's value
>>> registry.resolve(str)
'base'
Resolution with custom classes:
>>> from coola.registry import TypeRegistry
>>> class Animal:
... pass
...
>>> class Dog(Animal):
... pass
...
>>> class Poodle(Dog):
... pass
...
>>> registry = TypeRegistry[str]()
>>> registry.register(Animal, "animal")
>>> registry.register(Dog, "dog")
>>> registry.resolve(Dog)
'dog'
>>> registry.resolve(Poodle) # Resolves to parent Dog
'dog'
coola.registry.TypeRegistry.unregister_fallback ¶
unregister_fallback(
predicate: Callable[[type], bool],
) -> T
Remove a fallback and return its value.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
predicate
|
Callable[[type], bool]
|
The predicate given to |
required |
Returns:
| Type | Description |
|---|---|
T
|
The value that was registered with the predicate. |
Raises:
| Type | Description |
|---|---|
KeyNotRegisteredError
|
If the predicate is not registered. |
Example
>>> from dataclasses import is_dataclass
>>> from coola.registry import TypeRegistry
>>> registry = TypeRegistry[str]()
>>> registry.register_fallback(is_dataclass, "dataclass")
>>> registry.unregister_fallback(is_dataclass)
'dataclass'
coola.registry.default ¶
Define the default registries of the packages built on
coola.registry.
coola.registry.default.make_default_registry_singleton ¶
make_default_registry_singleton(
registry_cls: Callable[[], R],
register_defaults: Callable[[R], None],
*,
plugin_group: str | None = None,
) -> LazySingleton[R]
Create a lazy singleton that builds and populates a default registry on first access.
This removes the near-identical _build_default_registry
functions previously copied in each package, and guarantees the same
thread-safe construction everywhere.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry_cls
|
Callable[[], R]
|
Callable (usually the registry class) that returns a new, empty registry. |
required |
register_defaults
|
Callable[[R], None]
|
Callable that populates the registry with the default entries. |
required |
plugin_group
|
str | None
|
Optional entry-point group. If set, every entry
point in this group is loaded after the defaults and called
with the registry, so third-party packages can register
their types without touching coola (see
|
None
|
Returns:
| Type | Description |
|---|---|
LazySingleton[R]
|
A |
Example
>>> from coola.registry.default import make_default_registry_singleton
>>> singleton = make_default_registry_singleton(dict, lambda r: r.update(a=1))
>>> singleton.get()
{'a': 1}
coola.registry.default.reset_default_registries ¶
reset_default_registries() -> None
Forget the default registries, so the next use of each of them builds a new one.
Every default registry (equality testers, hashers, summarizers,
...) is rebuilt with its default entries and its plugins, and the
registrations that were made with the register_* functions, and
the calls to clear or unregister, are dropped. A registry
that was obtained before the reset is not changed.
Example
>>> from coola.equality.tester import get_default_registry
>>> from coola.registry.default import reset_default_registries
>>> get_default_registry().clear()
>>> reset_default_registries()
>>> get_default_registry().has_equality_tester(int)
True
coola.registry.default.temporary_default_registries ¶
temporary_default_registries() -> Generator[None]
Restore the default registries when the context exits.
Inside the context, the default registries that are already built
are replaced by a copy (copy), so a registration, an
unregister or a clear does not change the original. The
default registries that are not built yet are built inside the
context and forgotten when it exits. It is meant for tests that
register a type, so they do not leak the registration to the other
tests of the process.
The default registries are process-wide: do not enter the context in several threads at the same time.
Yields:
| Type | Description |
|---|---|
Generator[None]
|
|
Example
>>> from coola.equality.tester import DefaultEqualityTester, get_default_registry
>>> from coola.registry.default import temporary_default_registries
>>> class Point:
... pass
...
>>> with temporary_default_registries():
... get_default_registry().register(Point, DefaultEqualityTester())
... get_default_registry().has_equality_tester(Point)
...
True
>>> get_default_registry().has_equality_tester(Point)
False
coola.registry.plugins ¶
Load the entry-point plugins that populate a registry.
coola.registry.plugins.LoadedPlugin
dataclass
¶
A plugin that was loaded and applied to a registry.
Attributes:
| Name | Type | Description |
|---|---|---|
group |
str
|
The entry-point group of the plugin, for example
|
name |
str
|
The name of the entry point. |
value |
str
|
The target of the entry point, for example
|
coola.registry.plugins.get_loaded_plugins ¶
get_loaded_plugins() -> list[LoadedPlugin]
Get the entry-point plugins that were loaded in this process.
A plugin is listed after it was applied to a default registry. The
default registries load their plugins the first time they are
built, so a registry that was not used yet has not loaded its
plugins. It makes it possible to find out which installed package
changed the behaviour of objects_are_equal, hash_object or
summarize, without reading importlib.metadata.
Returns:
| Type | Description |
|---|---|
list[LoadedPlugin]
|
The loaded plugins, in the order in which they were loaded. |
list[LoadedPlugin]
|
The plugins that failed to load, and all the plugins when |
list[LoadedPlugin]
|
|
Example
>>> from coola.registry.plugins import get_loaded_plugins
>>> plugins = get_loaded_plugins()
>>> isinstance(plugins, list)
True
coola.registry.plugins.load_registry_plugins ¶
load_registry_plugins(registry: object, group: str) -> None
Populate a registry from third-party entry points.
Each entry point in group must resolve to a callable that takes
the registry and registers its types on it. A plugin that fails to
load or run emits a RuntimeWarning and is skipped, so one broken
plugin cannot break coola. The failure is also logged with its
traceback. If the COOLA_STRICT_PLUGINS environment variable is
set (e.g. in CI), every plugin is still attempted and all failures
are then raised together as an ExceptionGroup, each annotated
with the plugin name and group. If the COOLA_DISABLE_PLUGINS
environment variable is set, no plugin is loaded. The plugins that
were loaded are listed by get_loaded_plugins.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
registry
|
object
|
The registry passed to each plugin. |
required |
group
|
str
|
The entry-point group to load, e.g.
|
required |
Raises:
| Type | Description |
|---|---|
ExceptionGroup
|
If |
Example
>>> from coola.registry.plugins import load_registry_plugins
>>> load_registry_plugins({}, "coola.nonexistent.group")