IO Utilities

async_create_dir(create_dir, warn_mode=WarnEnum.DEBUG) async

Asynchronous create directory if create_dir doesn't exist.

Parameters:

Name	Type	Description	Default
create_dir	(str, required)	Create directory path.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Raises:

Type	Description
ValueError	If create_dir argument length is out of range.
OSError	When warning mode is set to ERROR and directory already exists.
OSError	If failed to create directory.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_create_dir(
    create_dir: str, warn_mode: WarnEnum | str = WarnEnum.DEBUG
) -> None:
    """Asynchronous create directory if `create_dir` doesn't exist.

    Args:
        create_dir (str           , required): Create directory path.
        warn_mode  (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.

    Raises:
        ValueError: If `create_dir` argument length is out of range.
        OSError   : When warning mode is set to ERROR and directory already exists.
        OSError   : If failed to create directory.
    """

    create_dir = create_dir.strip()
    if (len(create_dir) < 1) or (MAX_PATH_LENGTH < len(create_dir)):
        raise ValueError(
            f"`create_dir` argument length {len(create_dir)} is out of range, "
            f"must be between 1 and {MAX_PATH_LENGTH} characters!"
        )

    if isinstance(warn_mode, str):
        warn_mode = WarnEnum(warn_mode.strip().upper())

    if not await aiofiles.os.path.isdir(create_dir):
        try:
            _message = f"Creating '{create_dir}' directory..."
            if warn_mode == WarnEnum.ALWAYS:
                logger.info(_message)
            elif warn_mode == WarnEnum.DEBUG:
                logger.debug(_message)

            await aiofiles.os.makedirs(create_dir)
        except OSError as err:
            if (err.errno == errno.EEXIST) and (warn_mode == WarnEnum.DEBUG):
                logger.debug(f"'{create_dir}' directory already exists!")
            else:
                logger.error(f"Failed to create '{create_dir}' directory!")
                raise

        _message = f"Successfully created '{create_dir}' directory."
        if warn_mode == WarnEnum.ALWAYS:
            logger.info(_message)
        elif warn_mode == WarnEnum.DEBUG:
            logger.debug(_message)

    elif warn_mode == WarnEnum.ERROR:
        raise OSError(errno.EEXIST, f"'{create_dir}' directory already exists!")

    return

async_get_file_checksum(file_path, hash_method=HashAlgoEnum.md5, chunk_size=4096, warn_mode=WarnEnum.DEBUG) async

Asynchronous get file checksum.

Parameters:

Name	Type	Description	Default
file_path	(str, required)	Target file path.	required
hash_method	HashAlgoEnum	Hash method. Defaults to HashAlgoEnum.md5.	md5
chunk_size	int	Chunk size. Defaults to 4096.	4096
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Raises:

Type	Description
ValueError	If file_path argument length is out of range.
ValueError	If chunk_size argument value is invalid.
OSError	When warning mode is set to ERROR and file doesn't exist.

Returns:

Type	Description
str | None	str | None: File checksum or None if file doesn't exist.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_get_file_checksum(
    file_path: str,
    hash_method: HashAlgoEnum = HashAlgoEnum.md5,
    chunk_size: int = 4096,
    warn_mode: WarnEnum | str = WarnEnum.DEBUG,
) -> str | None:
    """Asynchronous get file checksum.

    Args:
        file_path   (str           , required): Target file path.
        hash_method (HashAlgoEnum  , optional): Hash method. Defaults to `HashAlgoEnum.md5`.
        chunk_size  (int           , optional): Chunk size. Defaults to 4096.
        warn_mode   (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                    Defaults to 'DEBUG'.

    Raises:
        ValueError: If `file_path` argument length is out of range.
        ValueError: If `chunk_size` argument value is invalid.
        OSError   : When warning mode is set to ERROR and file doesn't exist.

    Returns:
        str | None: File checksum or None if file doesn't exist.
    """

    file_path = file_path.strip()
    if (len(file_path) < 1) or (MAX_PATH_LENGTH < len(file_path)):
        raise ValueError(
            f"`file_path` argument length {len(file_path)} is out of range, "
            f"must be between 1 and {MAX_PATH_LENGTH} characters!"
        )

    if chunk_size < 10:
        raise ValueError(
            f"`chunk_size` argument value {chunk_size} is invalid, must be greater than 10!"
        )

    if isinstance(warn_mode, str):
        warn_mode = WarnEnum(warn_mode.strip().upper())

    _file_checksum: str | None = None
    if await aiofiles.os.path.isfile(file_path):
        _file_hash = hashlib.new(hash_method.value)
        async with aiofiles.open(file_path, "rb") as _file:
            while True:
                _file_chunk = await _file.read(chunk_size)
                if not _file_chunk:
                    break
                _file_hash.update(_file_chunk)

        _file_checksum = _file_hash.hexdigest()
    else:
        _message = f"'{file_path}' file doesn't exist!"
        if warn_mode == WarnEnum.ALWAYS:
            logger.warning(_message)
        elif warn_mode == WarnEnum.DEBUG:
            logger.debug(_message)
        elif warn_mode == WarnEnum.ERROR:
            raise OSError(errno.ENOENT, _message)

    return _file_checksum

