Register own HWM Store class¶

@etl_entities.hwm_store.hwm_store_class_registry.register_hwm_store_class¶

Decorator for registering some Store class with a name.

Thin wrapper for HWMStoreClassRegistry.add.

Examples

from etl_entities.hwm_store import (
    HWMStoreClassRegistry,
    register_hwm_store_class,
    BaseHWMStore,
)

@register_hwm_store_class("my_store")
class MyHWMStore(BaseHWMStore): ...

HWMStoreClassRegistry.get("my_store") == MyHWMStore

class etl_entities.hwm_store.hwm_store_class_registry.HWMStoreClassRegistry¶

Registry class of different HWM stores.

classmethod add(alias: str, klass: type[BaseHWMStore]) → None¶

Add alias for HWM Store class.

This alias then can be used by detect_hwm_store.

Examples

>>> from etl_entities.hwm_store import HWMStoreClassRegistry, BaseHWMStore
>>> HWMStoreClassRegistry.get("my_store")  # raise KeyError
Traceback (most recent call last):
    ...
KeyError: "Unknown HWM Store type 'my_store'"

>>> class MyHWMStore(BaseHWMStore): ...
>>> HWMStoreClassRegistry.add("my_store", MyHWMStore)
>>> HWMStoreClassRegistry.get("my_store")
<class 'etl_entities.hwm_store.hwm_store_class_registry.MyHWMStore'>

classmethod aliases() → Collection[str]¶: Returl all known HWM store aliases, like memory or yaml

classmethod get(alias: str | None = None) → type¶

Return HWM Store class by its alias, or return default HWM Store class.

Examples

>>> from etl_entities.hwm_store import HWMStoreClassRegistry
>>> HWMStoreClassRegistry.get("memory")
<class 'etl_entities.hwm_store.memory_hwm_store.MemoryHWMStore'>
>>> HWMStoreClassRegistry.get("unknown")
Traceback (most recent call last):
    ...
KeyError: "Unknown HWM Store type 'unknown'"
>>> HWMStoreClassRegistry.get()  # some default HWM Store, see `set_default`
<class 'NoneType'>

classmethod set_default(klass: type[BaseHWMStore]) → None¶

Set specific HWM store class as default HWM Store implementation.

Examples

>>> from etl_entities.hwm_store import HWMStoreClassRegistry, BaseHWMStore
>>> class MyHWMStore(BaseHWMStore): ...
>>> HWMStoreClassRegistry.set_default(MyHWMStore)
>>> HWMStoreClassRegistry.get()
<class 'etl_entities.hwm_store.hwm_store_class_registry.MyHWMStore'>