Bases: BaseModel
HWM Store settings.
HWM Store is used for incremental strategy. See etl-entities documentation <https://etl-entities.readthedocs.io/en/stable/hwm_store/base_hwm_store.html>_.
.. note::
For now, the only supported HWMStore type for now is `Horizon <https://data-horizon.readthedocs.io/>`_.
Examples
.. code-block:: yaml
:caption: config.yml
hwm_store:
# Set the HWM Store connection URL
enabled: true
type: horizon
url: http://horizon:8000
user: some_user
password: changeme
namespace: syncmaster_internal
Source code in syncmaster/worker/settings/hwm_store.py
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68 | class HWMStoreSettings(BaseModel):
"""HWM Store settings.
HWM Store is used for incremental strategy. See `etl-entities documentation <https://etl-entities.readthedocs.io/en/stable/hwm_store/base_hwm_store.html>`_.
.. note::
For now, the only supported HWMStore type for now is `Horizon <https://data-horizon.readthedocs.io/>`_.
Examples
--------
.. code-block:: yaml
:caption: config.yml
hwm_store:
# Set the HWM Store connection URL
enabled: true
type: horizon
url: http://horizon:8000
user: some_user
password: changeme
namespace: syncmaster_internal
"""
enabled: bool = Field(
default=False,
description="Enable or disable HWM Store",
)
type: Literal["horizon"] | None = Field(
default=None,
description="HWM Store type",
)
url: str | None = Field(
default=None,
description="HWM Store URL",
)
user: str | None = Field(
default=None,
description="HWM Store user",
)
password: str | None = Field(
default=None,
description="HWM Store password",
)
namespace: str | None = Field(
default=None,
description="HWM Store namespace",
)
@model_validator(mode="after")
def check_required_fields_if_enabled(self):
if self.enabled:
missing_fields = [
field for field in ("type", "url", "user", "password", "namespace") if getattr(self, field) is None
]
if missing_fields:
fields_str = ", ".join(missing_fields)
msg = f"All fields must be set with enabled HWMStore. Missing: {fields_str}"
raise ValueError(msg)
return self
|