async_read_config_file(config_path) async

Asynchronous read config file (YAML, JSON, TOML, INI).

Parameters:

Name	Type	Description	Default
config_path	(str | Path, required)	Config file path.	required

Raises:

Type	Description
FileNotFoundError	If config file is not found.
ValueError	If config file format is not supported.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Config file data as dictionary.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_read_config_file(config_path: str | Path) -> dict[str, Any]:
    """Asynchronous read config file (YAML, JSON, TOML, INI).

    Args:
        config_path (str | Path, required): Config file path.

    Raises:
        FileNotFoundError: If config file is not found.
        ValueError        : If config file format is not supported.

    Returns:
        dict[str, Any]: Config file data as dictionary.
    """

    _config: dict[str, Any] = {}

    if isinstance(config_path, str):
        config_path = Path(config_path)

    if not await aiofiles.os.path.isfile(config_path):
        raise FileNotFoundError(f"Not found '{config_path}' config file!")

    _suffix = config_path.suffix.lower()
    if _suffix in (".yaml", ".yml"):
        _config = await async_read_yaml_file(config_path)
    elif _suffix == ".json":
        _config = await async_read_json_file(config_path)
    elif _suffix == ".toml":
        _config = await async_read_toml_file(config_path)
    elif _suffix in (".ini", ".cfg"):
        _config = await async_read_ini_file(config_path)
    else:
        raise ValueError(
            f"Unsupported config file format '{_suffix}' for '{config_path}'!"
        )

    return _config

async_read_ini_file(file_path) async

Asynchronous read INI config file.

Parameters:

Name	Type	Description	Default
file_path	(str | Path, required)	INI config file path.	required

Raises:

Type	Description
FileNotFoundError	If INI config file is not found.
Exception	If failed to read INI config file.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: INI config file data as dictionary.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_read_ini_file(file_path: str | Path) -> dict[str, Any]:
    """Asynchronous read INI config file.

    Args:
        file_path (str | Path, required): INI config file path.

    Raises:
        FileNotFoundError: If INI config file is not found.
        Exception        : If failed to read INI config file.

    Returns:
        dict[str, Any]: INI config file data as dictionary.
    """

    _config: dict[str, Any] = {}

    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not await aiofiles.os.path.isfile(file_path):
        raise FileNotFoundError(f"Not found '{file_path}' INI config file!")

    try:
        _content: str = ""
        async with aiofiles.open(file_path, "r", encoding="utf-8") as _file:
            _content = await _file.read()

        _config_parser = configparser.ConfigParser()
        _config_parser.read_string(_content)
        for _section in _config_parser.sections():
            _config[_section] = dict(_config_parser.items(_section))

    except Exception:
        logger.error(f"Failed to read '{file_path}' INI config file!")
        raise

    return _config

async_read_json_file(file_path) async

Asynchronous read JSON file.

Parameters:

Name	Type	Description	Default
file_path	(str | Path, required)	JSON file path.	required

Raises:

Type	Description
FileNotFoundError	If JSON file is not found.
Exception	If failed to read JSON file.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: JSON file data as dictionary.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_read_json_file(file_path: str | Path) -> dict[str, Any]:
    """Asynchronous read JSON file.

    Args:
        file_path (str | Path, required): JSON file path.

    Raises:
        FileNotFoundError: If JSON file is not found.
        Exception        : If failed to read JSON file.

    Returns:
        dict[str, Any]: JSON file data as dictionary.
    """

    _data: dict[str, Any] = {}

    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not await aiofiles.os.path.isfile(file_path):
        raise FileNotFoundError(f"Not found '{file_path}' JSON file!")

    try:
        async with aiofiles.open(file_path, "r", encoding="utf-8") as _file:
            _content = await _file.read()
            _data = json.loads(_content) or {}
    except Exception:
        logger.error(f"Failed to read '{file_path}' JSON file!")
        raise

    return _data

