Column Datetime HWM¶

class etl_entities.hwm.column.datetime_hwm.ColumnDateTimeHWM(*, name: str, description: str = '', source: str | None = None, value: datetime | None = None, expression: Any = None, modified_time: datetime = None)¶

HWM based on tracking latest column value of type datetime.datetime.

Parameters:

namestr: HWM unique name
valuedatetime.datetime or None, default: None: HWM value
descriptionstr, default: "": Description of HWM
sourceAny, default: None: HWM source, e.g. table name
expressionAny, default: None: Expression used to generate HWM value, e.g. column, CAST(column as TYPE)
modified_timedatetime.datetime, default: current datetime: HWM value modification time

Examples

from datetime import datetime
from etl_entities.hwm import ColumnDateTimeHWM

hwm = DateTimeHWM(
    name="long_unique_name",
    source="myschema.mytable",
    expression="my_timestamp_column",
    value=datetime(year=2021, month=12, day=31, hour=11, minute=22, second=33),
)

__add__(value: ColumnValueType) → ColumnHWMType¶

Increase HWM value and return copy of HWM

Parameters:

valueAny or None

Should be compatible with value attribute type.

For example, you cannot add str to int value, but you can add int to int.

Returns:

resultColumnHWM: HWM with new value

Examples

>>> hwm = ColumnHWM(value=100, name="my_hwm")
>>> hwm = hwm + 2
>>> hwm.value
102

__eq__(other)¶

Checks equality of two HWM instances

Parameters:

otheretl_entities.hwm.column_hwm.ColumnHWM: You can compare two etl_entities.hwm.column_hwm.ColumnHWM instances, obj:etl_entities.hwm.column_hwm.ColumnHWM with an object, if its value is comparable with the value attribute of HWM

Returns:

resultbool: True if both inputs are the same, False otherwise.

__lt__(other)¶

Checks current HWM value is less than another one.

Returns:

resultbool: True if current HWM value is less than provided value, False otherwise.

Raises:

NotImplementedError:: If someone tries to compare HWMs with different fields, like name, source or expression

__sub__(value: ColumnValueType) → ColumnHWMType¶

Decrease HWM value, and return copy of HWM

Parameters:

valueAny or None

Should be compatible with value attribute type.

For example, you cannot subtract str from int value, but you can subtract int from int.

Returns:

resultColumnHWM: HWM copy with new value

Examples

>>> hwm = ColumnHWM(value=100, name="my_hwm")
>>> hwm = hwm - 2
>>> hwm.value
98

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters:

include – fields to include in new model
exclude – fields to exclude from new model, as with values this takes precedence over include
update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep – set to True to make a deep copy of the model

Returns:

new model instance

classmethod deserialize(inp: dict) → HWMType¶

Return HWM from dict representation

Returns:

resultHWM: Deserialized HWM

Examples

>>> from etl_entities.hwm import ColumnIntHWM
>>> hwm = ColumnIntHWM.deserialize(
...     {
...         "type": "column_int",
...         "name": "my_hwm",
...         "value": "1",
...         "entity": "some_column",
...         "description": "some description",
...     }
... )
>>> type(hwm)
<class 'etl_entities.hwm.column.int_hwm.ColumnIntHWM'>
>>> hwm.name
'my_hwm'
>>> hwm.value
1
>>> hwm.entity
'some_column'
>>> hwm.description
'some description'
>>> ColumnIntHWM.deserialize({"type": "column_date"})
Traceback (most recent call last):
    ...
ValueError: Type 'column_date' does not match class 'ColumnIntHWM'

dict(*, include: AbstractSetIntStr | MappingIntStrAny | None = None, exclude: AbstractSetIntStr | MappingIntStrAny | None = None, by_alias: bool = False, skip_defaults: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) → DictStrAny¶: Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

json(*, include: AbstractSetIntStr | MappingIntStrAny | None = None, exclude: AbstractSetIntStr | MappingIntStrAny | None = None, by_alias: bool = False, skip_defaults: bool | None = None, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Callable[[Any], Any] | None = None, models_as_dict: bool = True, **dumps_kwargs: Any) → str¶

Generate a JSON representation of the model, include and exclude arguments as per dict().

encoder is an optional function to supply as default to json.dumps(), other arguments as per json.dumps().

set_value(value: ValueType | None) → HWMType¶

Replaces current HWM value with the passed one, and return HWM.

Note

Changes HWM value in place instead of returning new one

Returns:

resultHWM: Self

Examples

>>> from etl_entities.hwm import ColumnIntHWM
>>> hwm = ColumnIntHWM(value=1, name="my_hwm")
>>> hwm = hwm.set_value(2)
>>> hwm.value
2