File List HWM#

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

File List HWM type

Parameters:

namestr: HWM name
directorypathlib.PosixPath: Path to directory
valueset of pathlib.Path, default: empty set: HWM value
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",
    directory=AbsolutePath("/folder/path"),
    value={"some/path", "another.file"},
)

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
`dict`(*[, include, exclude, by_alias, ...])	Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
`get_absolute_paths`()	Returns set of files with absolute paths
`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.
`update`(value)	Updates current HWM value with some implementation-specific logic, and return HWM.

__add__(value: str | PathLike | Iterable[str | PathLike])#

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

Returns:

resultFileListHWM: HWM copy with new value

Examples

from etl_entities.hwm import FileListHWM

hwm1 = FileListHWM(value={"some/path"}, ...)
hwm2 = FileListHWM(value={"some/path", "another.file"}, ...)

assert hwm1 + "another.file" == hwm2
# same as FileListHWM(value=hwm1.value | {"another.file"}, ...)

__sub__(value: str | PathLike | Iterable[str | PathLike])#

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

Returns:

resultFileListHWM: HWM copy with new value

Examples

from etl_entities.hwm import FileListHWM

hwm1 = FileListHWM(value={"some/path"}, ...)
hwm2 = FileListHWM(value={"some/path", "another.file"}, ...)

assert hwm1 - "another.file" == hwm2
# same as FileListHWM(value=hwm1.value - {"another.file"}, ...)

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/path.py"}, ...)

assert hwm.covers("some/path.py")  # path in HWM
assert not hwm.covers("another/path.py")  # path not in HWM

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.

get_absolute_paths() → frozenset[AbsolutePath]#

Returns set of files with absolute paths

Returns:

resultfrosenzet of pathlib.PosixPath: Copy of HWM with updated value

Examples

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

hwm = FileListHWM(value={"some/path"}, directory=AbsolutePath("/absolute/path"), ...)

assert hwm.get_absolute_paths() == frozenset({"/absolute/path/some/path"})

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

update(value: str | PathLike | Iterable[str | PathLike])#

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

Note

Changes HWM value in place

Returns:

resultFileHWM: Self

Examples

from etl_entities.hwm import FileListHWM
from etl_entities.instance import RelativePath

hwm = FileListHWM(value=["some/existing_path.py"], ...)

# new paths are appended
hwm.update("some/new_path.py")
assert hwm.value == frozenset(
    {
        RelativePath("some/existing_path.py"),
        RelativePath("some/new_path.py"),
    }
)

# existing paths do nothing
hwm.update("some/existing_path.py")
assert hwm.value == frozenset(
    {
        RelativePath("some/existing_path.py"),
        RelativePath("some/new_path.py"),
    }
)