Io

`zenml.io` `special`

The io module handles file operations for the ZenML package. It offers a standard interface for reading, writing and manipulating files and directories. It is heavily influenced and inspired by the io module of tfx.

`utils`

`convert_to_str(path)`

Converts a PathType to a str using UTF-8.

Source code in zenml/io/utils.py

def convert_to_str(path: PathType) -> str:
    """Converts a PathType to a str using UTF-8."""
    if isinstance(path, str):
        return path
    else:
        return path.decode("utf-8")

`copy_dir(source_dir, destination_dir, overwrite=False)`

Copies dir from source to destination.

Parameters:

Name	Type	Description	Default
`source_dir`	`str`	Path to copy from.	required
`destination_dir`	`str`	Path to copy to.	required
`overwrite`	`bool`	Boolean. If false, function throws an error before overwrite.	`False`

Source code in zenml/io/utils.py

def copy_dir(
    source_dir: str, destination_dir: str, overwrite: bool = False
) -> None:
    """Copies dir from source to destination.

    Args:
        source_dir: Path to copy from.
        destination_dir: Path to copy to.
        overwrite: Boolean. If false, function throws an error before overwrite.
    """
    for source_file in listdir(source_dir):
        source_path = os.path.join(source_dir, convert_to_str(source_file))
        destination_path = os.path.join(
            destination_dir, convert_to_str(source_file)
        )
        if isdir(source_path):
            if source_path == destination_dir:
                # if the destination is a subdirectory of the source, we skip
                # copying it to avoid an infinite loop.
                return
            copy_dir(source_path, destination_path, overwrite)
        else:
            create_dir_recursive_if_not_exists(
                str(Path(destination_path).parent)
            )
            copy(str(source_path), str(destination_path), overwrite)

`create_dir_if_not_exists(dir_path)`

Creates directory if it does not exist.

Parameters:

Name	Type	Description	Default
`dir_path`	`str`	Local path in filesystem.	required

Source code in zenml/io/utils.py

def create_dir_if_not_exists(dir_path: str) -> None:
    """Creates directory if it does not exist.

    Args:
        dir_path: Local path in filesystem.
    """
    if not isdir(dir_path):
        mkdir(dir_path)

`create_dir_recursive_if_not_exists(dir_path)`

Creates directory recursively if it does not exist.

Parameters:

Name	Type	Description	Default
`dir_path`	`str`	Local path in filesystem.	required

Source code in zenml/io/utils.py

def create_dir_recursive_if_not_exists(dir_path: str) -> None:
    """Creates directory recursively if it does not exist.

    Args:
        dir_path: Local path in filesystem.
    """
    if not isdir(dir_path):
        makedirs(dir_path)

`create_file_if_not_exists(file_path, file_contents='{}')`

Creates file if it does not exist.

Parameters:

Name	Type	Description	Default
`file_path`	`str`	Local path in filesystem.	required
`file_contents`	`str`	Contents of file.	`'{}'`

Source code in zenml/io/utils.py

def create_file_if_not_exists(
    file_path: str, file_contents: str = "{}"
) -> None:
    """Creates file if it does not exist.

    Args:
        file_path: Local path in filesystem.
        file_contents: Contents of file.

    """
    full_path = Path(file_path)
    if not exists(file_path):
        create_dir_recursive_if_not_exists(str(full_path.parent))
        with open(str(full_path), "w") as f:
            f.write(file_contents)

`find_files(dir_path, pattern)`

Find files in a directory that match pattern.

Parameters:

Name	Type	Description	Default
`dir_path`	`Union[bytes, str]`	Path to directory.	required
`pattern`	`str`	pattern like *.png.	required

Yields:

Type	Description
`Iterable[str]`	All matching filenames if found.

Source code in zenml/io/utils.py

def find_files(dir_path: PathType, pattern: str) -> Iterable[str]:
    """Find files in a directory that match pattern.

    Args:
        dir_path: Path to directory.
        pattern: pattern like *.png.

    Yields:
         All matching filenames if found.
    """
    for root, dirs, files in walk(dir_path):
        for basename in files:
            if fnmatch.fnmatch(convert_to_str(basename), pattern):
                filename = os.path.join(
                    convert_to_str(root), convert_to_str(basename)
                )
                yield filename

`get_global_config_directory()`

Returns the global config directory for ZenML.

Source code in zenml/io/utils.py

def get_global_config_directory() -> str:
    """Returns the global config directory for ZenML."""
    env_var_path = os.getenv(ENV_ZENML_CONFIG_PATH)
    if env_var_path:
        return str(Path(env_var_path).resolve())
    return click.get_app_dir(APP_NAME)

