API¶
Much of HYDROLIB-core's classes that represent input data are based on BaseModel
and FileModel
.
These build upon the pydantic package for data validation.
BaseModel and FileModel¶
Here we define our Pydantic BaseModel
with custom settings,
as well as a FileModel
that inherits from a BaseModel
but
also represents a file on disk.
BaseModel (BaseModel)
pydantic-model
¶
Config
¶
alias_generator(string: str) -> str
¶
Construct a key name from a given field name. The given field name may be a Pydantic Field alias, and the key name is intended to be used as a BaseModel class member variable name.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
string |
str |
input field name |
required |
Source code in hydrolib/core/basemodel.py
def to_key(string: str) -> str:
"""
Construct a key name from a given field name.
The given field name may be a Pydantic Field alias, and the key name
is intended to be used as a BaseModel class member variable name.
Args:
string (str): input field name
"""
# First replace any leading digits, because those are undesirable
# in variable names.
digitdict = {
"0": "zero",
"1": "one",
"2": "two",
"3": "three",
"4": "four",
"5": "five",
"6": "six",
"7": "seven",
"8": "eight",
"9": "nine",
}
m = re.search(r"^\d+", string)
if m:
digitstring = string[0 : m.end()]
for key, val in digitdict.items():
digitstring = digitstring.replace(key, val)
string = digitstring + string[m.end() :]
# Next, replace spaces and hyphens in the potential variable name.
return string.lower().replace(" ", "_").replace("-", "")
__init__(self, **data: Any) -> None
special
¶
Initialize a BaseModel with the provided data.
Exceptions:
Type | Description |
---|---|
ValidationError |
A validation error when the data is invalid. |
Source code in hydrolib/core/basemodel.py
def __init__(self, **data: Any) -> None:
"""Initialize a BaseModel with the provided data.
Raises:
ValidationError: A validation error when the data is invalid.
"""
try:
super().__init__(**data)
except ValidationError as e:
# Give a special message for faulty list input
for re in e.raw_errors:
if (
hasattr(re, "_loc")
and hasattr(re.exc, "msg_template")
and isinstance(data.get(to_key(re._loc)), list)
):
re.exc.msg_template += (
f". The key {re._loc} might be duplicated in the input file."
)
# Update error with specific model location name
identifier = self._get_identifier(data)
if identifier is None:
raise e
else:
# If there is an identifier, include this in the ValidationError messages.
raise ValidationError([ErrorWrapper(e, loc=identifier)], self.__class__)
is_file_link(self) -> bool
¶
Generic attribute for models backed by a file.
Source code in hydrolib/core/basemodel.py
def is_file_link(self) -> bool:
"""Generic attribute for models backed by a file."""
return False
is_intermediate_link(self) -> bool
¶
Generic attribute for models that have children fields that could contain files.
Source code in hydrolib/core/basemodel.py
def is_intermediate_link(self) -> bool:
"""Generic attribute for models that have children fields that could contain files."""
return self.is_file_link()
show_tree(self, indent = 0)
¶
Recursive print function for showing a tree of a model.
Source code in hydrolib/core/basemodel.py
def show_tree(self, indent=0):
"""Recursive print function for showing a tree of a model."""
angle = "∟" if indent > 0 else ""
# Only print if we're backed by a file
if self.is_file_link():
print(" " * indent * 2, angle, self)
# Otherwise we recurse through the fields of a model
for _, value in self:
# Handle lists of items
if not isinstance(value, list):
value = [value]
for v in value:
if hasattr(v, "is_intermediate_link") and v.is_intermediate_link():
# If the field is only an intermediate, print the name only
if not v.is_file_link():
print(" " * (indent * 2 + 2), angle, v.__class__.__name__)
v.show_tree(indent + 1)
DiskOnlyFileModel (FileModel)
pydantic-model
¶
DiskOnlyFileModel provides a stub implementation for file based models which are not explicitly implemented within hydrolib.core.
It implements the FileModel with a void parser and serializer, and a save method which copies the file associated with the FileModel to a new location if it exists.
We further explicitly assume that when the filepath is None, no file will be written.
Actual file model implementations should not inherit from the DiskOnlyFileModel and instead inherit directly from FileModel.
is_intermediate_link(self) -> bool
¶
Generic attribute for models that have children fields that could contain files.
Source code in hydrolib/core/basemodel.py
def is_intermediate_link(self) -> bool:
# If the filepath is not None, there is an underlying file, and as such we need
# to traverse it.
return self.filepath is not None
FileCasingResolver
¶
Class for resolving file path in a case-insensitive manner.
__init__(self) -> None
special
¶
Initializes a new instance of the FileCasingResolver
class.
Source code in hydrolib/core/basemodel.py
def __init__(self) -> None:
"""Initializes a new instance of the `FileCasingResolver` class."""
self._has_initialized = False
self._resolve_casing = False
initialize_resolve_casing(self, resolve_casing: bool) -> None
¶
Initialize the setting to resolve casing or not. Can only be set once.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
resolve_casing |
bool |
Whether or not to resolve the file casing. |
required |
Source code in hydrolib/core/basemodel.py
def initialize_resolve_casing(self, resolve_casing: bool) -> None:
"""Initialize the setting to resolve casing or not. Can only be set once.
Args:
resolve_casing (bool): Whether or not to resolve the file casing.
"""
if self._has_initialized:
return
self._resolve_casing = resolve_casing
self._has_initialized = True
resolve(self, path: Path) -> Path
¶
Resolve the casing of a file path when the file does exist but not with the exact casing.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path |
Path |
The path of the file or directory for which the casing needs to be resolved. |
required |
Returns:
Type | Description |
---|---|
Path |
The file path with the matched casing if a match exists; otherwise, the original file path. |
Exceptions:
Type | Description |
---|---|
NotImplementedError |
When this function is called with an operating system other than Windows, Linux or MacOS. |
Source code in hydrolib/core/basemodel.py
def resolve(self, path: Path) -> Path:
"""Resolve the casing of a file path when the file does exist but not with the exact casing.
Args:
path (Path): The path of the file or directory for which the casing needs to be resolved.
Returns:
Path: The file path with the matched casing if a match exists; otherwise, the original file path.
Raises:
NotImplementedError: When this function is called with an operating system other than Windows, Linux or MacOS.
"""
if not self._resolve_casing:
return path
operating_system = get_operating_system()
if operating_system == OperatingSystem.WINDOWS:
return self._resolve_casing_windows(path)
if operating_system == OperatingSystem.LINUX:
return self._resolve_casing_linux(path)
if operating_system == OperatingSystem.MACOS:
return self._resolve_casing_macos(path)
else:
raise NotImplementedError(
f"Path case resolving for operating system {operating_system} is not supported yet."
)
FileLoadContext
¶
FileLoadContext provides the context necessary to resolve paths during the init of a FileModel, as well as ensure the relevant models are only read once.
__init__(self) -> None
special
¶
Create a new empty FileLoadContext.
Source code in hydrolib/core/basemodel.py
def __init__(self) -> None:
"""Create a new empty FileLoadContext."""
self._path_resolver = FilePathResolver()
self._cache = FileModelCache()
self._file_casing_resolver = FileCasingResolver()
get_current_parent(self) -> Path
¶
Get the current absolute path with which files are resolved.
If the current mode is relative to the parent, the latest added parent is added. If the current mode is relative to an anchor path, the latest added anchor path is returned.
Returns:
Type | Description |
---|---|
Path |
The absolute path to the current parent. |
Source code in hydrolib/core/basemodel.py
def get_current_parent(self) -> Path:
"""Get the current absolute path with which files are resolved.
If the current mode is relative to the parent, the latest added
parent is added. If the current mode is relative to an anchor
path, the latest added anchor path is returned.
Returns:
Path: The absolute path to the current parent.
"""
return self._path_resolver.get_current_parent()
pop_last_parent(self) -> None
¶
Pop the last added parent off this FileLoadContext.
Source code in hydrolib/core/basemodel.py
def pop_last_parent(self) -> None:
"""Pop the last added parent off this FileLoadContext."""
self._path_resolver.pop_last_parent()
push_new_parent(self, parent_path: Path, relative_mode: ResolveRelativeMode) -> None
¶
Push a new parent_path with the given relative_mode on this FileLoadContext.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
parent_path |
Path |
The parent path to be added to this FileLoadContext. |
required |
relative_mode |
ResolveRelativeMode |
The relative mode. |
required |
Source code in hydrolib/core/basemodel.py
def push_new_parent(
self, parent_path: Path, relative_mode: ResolveRelativeMode
) -> None:
"""Push a new parent_path with the given relative_mode on this
FileLoadContext.
Args:
parent_path (Path): The parent path to be added to this FileLoadContext.
relative_mode (ResolveRelativeMode): The relative mode.
"""
self._path_resolver.push_new_parent(parent_path, relative_mode)
register_model(self, path: Path, model: FileModel) -> None
¶
Associate the provided model with the provided path.
Relative paths will be resolved based on the current state of the FileLoadContext.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path |
Path |
The relative path from which the model was loaded. |
required |
model |
FileModel |
The loaded model. |
required |
Source code in hydrolib/core/basemodel.py
def register_model(self, path: Path, model: "FileModel") -> None:
"""Associate the provided model with the provided path.
Relative paths will be resolved based on the current state of the
FileLoadContext.
Args:
path (Path): The relative path from which the model was loaded.
model (FileModel): The loaded model.
"""
absolute_path = self._path_resolver.resolve(path)
self._cache.register_model(absolute_path, model)
resolve(self, path: Path) -> Path
¶
Resolve the provided path.
If path is already absolute, it will be returned as is. Otherwise it will be resolved based on the current state of this FileLoadContext.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path |
Path |
The path to be resolved. |
required |
Returns:
Type | Description |
---|---|
Path |
An absolute path resolved based on the current state. |
Source code in hydrolib/core/basemodel.py
def resolve(self, path: Path) -> Path:
"""Resolve the provided path.
If path is already absolute, it will be returned as is. Otherwise
it will be resolved based on the current state of this FileLoadContext.
Args:
path (Path): The path to be resolved.
Returns:
Path: An absolute path resolved based on the current state.
"""
return self._path_resolver.resolve(path)
resolve_casing(self, file_path: Path) -> Path
¶
Resolve the file casing for the provided file path.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
file_path |
Path |
The file path to resolve the casing for. |
required |
Returns:
Type | Description |
---|---|
Path |
The resolved file path. |
Source code in hydrolib/core/basemodel.py
def resolve_casing(self, file_path: Path) -> Path:
"""Resolve the file casing for the provided file path.
Args:
file_path (Path): The file path to resolve the casing for.
Returns:
Path: The resolved file path.
"""
return self._file_casing_resolver.resolve(file_path)
retrieve_model(self, path: Optional[pathlib.Path]) -> Optional[FileModel]
¶
Retrieve the model associated with the path.
If no model has been associated with the provided path, or the path is None, then None will be returned. Relative paths will be resolved based on the current state of the FileLoadContext.
Returns:
Type | Description |
---|---|
[Optional[FileModel]] |
The file model associated with the provided path if it has been registered, else None. |
Source code in hydrolib/core/basemodel.py
def retrieve_model(self, path: Optional[Path]) -> Optional["FileModel"]:
"""Retrieve the model associated with the path.
If no model has been associated with the provided path, or the path is None,
then None will be returned. Relative paths will be resolved based on the
current state of the FileLoadContext.
Returns:
[Optional[FileModel]]:
The file model associated with the provided path if it has been
registered, else None.
"""
if path is None:
return None
absolute_path = self._path_resolver.resolve(path)
return self._cache.retrieve_model(absolute_path)
set_resolve_casing(self, resolve_casing: bool) -> None
¶
Set the setting to resolve casing or not.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
resolve_casing |
bool |
Whether or not to resolve the file casing. |
required |
Source code in hydrolib/core/basemodel.py
def set_resolve_casing(self, resolve_casing: bool) -> None:
"""Set the setting to resolve casing or not.
Args:
resolve_casing (bool): Whether or not to resolve the file casing.
"""
self._file_casing_resolver.initialize_resolve_casing(resolve_casing)
FileModel (BaseModel, ABC)
pydantic-model
¶
Base class to represent models with a file representation.
It therefore always has a filepath
and if it is given on
initilization, it will parse that file. The filepath can be
relative, in which case the paths are expected to be resolved
relative to some root model. If a path is absolute, this path
will always be used, regardless of a root parent.
When saving a model, if the current filepath is relative, the last resolved absolute path will be used. If the model has just been read, the
This class extends the validate
option of Pydantic,
so when when a Path is given to a field with type FileModel
,
it doesn't error, but actually initializes the FileModel
.
Attributes:
Name | Type | Description |
---|---|---|
filepath |
Optional[Path] |
The path of this FileModel. This path can be either absolute or relative. If it is a relative path, it is assumed to be resolved from some root model. |
save_location |
Path |
A readonly property corresponding with the (current) save location of this FileModel. If read from a file or after saving recursively or after calling synchronize_filepath, this value will be updated to its new state. If made from memory and filepath is not set, it will correspond with cwd / filename.extension |
__init__(self, filepath: Optional[pathlib.Path] = None, resolve_casing: bool = False, *args, **kwargs)
special
¶
Create a new FileModel from the given filepath.
If no filepath is provided, the model is initialized as an empty model with default values. If the filepath is provided, it is read from disk.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
filepath |
Optional[Path] |
The file path. Defaults to None. |
None |
resolve_casing |
bool |
Whether or not to resolve the file name references so that they match the case with what is on disk. Defaults to False. |
False |
Source code in hydrolib/core/basemodel.py
def __init__(
self,
filepath: Optional[Path] = None,
resolve_casing: bool = False,
*args,
**kwargs,
):
"""Create a new FileModel from the given filepath.
If no filepath is provided, the model is initialized as an empty
model with default values.
If the filepath is provided, it is read from disk.
Args:
filepath (Optional[Path], optional): The file path. Defaults to None.
resolve_casing (bool, optional): Whether or not to resolve the file name references so that they match the case with what is on disk. Defaults to False.
"""
if not filepath:
super().__init__(*args, **kwargs)
return
with file_load_context() as context:
context.set_resolve_casing(resolve_casing)
self._absolute_anchor_path = context.get_current_parent()
loading_path = context.resolve(filepath)
loading_path = context.resolve_casing(loading_path)
filepath = self._get_updated_file_path(filepath, loading_path)
context.register_model(filepath, self)
logger.info(f"Loading data from {filepath}")
data = self._load(loading_path)
data["filepath"] = filepath
kwargs.update(data)
# Note: the relative mode needs to be obtained from the data directly
# because self._relative_mode has not been resolved yet (this is done as
# part of the __init__), however during the __init__ we need to already
# have pushed the new parent. As such we cannot move this call later.
relative_mode = self._get_relative_mode_from_data(data)
context.push_new_parent(filepath.parent, relative_mode)
super().__init__(*args, **kwargs)
self._post_init_load()
context.pop_last_parent()
__new__(cls, filepath: Optional[pathlib.Path] = None, *args, **kwargs)
special
staticmethod
¶
Create a new model. If the file at the provided file path was already parsed, this instance is returned.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
filepath |
Optional[Path] |
The file path to the file. Defaults to None. |
None |
Returns:
Type | Description |
---|---|
FileModel |
A file model. |
Source code in hydrolib/core/basemodel.py
def __new__(cls, filepath: Optional[Path] = None, *args, **kwargs):
"""Create a new model.
If the file at the provided file path was already parsed, this instance is returned.
Args:
filepath (Optional[Path], optional): The file path to the file. Defaults to None.
Returns:
FileModel: A file model.
"""
with file_load_context() as context:
if (file_model := context.retrieve_model(filepath)) is not None:
return file_model
else:
return super().__new__(cls)
__str__(self) -> str
special
¶
Return str(self).
Source code in hydrolib/core/basemodel.py
def __str__(self) -> str:
return str(self.filepath if self.filepath else "")
is_file_link(self) -> bool
¶
Generic attribute for models backed by a file.
Source code in hydrolib/core/basemodel.py
def is_file_link(self) -> bool:
return True
save(self, filepath: Optional[pathlib.Path] = None, recurse: bool = False) -> None
¶
Save the model to disk.
If recurse is set to True, all of the child FileModels will be saved as well. Relative child models are stored relative to this Model, according to the model file hierarchy specified with the respective filepaths. Absolute paths will be written to their respective locations. Note that this will overwrite any existing files that are stored in this location.
Note that if recurse is set to True, the save_location properties of the children are updated to their respective new locations.
If filepath it is specified, the filepath of this FileModel is set to the specified path before the save operation is executed. If none is specified it will use the current filepath.
If the used filepath is relative, it will be stored at the current save_location. If you only want to save a child model of some root model, it is recommended to first call synchronize_filepaths on the root model, to ensure the child model's save_location is correctly determined.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
filepath |
Optional[Path] |
The file path at which this model is saved. If None is specified it defaults to the filepath currently stored in the filemodel. Defaults to None. |
None |
recurse |
bool |
Whether to save all children of this FileModel (when set to True), or only save this model (when set to False). Defaults to False. |
False |
Source code in hydrolib/core/basemodel.py
def save(self, filepath: Optional[Path] = None, recurse: bool = False) -> None:
"""Save the model to disk.
If recurse is set to True, all of the child FileModels will be saved as well.
Relative child models are stored relative to this Model, according to the
model file hierarchy specified with the respective filepaths.
Absolute paths will be written to their respective locations. Note that this
will overwrite any existing files that are stored in this location.
Note that if recurse is set to True, the save_location properties of the
children are updated to their respective new locations.
If filepath it is specified, the filepath of this FileModel is set to the
specified path before the save operation is executed. If none is specified
it will use the current filepath.
If the used filepath is relative, it will be stored at the current
save_location. If you only want to save a child model of some root model, it is
recommended to first call synchronize_filepaths on the root model, to ensure
the child model's save_location is correctly determined.
Args:
filepath (Optional[Path], optional):
The file path at which this model is saved. If None is specified
it defaults to the filepath currently stored in the filemodel.
Defaults to None.
recurse (bool, optional):
Whether to save all children of this FileModel (when set to True),
or only save this model (when set to False). Defaults to False.
"""
if filepath is not None:
self.filepath = filepath
# Handle save
with file_load_context() as context:
context.push_new_parent(self._absolute_anchor_path, self._relative_mode)
if recurse:
self._save_tree(context)
else:
self._save_instance()
save_location: Optional[pathlib.Path]
property
readonly
¶
Get the current save location which will be used when calling save()
This value can be None if the filepath is None and no name can be generated.
Returns:
Type | Description |
---|---|
Path |
The location at which this model will be saved. |
synchronize_filepaths(self) -> None
¶
Synchronize the save_location properties of all child models respective to this FileModel's save_location.
Source code in hydrolib/core/basemodel.py
def synchronize_filepaths(self) -> None:
"""Synchronize the save_location properties of all child models respective to
this FileModel's save_location.
"""
def sync_pre(model: BaseModel, acc: FileLoadContext) -> FileLoadContext:
if isinstance(model, FileModel):
acc.push_new_parent(model.filepath.parent, model._relative_mode) # type: ignore[arg-type]
return acc
def sync_post(model: BaseModel, acc: FileLoadContext) -> FileLoadContext:
if isinstance(model, FileModel):
acc.pop_last_parent()
model._absolute_anchor_path = acc.get_current_parent()
return acc
traverser = ModelTreeTraverser[FileLoadContext](
should_traverse=_should_traverse,
should_execute=_should_execute,
pre_traverse_func=sync_pre,
post_traverse_func=sync_post,
)
with file_load_context() as context:
context.push_new_parent(self._absolute_anchor_path, self._relative_mode)
traverser.traverse(self, context)
FileModelCache
¶
FileModelCache provides a simple structure to register and retrieve FileModel objects.
__init__(self)
special
¶
Create a new empty FileModelCache.
Source code in hydrolib/core/basemodel.py
def __init__(self):
"""Create a new empty FileModelCache."""
self._cache_dict: Dict[Path, "FileModel"] = {}
register_model(self, path: Path, model: FileModel) -> None
¶
Register the model with the specified path in this FileModelCache.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path |
Path |
The path to associate the model with. |
required |
model |
FileModel |
The model to be associated with the path. |
required |
Source code in hydrolib/core/basemodel.py
def register_model(self, path: Path, model: "FileModel") -> None:
"""Register the model with the specified path in this FileModelCache.
Args:
path (Path): The path to associate the model with.
model (FileModel): The model to be associated with the path.
"""
self._cache_dict[path] = model
retrieve_model(self, path: Path) -> Optional[FileModel]
¶
Retrieve the model associated with the (absolute) path if it has been registered before, otherwise return None.
Returns:
Type | Description |
---|---|
[Optional[FileModel]] |
The FileModel associated with the Path if it has been registered before, otherwise None. |
Source code in hydrolib/core/basemodel.py
def retrieve_model(self, path: Path) -> Optional["FileModel"]:
"""Retrieve the model associated with the (absolute) path if
it has been registered before, otherwise return None.
Returns:
[Optional[FileModel]]:
The FileModel associated with the Path if it has been registered
before, otherwise None.
"""
return self._cache_dict.get(path, None)
FilePathResolver
¶
FilePathResolver is responsible for resolving relative paths.
The current state to which paths are resolved can be altered by pushing a new parent path to the FilePathResolver, or removing the latest added parent path from the FilePathResolver
__init__(self) -> None
special
¶
Create a new empty FilePathResolver.
Source code in hydrolib/core/basemodel.py
def __init__(self) -> None:
"""Create a new empty FilePathResolver."""
self._anchors: List[Path] = []
self._parents: List[Tuple[Path, ResolveRelativeMode]] = []
get_current_parent(self) -> Path
¶
Get the current absolute path with which files are resolved.
If the current mode is relative to the parent, the latest added parent is added. If the current mode is relative to an anchor path, the latest added anchor path is returned.
Returns:
Type | Description |
---|---|
Path |
The absolute path to the current parent. |
Source code in hydrolib/core/basemodel.py
def get_current_parent(self) -> Path:
"""Get the current absolute path with which files are resolved.
If the current mode is relative to the parent, the latest added
parent is added. If the current mode is relative to an anchor
path, the latest added anchor path is returned.
Returns:
Path: The absolute path to the current parent.
"""
if self._anchor:
return self._anchor
return self._direct_parent
pop_last_parent(self) -> None
¶
Pop the last added parent from this FilePathResolver
If there are currently no parents defined, nothing will happen.
Source code in hydrolib/core/basemodel.py
def pop_last_parent(self) -> None:
"""Pop the last added parent from this FilePathResolver
If there are currently no parents defined, nothing will happen.
"""
if not self._parents:
return
_, relative_mode = self._parents.pop()
if relative_mode == ResolveRelativeMode.ToAnchor:
self._anchors.pop()
push_new_parent(self, parent_path: Path, relative_mode: ResolveRelativeMode) -> None
¶
Push a new parent_path with the given relative_mode to this FilePathResolver.
Relative paths added to this FilePathResolver will be resolved with respect to the current state, i.e. similar to FilePathResolver.resolve.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
parent_path |
Path |
The parent path |
required |
relative_mode |
ResolveRelativeMode |
The relative mode used to resolve. |
required |
Source code in hydrolib/core/basemodel.py
def push_new_parent(
self, parent_path: Path, relative_mode: ResolveRelativeMode
) -> None:
"""Push a new parent_path with the given relative_mode to this FilePathResolver.
Relative paths added to this FilePathResolver will be resolved with respect
to the current state, i.e. similar to FilePathResolver.resolve.
Args:
parent_path (Path): The parent path
relative_mode (ResolveRelativeMode): The relative mode used to resolve.
"""
absolute_parent_path = self.resolve(parent_path)
if relative_mode == ResolveRelativeMode.ToAnchor:
self._anchors.append(absolute_parent_path)
self._parents.append((absolute_parent_path, relative_mode))
resolve(self, path: Path) -> Path
¶
Resolve the provided path to an absolute path given the current state.
If the provided path is already absolute, it will be returned as is.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
path |
Path |
The path to resolve |
required |
Returns:
Type | Description |
---|---|
Path |
An absolute path resolved given the current state. |
Source code in hydrolib/core/basemodel.py
def resolve(self, path: Path) -> Path:
"""Resolve the provided path to an absolute path given the current state.
If the provided path is already absolute, it will be returned as is.
Args:
path (Path): The path to resolve
Returns:
Path: An absolute path resolved given the current state.
"""
if path.is_absolute():
return path
parent = self.get_current_parent()
return (parent / path).resolve()
ModelTreeTraverser (Generic)
¶
ModelTreeTraverser is responsible for traversing a ModelTree using the provided functions.
The ModelTreeTraverser will only traverse BaseModel and derived objects.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
Generic |
[type] |
The type of Accumulator to be used. |
required |
__init__(self, should_traverse: Optional[Callable[[hydrolib.core.basemodel.BaseModel, ~TAcc], bool]] = None, should_execute: Optional[Callable[[hydrolib.core.basemodel.BaseModel, ~TAcc], bool]] = None, pre_traverse_func: Optional[Callable[[hydrolib.core.basemodel.BaseModel, ~TAcc], ~TAcc]] = None, post_traverse_func: Optional[Callable[[hydrolib.core.basemodel.BaseModel, ~TAcc], ~TAcc]] = None)
special
¶
Create a new ModelTreeTraverser with the given functions.
If a predicate it is not defined, it is assumed to always be true, i.e. we will always traverse to the next node, or always execute the traverse functions.
If a traverse function is not defined, it will be skipped.
The traverse functions share an accumulator, i.e. the accumulator argument is passed through all evaluated traverse functions. It is expected that the traverse function return the (potentially) changed accumulator.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
should_traverse |
Optional[Callable[[BaseModel, TAcc], bool]] |
Function to evaluate whether to traverse to the provided BaseModel. Defaults to None. |
None |
should_execute |
Optional[Callable[[BaseModel, TAcc], bool]] |
Function to evaluate whether to execute the traverse functions for the provided BaseModel. Defaults to None. |
None |
pre_traverse_func |
Callable[[BaseModel, TAcc], TAcc] |
Traverse function executed before we traverse into the next BaseModel, i.e. top-down traversal. Defaults to None. |
None |
post_traverse_func |
Callable[[BaseModel, TAcc], TAcc] |
Traverse function executed after we traverse into the next BaseModel, i.e. bottom-up traversal. Defaults to None. |
None |
Source code in hydrolib/core/basemodel.py
def __init__(
self,
should_traverse: Optional[Callable[[BaseModel, TAcc], bool]] = None,
should_execute: Optional[Callable[[BaseModel, TAcc], bool]] = None,
pre_traverse_func: Optional[Callable[[BaseModel, TAcc], TAcc]] = None,
post_traverse_func: Optional[Callable[[BaseModel, TAcc], TAcc]] = None,
):
"""Create a new ModelTreeTraverser with the given functions.
If a predicate it is not defined, it is assumed to always be true, i.e. we will
always traverse to the next node, or always execute the traverse functions.
If a traverse function is not defined, it will be skipped.
The traverse functions share an accumulator, i.e. the accumulator argument
is passed through all evaluated traverse functions. It is expected that the
traverse function return the (potentially) changed accumulator.
Args:
should_traverse (Optional[Callable[[BaseModel, TAcc], bool]], optional):
Function to evaluate whether to traverse to the provided BaseModel. Defaults to None.
should_execute (Optional[Callable[[BaseModel, TAcc], bool]], optional):
Function to evaluate whether to execute the traverse functions for the
provided BaseModel. Defaults to None.
pre_traverse_func (Callable[[BaseModel, TAcc], TAcc], optional):
Traverse function executed before we traverse into the next BaseModel,
i.e. top-down traversal. Defaults to None.
post_traverse_func (Callable[[BaseModel, TAcc], TAcc], optional):
Traverse function executed after we traverse into the next BaseModel,
i.e. bottom-up traversal. Defaults to None.
"""
self._should_traverse_func = should_traverse
self._should_execute_func = should_execute
self._pre_traverse_func = pre_traverse_func
self._post_traverse_func = post_traverse_func
traverse(self, model: BaseModel, acc: ~TAcc) -> ~TAcc
¶
Traverse the model tree of BaseModels including the model as the root, with the provided state of the acc and return the final accumulator.
The actual executed functions as well as the predicates defining whether these functions should be executed for this model as well as whether child BaseModel objects should be traversed are provided in the constructor of the ModelTreeTraverser.
The final accumulator is returned.
Parameters:
Name | Type | Description | Default |
---|---|---|---|
model |
BaseModel |
The root model in which the traversal of the model tree starts. |
required |
acc |
TAcc |
The current accumulator. |
required |
Returns:
Type | Description |
---|---|
TAcc |
The accumulator after the traversal of the model tree. |
Source code in hydrolib/core/basemodel.py
def traverse(self, model: BaseModel, acc: TAcc) -> TAcc:
"""Traverse the model tree of BaseModels including the model as the root, with
the provided state of the acc and return the final accumulator.
The actual executed functions as well as the predicates defining whether these
functions should be executed for this model as well as whether child BaseModel
objects should be traversed are provided in the constructor of the
ModelTreeTraverser.
The final accumulator is returned.
Args:
model (BaseModel):
The root model in which the traversal of the model tree starts.
acc (TAcc):
The current accumulator.
Returns:
TAcc: The accumulator after the traversal of the model tree.
"""
if self._should_execute_pre(model, acc):
acc = self._pre_traverse_func(model, acc) # type: ignore[arg-type]
for _, value in model:
if not isinstance(value, list):
value = [value]
for v in value:
if self._should_traverse(v, acc):
acc = self.traverse(v, acc)
if self._should_execute_post(model, acc):
acc = self._post_traverse_func(model, acc) # type: ignore[arg-type]
return acc
ParsableFileModel (FileModel)
pydantic-model
¶
ParsableFileModel defines a FileModel which can be parsed and serialized with a serializer .
Each ParsableFileModel has a default _filename and _ext, which are used to generate the file name of any instance where the filepath is not (yet) set.
Children of the ParsableFileModel are expected to implement a serializer function which takes a Path and Dict and writes the ParsableFileModel to disk, and a parser function which takes a Path and outputs a Dict.
If more complicated solutions are required, a ParsableFileModel child can also opt to overwrite the _serialize and _parse methods, to skip the _get_serializer and _get_parser methods respectively.
ResolveRelativeMode (IntEnum)
¶
ResolveRelativeMode defines the possible resolve modes used within the FilePathResolver.
It determines how the relative paths to child models within some FileModel should be interpreted. By default it should be relative to the parent model, i.e. the model in which the mode is defined. However there exist some exceptions, where the relative paths should be evaluated relative to some other folder, regardless of the current parent location.
file_load_context()
¶
Provide a FileLoadingContext. If none has been created in the context of this call stack yet, a new one will be created, which will be maintained until it goes out of scope.
Yields
[FileLoadContext]: The file load context.
Source code in hydrolib/core/basemodel.py
@contextmanager
def file_load_context():
"""Provide a FileLoadingContext. If none has been created in the context of
this call stack yet, a new one will be created, which will be maintained
until it goes out of scope.
Yields:
[FileLoadContext]: The file load context.
"""
file_loading_context = context_file_loading.get(None)
context_reset_token = None
if not file_loading_context:
file_loading_context = FileLoadContext()
context_reset_token = context_file_loading.set(file_loading_context)
try:
yield file_loading_context
finally:
if context_reset_token is not None:
context_file_loading.reset(context_reset_token)
validator_set_default_disk_only_file_model_when_none() -> classmethod
¶
Validator to ensure a default empty DiskOnlyFileModel is created when the corresponding field is initialized with None.
Returns:
Type | Description |
---|---|
classmethod |
Validator to adjust None values to empty DiskOnlyFileModel objects |
Source code in hydrolib/core/basemodel.py
def validator_set_default_disk_only_file_model_when_none() -> classmethod:
"""Validator to ensure a default empty DiskOnlyFileModel is created
when the corresponding field is initialized with None.
Returns:
classmethod: Validator to adjust None values to empty DiskOnlyFileModel objects
"""
def adjust_none(v: Any, field: ModelField) -> Any:
if field.type_ is DiskOnlyFileModel and v is None:
return {"filepath": None}
return v
return validator("*", allow_reuse=True, pre=True)(adjust_none)