File List HWM¶

class etl_entities.hwm.file.file_list_hwm.FileListHWM(*, name: str, description: str = '', directory: AbsolutePath | None = None, value: FrozenSet[AbsolutePath] = None, expression: Any = None, modified_time: datetime = None)¶

HWM based on tracking list of file names.

Parameters:

namestr: HWM unique name
valueset of pathlib.Path, default: empty set: HWM value
directorypathlib.Path, default: None: Directory for HWM value. All paths in value must be relative to this directory.
descriptionstr, default: "": Description of HWM
expressionAny, default: None: HWM expression
modified_timedatetime.datetime, default: current datetime: HWM value modification time

Examples

from etl_entities.hwm import FileListHWM
from etl_entities.instance import AbsolutePath

hwm = FileListHWM(
    name="hwm_name",
    value={"/some/path", "/another.file"},
)

__add__(value: str | PathLike | Iterable[str | PathLike]) → FileListHWMType¶

Adds path or paths to HWM value, and return copy of HWM

Parameters:

valuestr or pathlib.PosixPath or typing.Iterable of them: Path or collection of paths to be added to value

Returns:

resultFileListHWM: HWM copy with new value

Examples

>>> from etl_entities.hwm import FileListHWM
>>> hwm = FileListHWM(value={"/some/path"}, name="my_hwm")
>>> hwm = hwm + "/another.file"
>>> sorted(hwm.value)
[AbsolutePath('/another.file'), AbsolutePath('/some/path')]

__sub__(value: str | PathLike | Iterable[str | PathLike]) → FileListHWMType¶

Remove path or paths from HWM value, and return copy of HWM

Parameters:

valuestr or pathlib.PosixPath or typing.Iterable of them: Path or collection of paths to be added to value

Returns:

resultFileListHWM: HWM copy with new value

Examples

>>> from etl_entities.hwm import FileListHWM
>>> hwm = FileListHWM(value={"/some/path", "/another.file"}, name="my_hwm")
>>> hwm = hwm - "/another.file"
>>> sorted(hwm.value)
[AbsolutePath('/some/path')]

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: str | PathLike) → bool¶

Return True if input value is already covered by HWM

Examples

>>> from etl_entities.hwm import FileListHWM
>>> hwm = FileListHWM(value={"/some/old_file.py"}, name="my_hwm")
>>> hwm.covers("/some/old_file.py")  # path in HWM
True
>>> hwm.covers("/some/new_file.py")  # path not in HWM
False

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

update(value: str | PathLike | Iterable[str | PathLike]) → FileListHWMType¶

Updates current HWM value with some implementation-specific logic, and return HWM.

Note

Changes HWM value in place

Returns:

resultFileListHWM: Self

Examples

>>> from etl_entities.hwm import FileListHWM
>>> hwm = FileListHWM(value=["/some/existing_path.py"], name="my_hwm")
>>> # new paths are appended
>>> hwm = hwm.update("/some/new_path.py")
>>> sorted(hwm.value)
[AbsolutePath('/some/existing_path.py'), AbsolutePath('/some/new_path.py')]
>>> # existing path already here
>>> hwm = hwm.update("/some/existing_path.py")
>>> sorted(hwm.value)
[AbsolutePath('/some/existing_path.py'), AbsolutePath('/some/new_path.py')]