async_read_toml_file(file_path) async

Asynchronous read TOML file.

Parameters:

Name	Type	Description	Default
file_path	(str | Path, required)	TOML file path.	required

Raises:

Type	Description
FileNotFoundError	If TOML file is not found.
Exception	If failed to read TOML file.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: TOML file data as dictionary.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_read_toml_file(file_path: str | Path) -> dict[str, Any]:
    """Asynchronous read TOML file.

    Args:
        file_path (str | Path, required): TOML file path.

    Raises:
        FileNotFoundError: If TOML file is not found.
        Exception        : If failed to read TOML file.

    Returns:
        dict[str, Any]: TOML file data as dictionary.
    """

    _data: dict[str, Any] = {}

    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not await aiofiles.os.path.isfile(file_path):
        raise FileNotFoundError(f"Not found '{file_path}' TOML file!")

    try:
        _content: str = ""
        if _binary_toml:
            async with aiofiles.open(file_path, "rb") as _file:
                _content = await _file.read()  # type: ignore
                _data = tomllib.load(_content) or {}
        else:
            async with aiofiles.open(file_path, "r", encoding="utf-8") as _file:
                _content = await _file.read()  # type: ignore
                _data = tomllib.loads(_content) or {}

    except Exception:
        logger.error(f"Failed to read '{file_path}' TOML file!")
        raise

    return _data

async_read_yaml_file(file_path) async

Asynchronous read YAML file.

Parameters:

Name	Type	Description	Default
file_path	(str | Path, required)	YAML file path.	required

Raises:

Type	Description
FileNotFoundError	If YAML file is not found.
Exception	If failed to read YAML file.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: YAML file data as dictionary.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_read_yaml_file(file_path: str | Path) -> dict[str, Any]:
    """Asynchronous read YAML file.

    Args:
        file_path (str | Path, required): YAML file path.

    Raises:
        FileNotFoundError: If YAML file is not found.
        Exception        : If failed to read YAML file.

    Returns:
        dict[str, Any]: YAML file data as dictionary.
    """

    _data: dict[str, Any] = {}

    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not await aiofiles.os.path.isfile(file_path):
        raise FileNotFoundError(f"Not found '{file_path}' YAML file!")

    try:
        async with aiofiles.open(file_path, "r", encoding="utf-8") as _file:
            _content = await _file.read()
            _data = yaml.safe_load(_content) or {}
    except Exception:
        logger.error(f"Failed to read '{file_path}' YAML file!")
        raise

    return _data

async_remove_dir(remove_dir, warn_mode=WarnEnum.DEBUG) async

Asynchronous remove directory if remove_dir exists.

Parameters:

Name	Type	Description	Default
remove_dir	(str, required)	Remove directory path.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Raises:

Type	Description
ValueError	If remove_dir argument length is out of range.
OSError	When warning mode is set to ERROR and directory doesn't exist.
OSError	If failed to remove directory.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_remove_dir(
    remove_dir: str, warn_mode: WarnEnum | str = WarnEnum.DEBUG
) -> None:
    """Asynchronous remove directory if `remove_dir` exists.

    Args:
        remove_dir (str           , required): Remove directory path.
        warn_mode  (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.

    Raises:
        ValueError: If `remove_dir` argument length is out of range.
        OSError   : When warning mode is set to ERROR and directory doesn't exist.
        OSError   : If failed to remove directory.
    """

    remove_dir = remove_dir.strip()
    if (len(remove_dir) < 1) or (MAX_PATH_LENGTH < len(remove_dir)):
        raise ValueError(
            f"`remove_dir` argument length {len(remove_dir)} is out of range, "
            f"must be between 1 and {MAX_PATH_LENGTH} characters!"
        )

    if isinstance(warn_mode, str):
        warn_mode = WarnEnum(warn_mode.strip().upper())

    if await aiofiles.os.path.isdir(remove_dir):
        try:
            _message = f"Removing '{remove_dir}' directory..."
            if warn_mode == WarnEnum.ALWAYS:
                logger.info(_message)
            elif warn_mode == WarnEnum.DEBUG:
                logger.debug(_message)

            await aioshutil.rmtree(remove_dir)
        except OSError as err:
            if (err.errno == errno.ENOENT) and (warn_mode == WarnEnum.DEBUG):
                logger.debug(f"'{remove_dir}' directory doesn't exist!")
            else:
                logger.error(f"Failed to remove '{remove_dir}' directory!")
                raise

        _message = f"Successfully removed '{remove_dir}' directory."
        if warn_mode == WarnEnum.ALWAYS:
            logger.info(_message)
        elif warn_mode == WarnEnum.DEBUG:
            logger.debug(_message)

    elif warn_mode == WarnEnum.ERROR:
        raise OSError(errno.ENOENT, f"'{remove_dir}' directory doesn't exist!")

    return