`get_grandparent(dir_path)`

Get grandparent of dir.

Parameters:

Name	Type	Description	Default
`dir_path`	`str`	Path to directory.	required

Returns:

Type	Description
`str`	The input path's parent's parent.

Source code in zenml/io/utils.py

def get_grandparent(dir_path: str) -> str:
    """Get grandparent of dir.

    Args:
        dir_path: Path to directory.

    Returns:
        The input path's parent's parent.
    """
    return Path(dir_path).parent.parent.stem

`get_parent(dir_path)`

Get parent of dir.

Parameters:

Name	Type	Description	Default
`dir_path`	`str`	Path to directory.	required

Returns:

Type	Description
`str`	Parent (stem) of the dir as a string.

Source code in zenml/io/utils.py

def get_parent(dir_path: str) -> str:
    """Get parent of dir.

    Args:
        dir_path: Path to directory.

    Returns:
        Parent (stem) of the dir as a string.
    """
    return Path(dir_path).parent.stem

`is_remote(path)`

Returns True if path exists remotely.

Parameters:

Name	Type	Description	Default
`path`	`str`	Any path as a string.	required

Returns:

Type	Description
`bool`	True if remote path, else False.

Source code in zenml/io/utils.py

def is_remote(path: str) -> bool:
    """Returns True if path exists remotely.

    Args:
        path: Any path as a string.

    Returns:
        True if remote path, else False.
    """
    return any(path.startswith(prefix) for prefix in REMOTE_FS_PREFIX)

`is_root(path)`

Returns true if path has no parent in local filesystem.

Parameters:

Name	Type	Description	Default
`path`	`str`	Local path in filesystem.	required

Returns:

Type	Description
`bool`	True if root, else False.

Source code in zenml/io/utils.py

def is_root(path: str) -> bool:
    """Returns true if path has no parent in local filesystem.

    Args:
        path: Local path in filesystem.

    Returns:
        True if root, else False.
    """
    return Path(path).parent == Path(path)

`read_file_contents_as_string(file_path)`

Reads contents of file.

Parameters:

Name	Type	Description	Default
`file_path`	`str`	Path to file.	required

Source code in zenml/io/utils.py

def read_file_contents_as_string(file_path: str) -> str:
    """Reads contents of file.

    Args:
        file_path: Path to file.
    """
    if not exists(file_path):
        raise FileNotFoundError(f"{file_path} does not exist!")
    return open(file_path).read()  # type: ignore[no-any-return]

`resolve_relative_path(path)`

Takes relative path and resolves it absolutely.

Parameters:

Name	Type	Description	Default
`path`	`str`	Local path in filesystem.	required

Returns:

Type	Description
`str`	Resolved path.

Source code in zenml/io/utils.py

def resolve_relative_path(path: str) -> str:
    """Takes relative path and resolves it absolutely.

    Args:
      path: Local path in filesystem.

    Returns:
        Resolved path.
    """
    if is_remote(path):
        return path
    return str(Path(path).resolve())

`write_file_contents_as_string(file_path, content)`

Writes contents of file.

Parameters:

Name	Type	Description	Default
`file_path`	`str`	Path to file.	required
`content`	`str`	Contents of file.	required

Source code in zenml/io/utils.py

def write_file_contents_as_string(file_path: str, content: str) -> None:
    """Writes contents of file.

    Args:
        file_path: Path to file.
        content: Contents of file.
    """
    with open(file_path, "w") as f:
        f.write(content)

Io

zenml.io special

utils

convert_to_str(path)

copy_dir(source_dir, destination_dir, overwrite=False)

create_dir_if_not_exists(dir_path)

create_dir_recursive_if_not_exists(dir_path)

create_file_if_not_exists(file_path, file_contents='{}')

find_files(dir_path, pattern)

get_global_config_directory()

get_grandparent(dir_path)

get_parent(dir_path)

is_remote(path)

is_root(path)

read_file_contents_as_string(file_path)

resolve_relative_path(path)

write_file_contents_as_string(file_path, content)

`zenml.io` `special`

`utils`

`convert_to_str(path)`

`copy_dir(source_dir, destination_dir, overwrite=False)`

`create_dir_if_not_exists(dir_path)`

`create_dir_recursive_if_not_exists(dir_path)`

`create_file_if_not_exists(file_path, file_contents='{}')`

`find_files(dir_path, pattern)`

`get_global_config_directory()`

`get_grandparent(dir_path)`

`get_parent(dir_path)`

`is_remote(path)`

`is_root(path)`

`read_file_contents_as_string(file_path)`

`resolve_relative_path(path)`

`write_file_contents_as_string(file_path, content)`