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:
- name
str
HWM name
- directory
pathlib.PosixPath
Path to directory
- value
set
ofpathlib.Path
, default: empty set HWM value
- modified_time
datetime.datetime
, default: current datetime HWM value modification time
- name
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 HWMdict
(*[, include, exclude, by_alias, ...])Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.
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"}, ...)
- 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: str | PathLike) bool #
Return
True
if input value is already covered by HWMExamples
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:
- result
frosenzet
ofpathlib.PosixPath
Copy of HWM with updated value
- result
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"), } )