async_remove_dirs(remove_dirs, warn_mode=WarnEnum.DEBUG) async

Asynchronous remove directories if remove_dirs exists.

Parameters:

Name	Type	Description	Default
remove_dirs	(list[str], required)	Remove directories paths as list.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Source code in src/potato_util/io/_async.py

@validate_call
async def async_remove_dirs(
    remove_dirs: list[str], warn_mode: WarnEnum | str = WarnEnum.DEBUG
) -> None:
    """Asynchronous remove directories if `remove_dirs` exists.

    Args:
        remove_dirs (list[str]     , required): Remove directories paths as list.
        warn_mode   (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                    Defaults to 'DEBUG'.
    """

    for _remove_dir in remove_dirs:
        await async_remove_dir(remove_dir=_remove_dir, warn_mode=warn_mode)

    return

async_remove_file(file_path, warn_mode=WarnEnum.DEBUG) async

Asynchronous remove file if file_path exists.

Parameters:

Name	Type	Description	Default
file_path	(str, required)	Remove file path.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Raises:

Type	Description
ValueError	If file_path argument length is out of range.
OSError	When warning mode is set to ERROR and file doesn't exist.
OSError	If failed to remove file.

Source code in src/potato_util/io/_async.py

@validate_call
async def async_remove_file(
    file_path: str, warn_mode: WarnEnum | str = WarnEnum.DEBUG
) -> None:
    """Asynchronous remove file if `file_path` exists.

    Args:
        file_path (str           , required): Remove file path.
        warn_mode (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.

    Raises:
        ValueError: If `file_path` argument length is out of range.
        OSError   : When warning mode is set to ERROR and file doesn't exist.
        OSError   : If failed to remove file.
    """

    file_path = file_path.strip()
    if (len(file_path) < 1) or (MAX_PATH_LENGTH < len(file_path)):
        raise ValueError(
            f"`file_path` argument length {len(file_path)} is out of range, "
            f"must be between 1 and {MAX_PATH_LENGTH} characters!"
        )

    if isinstance(warn_mode, str):
        warn_mode = WarnEnum(warn_mode.strip().upper())

    if await aiofiles.os.path.isfile(file_path):
        try:
            _message = f"Removing '{file_path}' file..."
            if warn_mode == WarnEnum.ALWAYS:
                logger.info(_message)
            elif warn_mode == WarnEnum.DEBUG:
                logger.debug(_message)

            await aiofiles.os.remove(file_path)
        except OSError as err:
            if (err.errno == errno.ENOENT) and (warn_mode == WarnEnum.DEBUG):
                logger.debug(f"'{file_path}' file doesn't exist!")
            else:
                logger.error(f"Failed to remove '{file_path}' file!")
                raise

        _message = f"Successfully removed '{file_path}' file."
        if warn_mode == WarnEnum.ALWAYS:
            logger.info(_message)
        elif warn_mode == WarnEnum.DEBUG:
            logger.debug(_message)

    elif warn_mode == WarnEnum.ERROR:
        raise OSError(errno.ENOENT, f"'{file_path}' file doesn't exist!")

    return

async_remove_files(file_paths, warn_mode=WarnEnum.DEBUG) async

Asynchronous remove files if file_paths exists.

Parameters:

Name	Type	Description	Default
file_paths	(list[str], required)	Remove file paths as list.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Source code in src/potato_util/io/_async.py

@validate_call
async def async_remove_files(
    file_paths: list[str], warn_mode: WarnEnum | str = WarnEnum.DEBUG
) -> None:
    """Asynchronous remove files if `file_paths` exists.

    Args:
        file_paths (list[str]     , required): Remove file paths as list.
        warn_mode  (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.
    """

    for _file_path in file_paths:
        await async_remove_file(file_path=_file_path, warn_mode=warn_mode)

    return

