Column Integer HWM

class etl_entities.hwm.column.int_hwm.ColumnIntHWM(*, name: str, description: str = '', column: str, value: StrictInt | None = None, expression: Any = None, modified_time: datetime = None)

Integer HWM type

Parameters:

columnstr

Column name

namestr

Table name

valueint or None, default: None

HWM value

descriptionstr, default: ""

Description of HWM

expressionAny, default: None

HWM expression, for example: CAST(column as TYPE)

modified_timedatetime.datetime, default: current datetime

HWM value modification time

Examples

from etl_entities.hwm import ColumnIntHWM

hwm_int = ColumnIntHWM(column="column_name", value=1, name="table_name")

Methods

copy(*[, include, exclude, update, deep])	Duplicate a model, optionally choose which fields to include, exclude and change.
covers(value)	Return True if input value is already covered by HWM
deserialize(inp)	Return HWM from dict representation
dict(*[, include, exclude, by_alias, ...])	Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
json(*[, include, exclude, by_alias, ...])	Generate a JSON representation of the model, include and exclude arguments as per dict().
set_value(value)	Replaces current HWM value with the passed one, and return HWM.

__add__(value)

Increase HWM value and return copy of HWM

Returns:

resultColumnHWM

HWM copy with new value

Examples

# assume val2 == val1 + inc

hwm1 = ColumnHWM(value=val1, ...)
hwm2 = ColumnHWM(value=val2, ...)

# same as ColumnHWM(value=hwm1.value + inc, ...)
assert hwm1 + inc == hwm2

__eq__(other)

Checks equality of two HWM instances

Returns:

resultbool

True if both inputs are the same, False otherwise.

Examples

from etl_entities.hwm import ColumnIntHWM

hwm1 = ColumnIntHWM(value=1, ...)
hwm2 = ColumnIntHWM(value=2, ...)

assert hwm1 == hwm1
assert hwm1 != hwm2

__lt__(other)

Checks current ColumnIntHWM value is less than another one

Returns:

resultbool

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

Examples

from etl_entities.hwm import ColumnIntHWM

hwm1 = ColumnIntHWM(value=1, ...)
hwm2 = ColumnIntHWM(value=2, ...)

assert hwm1 < hwm2
assert hwm1 > hwm2

assert hwm1 < 2
assert hwm1 > 0

hwm3 = ColumnIntHWM(value=None, ...)
assert hwm1 < hwm3  # will raise TypeError
assert hwm1 < None  # same thing

__sub__(value)

Decrease HWM value, and return copy of HWM

Returns:

resultColumnHWM

HWM copy with new value

Examples

# assume val2 == val1 - dec

hwm1 = ColumnHWM(value=val1, ...)
hwm2 = ColumnHWM(value=val2, ...)

# same as ColumnHWM(value=hwm1.value - dec, ...)
assert hwm1 - dec == hwm2

copy(*, include: AbstractSetIntStr | MappingIntStrAny | None = None, exclude: AbstractSetIntStr | MappingIntStrAny | None = None, update: DictStrAny | None = None, deep: bool = False) → Model

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

covers(value: ColumnValueType | None) → bool

Return True if input value is already covered by HWM

Examples

hwm = ColumnHWM(column="column_name", value=1)

assert hwm.covers(0)  # 0 <= 1
assert hwm.covers(1)  # 1 <= 1
assert hwm.covers(0.5)  # 0.5 <= 1
assert not hwm.covers(2)  # 2 > 1

empty_hwm = ColumnHWM(column="column_name")

assert not empty_hwm.covers(0)  # non comparable with None
assert not empty_hwm.covers(1)  # non comparable with None
assert not empty_hwm.covers(0.5)  # non comparable with None
assert not empty_hwm.covers(2)  # non comparable with None

classmethod deserialize(inp: dict)

Return HWM from dict representation

Returns:

resultHWM

Deserialized HWM

Examples

from etl_entities.hwm import ColumnIntHWM

assert ColumnIntHWM.deserialize(
    {
        "value": "1",
        "type": "int",
        "column": "column_name",
        "name": "name",
    }
) == ColumnIntHWM(value=1, ...)

ColumnIntHWM.deserialize({"type": "date"})  # raises ValueError

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) → unicode

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) → HWM

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, ...)

hwm.set_value(2)
assert hwm.value == 2