create_dir(create_dir, warn_mode=WarnEnum.DEBUG)

Create directory if create_dir doesn't exist.

Parameters:

Name	Type	Description	Default
create_dir	(str, required)	Create directory path.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Raises:

Type	Description
ValueError	If create_dir argument length is out of range.
OSError	When warning mode is set to ERROR and directory already exists.
OSError	If failed to create directory.

Source code in src/potato_util/io/_sync.py

@validate_call
def create_dir(create_dir: str, warn_mode: WarnEnum | str = WarnEnum.DEBUG) -> None:
    """Create directory if `create_dir` doesn't exist.

    Args:
        create_dir (str           , required): Create directory path.
        warn_mode  (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.

    Raises:
        ValueError: If `create_dir` argument length is out of range.
        OSError   : When warning mode is set to ERROR and directory already exists.
        OSError   : If failed to create directory.
    """

    create_dir = create_dir.strip()
    if (len(create_dir) < 1) or (MAX_PATH_LENGTH < len(create_dir)):
        raise ValueError(
            f"`create_dir` argument length {len(create_dir)} is out of range, "
            f"must be between 1 and {MAX_PATH_LENGTH} characters!"
        )

    if isinstance(warn_mode, str):
        warn_mode = WarnEnum(warn_mode.strip().upper())

    if not os.path.isdir(create_dir):
        try:
            _message = f"Creating '{create_dir}' directory..."
            if warn_mode == WarnEnum.ALWAYS:
                logger.info(_message)
            elif warn_mode == WarnEnum.DEBUG:
                logger.debug(_message)

            os.makedirs(create_dir)
        except OSError as err:
            if (err.errno == errno.EEXIST) and (warn_mode == WarnEnum.DEBUG):
                logger.debug(f"'{create_dir}' directory already exists!")
            else:
                logger.error(f"Failed to create '{create_dir}' directory!")
                raise

        _message = f"Successfully created '{create_dir}' directory."
        if warn_mode == WarnEnum.ALWAYS:
            logger.info(_message)
        elif warn_mode == WarnEnum.DEBUG:
            logger.debug(_message)

    elif warn_mode == WarnEnum.ERROR:
        raise OSError(errno.EEXIST, f"'{create_dir}' directory already exists!")

    return

get_file_checksum(file_path, hash_method=HashAlgoEnum.md5, chunk_size=4096, warn_mode=WarnEnum.DEBUG)

Get file checksum.

Parameters:

Name	Type	Description	Default
file_path	(str, required)	Target file path.	required
hash_method	HashAlgoEnum	Hash method. Defaults to HashAlgoEnum.md5.	md5
chunk_size	int	Chunk size. Defaults to 4096.	4096
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Raises:

Type	Description
ValueError	If file_path argument length is out of range.
ValueError	If chunk_size argument value is invalid.
OSError	When warning mode is set to ERROR and file doesn't exist.

Returns:

Type	Description
str | None	str | None: File checksum or None if file doesn't exist.

Source code in src/potato_util/io/_sync.py

@validate_call
def get_file_checksum(
    file_path: str,
    hash_method: HashAlgoEnum = HashAlgoEnum.md5,
    chunk_size: int = 4096,
    warn_mode: WarnEnum | str = WarnEnum.DEBUG,
) -> str | None:
    """Get file checksum.

    Args:
        file_path   (str           , required): Target file path.
        hash_method (HashAlgoEnum  , optional): Hash method. Defaults to `HashAlgoEnum.md5`.
        chunk_size  (int           , optional): Chunk size. Defaults to 4096.
        warn_mode   (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                    Defaults to 'DEBUG'.

    Raises:
        ValueError: If `file_path` argument length is out of range.
        ValueError: If `chunk_size` argument value is invalid.
        OSError   : When warning mode is set to ERROR and file doesn't exist.

    Returns:
        str | None: File checksum or None if file doesn't exist.
    """

    file_path = file_path.strip()
    if (len(file_path) < 1) or (MAX_PATH_LENGTH < len(file_path)):
        raise ValueError(
            f"`file_path` argument length {len(file_path)} is out of range, "
            f"must be between 1 and {MAX_PATH_LENGTH} characters!"
        )

    if chunk_size < 10:
        raise ValueError(
            f"`chunk_size` argument value {chunk_size} is invalid, must be greater than 10!"
        )

    if isinstance(warn_mode, str):
        warn_mode = WarnEnum(warn_mode.strip().upper())

    _file_checksum: str | None = None
    if os.path.isfile(file_path):
        _file_hash = hashlib.new(hash_method.value)
        with open(file_path, "rb") as _file:
            while True:
                _file_chunk = _file.read(chunk_size)
                if not _file_chunk:
                    break
                _file_hash.update(_file_chunk)

        _file_checksum = _file_hash.hexdigest()
    else:
        _message = f"'{file_path}' file doesn't exist!"
        if warn_mode == WarnEnum.ALWAYS:
            logger.warning(_message)
        elif warn_mode == WarnEnum.DEBUG:
            logger.debug(_message)
        elif warn_mode == WarnEnum.ERROR:
            raise OSError(errno.ENOENT, _message)

    return _file_checksum

read_config_file(config_path)

Read config file (YAML, JSON, TOML, INI).

Parameters:

Name	Type	Description	Default
config_path	(str | Path, required)	Config file path.	required

Raises:

Type	Description
FileNotFoundError	If config file is not found.
ValueError	If config file format is not supported.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: Config file data as dictionary.

Source code in src/potato_util/io/_sync.py

@validate_call
def read_config_file(config_path: str | Path) -> dict[str, Any]:
    """Read config file (YAML, JSON, TOML, INI).

    Args:
        config_path (str | Path, required): Config file path.

    Raises:
        FileNotFoundError: If config file is not found.
        ValueError       : If config file format is not supported.

    Returns:
        dict[str, Any]: Config file data as dictionary.
    """

    _config: dict[str, Any] = {}

    if isinstance(config_path, str):
        config_path = Path(config_path)

    if not os.path.isfile(config_path):
        raise FileNotFoundError(f"Not found '{config_path}' config file!")

    _suffix = config_path.suffix.lower()
    if _suffix in (".yaml", ".yml"):
        _config = read_yaml_file(file_path=config_path)
    elif _suffix == ".json":
        _config = read_json_file(file_path=config_path)
    elif _suffix == ".toml":
        _config = read_toml_file(file_path=config_path)
    elif _suffix in (".ini", ".cfg"):
        _config = read_ini_file(file_path=config_path)
    else:
        raise ValueError(
            f"Unsupported config file format '{config_path.suffix}' for '{config_path}'!"
        )

    return _config

read_ini_file(file_path)

Read INI config file.

Parameters:

Name	Type	Description	Default
file_path	(str | Path, required)	INI config file path.	required

Raises:

Type	Description
FileNotFoundError	If INI config file is not found.
Exception	If failed to read INI config file.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: INI config file data as dictionary.

Source code in src/potato_util/io/_sync.py

@validate_call
def read_ini_file(file_path: str | Path) -> dict[str, Any]:
    """Read INI config file.

    Args:
        file_path (str | Path, required): INI config file path.

    Raises:
        FileNotFoundError: If INI config file is not found.
        Exception        : If failed to read INI config file.

    Returns:
        dict[str, Any]: INI config file data as dictionary.
    """

    _config: dict[str, Any] = {}

    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not os.path.isfile(file_path):
        raise FileNotFoundError(f"Not found '{file_path}' INI config file!")

    try:
        _config_parser = configparser.ConfigParser()
        _config_parser.read(file_path)
        for _section in _config_parser.sections():
            _config[_section] = dict(_config_parser.items(_section))

    except Exception:
        logger.error(f"Failed to read '{file_path}' INI config file!")
        raise

    return _config

read_json_file(file_path)

Read JSON file.

Parameters:

Name	Type	Description	Default
file_path	(str | Path, required)	JSON file path.	required

Raises:

Type	Description
FileNotFoundError	If JSON file is not found.
Exception	If failed to read JSON file.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: JSON file data as dictionary.

Source code in src/potato_util/io/_sync.py

@validate_call
def read_json_file(file_path: str | Path) -> dict[str, Any]:
    """Read JSON file.

    Args:
        file_path (str | Path, required): JSON file path.

    Raises:
        FileNotFoundError: If JSON file is not found.
        Exception        : If failed to read JSON file.

    Returns:
        dict[str, Any]: JSON file data as dictionary.
    """

    _data: dict[str, Any] = {}

    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not os.path.isfile(file_path):
        raise FileNotFoundError(f"Not found '{file_path}' JSON file!")

    try:
        with open(file_path, encoding="utf-8") as _file:
            _data = json.load(_file) or {}
    except Exception:
        logger.error(f"Failed to read '{file_path}' JSON file!")
        raise

    return _data

read_toml_file(file_path)

Read TOML file.

Parameters:

Name	Type	Description	Default
file_path	(str | Path, required)	TOML file path.	required

Raises:

Type	Description
FileNotFoundError	If TOML file is not found.
Exception	If failed to read TOML file.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: TOML file data as dictionary.

Source code in src/potato_util/io/_sync.py

@validate_call
def read_toml_file(file_path: str | Path) -> dict[str, Any]:
    """Read TOML file.

    Args:
        file_path (str | Path, required): TOML file path.

    Raises:
        FileNotFoundError: If TOML file is not found.
        Exception        : If failed to read TOML file.

    Returns:
        dict[str, Any]: TOML file data as dictionary.
    """

    _data: dict[str, Any] = {}

    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not os.path.isfile(file_path):
        raise FileNotFoundError(f"Not found '{file_path}' TOML file!")

    try:
        if _binary_toml:
            with open(file_path, "rb") as _file:
                _data = tomllib.load(_file) or {}  # type: ignore
        else:
            with open(file_path, encoding="utf-8") as _file:
                _data = tomllib.load(_file) or {}  # type: ignore
    except Exception:
        logger.error(f"Failed to read '{file_path}' TOML file!")
        raise

    return _data

read_yaml_file(file_path)

Read YAML file.

Parameters:

Name	Type	Description	Default
file_path	(str | Path, required)	YAML file path.	required

Raises:

Type	Description
FileNotFoundError	If YAML file is not found.
Exception	If failed to read YAML file.

Returns:

Type	Description
dict[str, Any]	dict[str, Any]: YAML file data as dictionary.

Source code in src/potato_util/io/_sync.py

@validate_call
def read_yaml_file(file_path: str | Path) -> dict[str, Any]:
    """Read YAML file.

    Args:
        file_path (str | Path, required): YAML file path.

    Raises:
        FileNotFoundError: If YAML file is not found.
        Exception        : If failed to read YAML file.

    Returns:
        dict[str, Any]: YAML file data as dictionary.
    """

    _data: dict[str, Any] = {}

    if isinstance(file_path, str):
        file_path = Path(file_path)

    if not os.path.isfile(file_path):
        raise FileNotFoundError(f"Not found '{file_path}' YAML file!")

    try:
        with open(file_path, encoding="utf-8") as _file:
            _data = yaml.safe_load(_file) or {}
    except Exception:
        logger.error(f"Failed to read '{file_path}' YAML file!")
        raise

    return _data

remove_dir(remove_dir, warn_mode=WarnEnum.DEBUG)

Remove directory if remove_dir exists.

Parameters:

Name	Type	Description	Default
remove_dir	(str, required)	Remove directory path.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Raises:

Type	Description
ValueError	If remove_dir argument length is out of range.
OSError	When warning mode is set to ERROR and directory doesn't exist.
OSError	If failed to remove directory.

Source code in src/potato_util/io/_sync.py

@validate_call
def remove_dir(remove_dir: str, warn_mode: WarnEnum | str = WarnEnum.DEBUG) -> None:
    """Remove directory if `remove_dir` exists.

    Args:
        remove_dir (str           , required): Remove directory path.
        warn_mode  (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.

    Raises:
        ValueError: If `remove_dir` argument length is out of range.
        OSError   : When warning mode is set to ERROR and directory doesn't exist.
        OSError   : If failed to remove directory.
    """

    remove_dir = remove_dir.strip()
    if (len(remove_dir) < 1) or (MAX_PATH_LENGTH < len(remove_dir)):
        raise ValueError(
            f"`remove_dir` argument length {len(remove_dir)} is out of range, "
            f"must be between 1 and {MAX_PATH_LENGTH} characters!"
        )

    if isinstance(warn_mode, str):
        warn_mode = WarnEnum(warn_mode.strip().upper())

    if os.path.isdir(remove_dir):
        try:
            _message = f"Removing '{remove_dir}' directory..."
            if warn_mode == WarnEnum.ALWAYS:
                logger.info(_message)
            elif warn_mode == WarnEnum.DEBUG:
                logger.debug(_message)

            shutil.rmtree(remove_dir)
        except OSError as err:
            if (err.errno == errno.ENOENT) and (warn_mode == WarnEnum.DEBUG):
                logger.debug(f"'{remove_dir}' directory doesn't exist!")
            else:
                logger.error(f"Failed to remove '{remove_dir}' directory!")
                raise

        _message = f"Successfully removed '{remove_dir}' directory."
        if warn_mode == WarnEnum.ALWAYS:
            logger.info(_message)
        elif warn_mode == WarnEnum.DEBUG:
            logger.debug(_message)

    elif warn_mode == WarnEnum.ERROR:
        raise OSError(errno.ENOENT, f"'{remove_dir}' directory doesn't exist!")

    return

remove_dirs(remove_dirs, warn_mode=WarnEnum.DEBUG)

Remove directories if remove_dirs exist.

Parameters:

Name	Type	Description	Default
remove_dirs	(list[str], required)	Remove directory paths as list.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Source code in src/potato_util/io/_sync.py

@validate_call
def remove_dirs(
    remove_dirs: list[str], warn_mode: WarnEnum | str = WarnEnum.DEBUG
) -> None:
    """Remove directories if `remove_dirs` exist.

    Args:
        remove_dirs (list[str]     , required): Remove directory paths as list.
        warn_mode   (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.
    """

    for _remove_dir in remove_dirs:
        remove_dir(remove_dir=_remove_dir, warn_mode=warn_mode)

    return

remove_file(file_path, warn_mode=WarnEnum.DEBUG)

Remove file if file_path exists.

Parameters:

Name	Type	Description	Default
file_path	(str, required)	Remove file path.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Raises:

Type	Description
ValueError	If file_path argument length is out of range.
OSError	When warning mode is set to ERROR and file doesn't exist.
OSError	If failed to remove file.

Source code in src/potato_util/io/_sync.py

@validate_call
def remove_file(file_path: str, warn_mode: WarnEnum | str = WarnEnum.DEBUG) -> None:
    """Remove file if `file_path` exists.

    Args:
        file_path (str           , required): Remove file path.
        warn_mode (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.

    Raises:
        ValueError: If `file_path` argument length is out of range.
        OSError   : When warning mode is set to ERROR and file doesn't exist.
        OSError   : If failed to remove file.
    """

    file_path = file_path.strip()
    if (len(file_path) < 1) or (MAX_PATH_LENGTH < len(file_path)):
        raise ValueError(
            f"`file_path` argument length {len(file_path)} is out of range, "
            f"must be between 1 and {MAX_PATH_LENGTH} characters!"
        )

    if isinstance(warn_mode, str):
        warn_mode = WarnEnum(warn_mode.strip().upper())

    if os.path.isfile(file_path):
        try:
            _message = f"Removing '{file_path}' file..."
            if warn_mode == WarnEnum.ALWAYS:
                logger.info(_message)
            elif warn_mode == WarnEnum.DEBUG:
                logger.debug(_message)

            os.remove(file_path)
        except OSError as err:
            if (err.errno == errno.ENOENT) and (warn_mode == WarnEnum.DEBUG):
                logger.debug(f"'{file_path}' file doesn't exist!")
            else:
                logger.error(f"Failed to remove '{file_path}' file!")
                raise

        _message = f"Successfully removed '{file_path}' file."
        if warn_mode == WarnEnum.ALWAYS:
            logger.info(_message)
        elif warn_mode == WarnEnum.DEBUG:
            logger.debug(_message)

    elif warn_mode == WarnEnum.ERROR:
        raise OSError(errno.ENOENT, f"'{file_path}' file doesn't exist!")

    return

remove_files(file_paths, warn_mode=WarnEnum.DEBUG)

Remove files if file_paths exist.

Parameters:

Name	Type	Description	Default
file_paths	(list[str], required)	Remove file paths as list.	required
warn_mode	WarnEnum | str	Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'. Defaults to 'DEBUG'.	DEBUG

Source code in src/potato_util/io/_sync.py

@validate_call
def remove_files(
    file_paths: list[str], warn_mode: WarnEnum | str = WarnEnum.DEBUG
) -> None:
    """Remove files if `file_paths` exist.

    Args:
        file_paths (list[str]     , required): Remove file paths as list.
        warn_mode  (WarnEnum | str, optional): Warning message mode, for example: 'ERROR', 'ALWAYS', 'DEBUG', 'IGNORE'.
                                                Defaults to 'DEBUG'.
    """

    for _file_path in file_paths:
        remove_file(file_path=_file_path, warn_mode=warn_mode)

    return