fs

CompletedCommand ¶

CompletedCommand(args: str | list[str], returncode: int, time_taken: float, stdout: str | bytes | None = None, stderr: str | bytes | None = None)

Bases: CompletedProcess

Source code in stdl/fs.py

def __init__(
    self,
    args: str | list[str],
    returncode: int,
    time_taken: float,
    stdout: str | bytes | None = None,
    stderr: str | bytes | None = None,
) -> None:
    super().__init__(args, returncode, stdout, stderr)

    self.time_taken = time_taken

stdout_lines `property` ¶

stdout_lines: list[str]

Returns the stdout as a list of lines.

stderr_lines `property` ¶

stderr_lines: list[str]

Returns the stderr as a list of lines.

PathBase ¶

PathBase(path: str | PathLike, *, abs: bool = False)

Bases: PathLike

Base class for typed filesystem paths.

Return-value policy: - methods that keep the same lexical path return self - methods that point at a different path return a new path object - methods that create a second path return the created path object

Initialize a PathBase object.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike`	The path.	required
`abs`	`bool`	Whether to use the absolute path.	`False`

Source code in stdl/fs.py

def __init__(
    self,
    path: str | PathLike,
    *,
    abs: bool = False,  # noqa: A002
) -> None:
    """
    Initialize a PathBase object.

    Args:
        path (str | PathLike): The path.
        abs (bool): Whether to use the absolute path.
    """
    self._path = os.fspath(path).replace("\\", SEP).replace("/", SEP)
    if abs:
        self._path = os.path.abspath(self._path)

path `property` ¶

path: str

The stored path string.

nodes `property` ¶

nodes: list[str]

The path as a list of nodes.

created `property` ¶

created: float

The time when the path was created as a UNIX timestamp.

modified `property` ¶

modified: float

The time when the path was last modified as a UNIX timestamp.

accessed `property` ¶

accessed: float

The time when the path was last accessed as a UNIX timestamp.

basename `property` ¶

basename: str

The base name of the path (without the parent directory).

abspath `property` ¶

abspath: str

The absolute path.

parents `property` ¶

parents: tuple[Directory, ...]

All parent directories from immediate parent to root.

is_absolute `property` ¶

is_absolute: bool

Whether the path is absolute.

is_symlink `property` ¶

is_symlink: bool

Whether the path is a symbolic link.

exists `property` ¶

exists: bool

Check if the path exists.

parent `property` ¶

parent: Directory

The parent directory.

resolve ¶

resolve(strict: bool = False) -> Self

Make the path absolute, resolving any symlinks.

Parameters:

Name	Type	Description	Default
`strict`	`bool`	If True and path doesn't exist, raises FileNotFoundError. Defaults to False.	`False`

Returns:

Type	Description
`Self`	The resolved path object.

Source code in stdl/fs.py

def resolve(self, strict: bool = False) -> Self:
    """
    Make the path absolute, resolving any symlinks.

    Args:
        strict: If True and path doesn't exist, raises FileNotFoundError. Defaults to False.

    Returns:
        The resolved path object.
    """
    resolved_path = os.path.realpath(self.path)
    if strict and not os.path.exists(resolved_path):
        raise FileNotFoundError(f"No such file: '{resolved_path}'")

    return _clone_with_path(self, resolved_path)

resolve_async `async` ¶

resolve_async(strict: bool = False) -> Self

Async version of resolve().

Source code in stdl/fs.py

async def resolve_async(self, strict: bool = False) -> Self:
    """Async version of resolve()."""
    return await run_fs_sync(self.resolve, strict=strict)

created_async `async` ¶

created_async() -> float

Async version of created.

Source code in stdl/fs.py

async def created_async(self) -> float:
    """Async version of created."""
    return await run_fs_sync(lambda: self.created)

modified_async `async` ¶

modified_async() -> float

Async version of modified.

Source code in stdl/fs.py

async def modified_async(self) -> float:
    """Async version of modified."""
    return await run_fs_sync(lambda: self.modified)

accessed_async `async` ¶

accessed_async() -> float

Async version of accessed.

Source code in stdl/fs.py

async def accessed_async(self) -> float:
    """Async version of accessed."""
    return await run_fs_sync(lambda: self.accessed)

ctime_async `async` ¶

ctime_async() -> float

Async version of ctime.

Source code in stdl/fs.py

async def ctime_async(self) -> float:
    """Async version of ctime."""
    return await self.created_async()

mtime_async `async` ¶

mtime_async() -> float

Async version of mtime.

Source code in stdl/fs.py

async def mtime_async(self) -> float:
    """Async version of mtime."""
    return await self.modified_async()

atime_async `async` ¶

atime_async() -> float

Async version of atime.

Source code in stdl/fs.py

async def atime_async(self) -> float:
    """Async version of atime."""
    return await self.accessed_async()

is_symlink_async `async` ¶

is_symlink_async() -> bool

Async version of is_symlink.

Source code in stdl/fs.py

async def is_symlink_async(self) -> bool:
    """Async version of is_symlink."""
    return await run_fs_sync(lambda: self.is_symlink)

rename ¶

rename(name: str) -> Self

Rename the path.

Parameters:

Name	Type	Description	Default
`name`	`str`	The new name.	required

Source code in stdl/fs.py

def rename(self, name: str) -> Self:
    """
    Rename the path.

    Args:
        name (str): The new name.
    """
    new_path = os.path.join(self.parent.path, name)
    os.rename(self.path, new_path)

    return _clone_with_path(self, new_path)

rename_async `async` ¶

rename_async(name: str) -> Self

Async version of rename().

Source code in stdl/fs.py

async def rename_async(self, name: str) -> Self:
    """Async version of rename()."""
    return await _run_fs_sync_shielded(self.rename, name)

chmod ¶

chmod(mode: int) -> Self

Change the path's permissions.

Parameters:

Name	Type	Description	Default
`mode`	`int`	The new permissions mode.	required

Source code in stdl/fs.py

def chmod(self, mode: int) -> Self:
    """
    Change the path's permissions.

    Args:
        mode (int): The new permissions mode.
    """
    os.chmod(self.path, mode)

    return self

chmod_async `async` ¶

chmod_async(mode: int) -> Self

Async version of chmod().

Source code in stdl/fs.py

async def chmod_async(self, mode: int) -> Self:
    """Async version of chmod()."""
    return await _run_fs_sync_shielded(self.chmod, mode)

chown ¶

chown(user: str, group: str) -> Self

Change the path's owner and group.

Parameters:

Name	Type	Description	Default
`user`	`str`	The new owner.	required
`group`	`str`	The new group.	required

Source code in stdl/fs.py

def chown(self, user: str, group: str) -> Self:
    """
    Change the path's owner and group.

    Args:
        user (str): The new owner.
        group (str): The new group.
    """
    shutil.chown(self.path, user, group)

    return self

chown_async `async` ¶

chown_async(user: str, group: str) -> Self

Async version of chown().

Source code in stdl/fs.py

async def chown_async(self, user: str, group: str) -> Self:
    """Async version of chown()."""
    return await _run_fs_sync_shielded(self.chown, user, group)

should_exist ¶

should_exist() -> Self

Raise FileNotFoundError if the path does not exist.

Source code in stdl/fs.py

def should_exist(self) -> Self:
    """Raise FileNotFoundError if the path does not exist."""
    if not self.exists:
        raise FileNotFoundError(f"No such path: '{self.path}'")

    return self

should_exist_async `async` ¶

should_exist_async() -> Self

Async version of should_exist().

Source code in stdl/fs.py

async def should_exist_async(self) -> Self:
    """Async version of should_exist()."""
    return await run_fs_sync(self.should_exist)

should_not_exist ¶

should_not_exist() -> Self

Raise FileExistsError if the path exists.

Source code in stdl/fs.py

def should_not_exist(self) -> Self:
    """Raise FileExistsError if the path exists."""
    if self.exists:
        raise FileExistsError(f"Path already exists: '{self.path}'")

    return self

should_not_exist_async `async` ¶

should_not_exist_async() -> Self

Async version of should_not_exist().

Source code in stdl/fs.py

async def should_not_exist_async(self) -> Self:
    """Async version of should_not_exist()."""
    return await run_fs_sync(self.should_not_exist)

to_path ¶

to_path() -> Path

Convert to pathlib.Path.

Returns:

Name	Type	Description
`Path`	`Path`	The pathlib.Path object.

Source code in stdl/fs.py

def to_path(self) -> Path:
    """
    Convert to pathlib.Path.

    Returns:
        Path: The pathlib.Path object.
    """
    return Path(self.path)

to_str ¶

to_str() -> str

Convert to string.

Returns:

Name	Type	Description
`str`	`str`	The path as a string.

Source code in stdl/fs.py

def to_str(self) -> str:
    """
    Convert to string.

    Returns:
        str: The path as a string.
    """
    return str(self)

stat ¶

stat(*, follow_symlinks: bool = True) -> stat_result

Return the stat info for this path.

Parameters:

Name	Type	Description	Default
`follow_symlinks`	`bool`	If True, follow symlinks. If False, return info about the link itself.	`True`

Returns:

Type	Description
`stat_result`	os.stat_result: The stat information.

Source code in stdl/fs.py

def stat(self, *, follow_symlinks: bool = True) -> os.stat_result:
    """
    Return the stat info for this path.

    Args:
        follow_symlinks: If True, follow symlinks. If False, return info about the link itself.

    Returns:
        os.stat_result: The stat information.
    """
    return os.stat(self.path, follow_symlinks=follow_symlinks)

stat_async `async` ¶

stat_async(*, follow_symlinks: bool = True) -> stat_result

Async version of stat().

Source code in stdl/fs.py

async def stat_async(self, *, follow_symlinks: bool = True) -> os.stat_result:
    """Async version of stat()."""
    return await run_fs_sync(self.stat, follow_symlinks=follow_symlinks)

samefile ¶

samefile(other: str | PathLike | PathBase) -> bool

Return True if both paths reference the same filesystem entry.

Source code in stdl/fs.py

def samefile(self, other: str | PathLike | PathBase) -> bool:
    """Return True if both paths reference the same filesystem entry."""
    if isinstance(other, PathBase):
        other = other.path

    return os.path.samefile(self.path, os.fspath(other))

samefile_async `async` ¶

samefile_async(other: str | PathLike | PathBase) -> bool

Async version of samefile().

Source code in stdl/fs.py

async def samefile_async(self, other: str | PathLike | PathBase) -> bool:
    """Async version of samefile()."""
    return await run_fs_sync(self.samefile, other)

readlink ¶

readlink() -> str

Return the path that this symbolic link points to.

Source code in stdl/fs.py

def readlink(self) -> str:
    """Return the path that this symbolic link points to."""
    return os.readlink(self.path)

readlink_async `async` ¶

readlink_async() -> str

Async version of readlink().

Source code in stdl/fs.py

async def readlink_async(self) -> str:
    """Async version of readlink()."""
    return await run_fs_sync(self.readlink)

owner ¶

owner() -> str

Return the filesystem owner name.

Source code in stdl/fs.py

def owner(self) -> str:
    """Return the filesystem owner name."""
    if sys.platform == "win32":
        raise NotImplementedError("owner() is not supported on Windows")

    import pwd

    return pwd.getpwuid(self.stat().st_uid).pw_name

owner_async `async` ¶

owner_async() -> str

Async version of owner().

Source code in stdl/fs.py

async def owner_async(self) -> str:
    """Async version of owner()."""
    return await run_fs_sync(self.owner)

group ¶

group() -> str

Return the filesystem group name.

Source code in stdl/fs.py

def group(self) -> str:
    """Return the filesystem group name."""
    if sys.platform == "win32":
        raise NotImplementedError("group() is not supported on Windows")

    import grp

    return grp.getgrgid(self.stat().st_gid).gr_name

group_async `async` ¶

group_async() -> str

Async version of group().

Source code in stdl/fs.py

async def group_async(self) -> str:
    """Async version of group()."""
    return await run_fs_sync(self.group)

relative_to ¶

relative_to(other: str | PathBase) -> str

Make this path relative to another path.

Parameters:

Name	Type	Description	Default
`other`	`str \| PathBase`	The base path to make this path relative to.	required

Returns:

Name	Type	Description
`str`	`str`	The relative path.

Raises:

Type	Description
`ValueError`	If this path is not relative to the other path.

Source code in stdl/fs.py

def relative_to(self, other: str | PathBase) -> str:
    """
    Make this path relative to another path.

    Args:
        other: The base path to make this path relative to.

    Returns:
        str: The relative path.

    Raises:
        ValueError: If this path is not relative to the other path.
    """
    if isinstance(other, PathBase):
        other = other.path
    other = os.path.abspath(other)
    self_abs = os.path.abspath(self.path)
    if not self_abs.startswith(other.rstrip(SEP) + SEP) and self_abs != other:
        raise ValueError(f"'{self.path}' is not relative to '{other}'")

    return os.path.relpath(self_abs, other)

expanduser ¶

expanduser() -> Self

Expand ~ and ~user constructs in the path.

Returns:

Type	Description
`Self`	The path object with the expanded path.

Source code in stdl/fs.py

def expanduser(self) -> Self:
    """
    Expand ~ and ~user constructs in the path.

    Returns:
        The path object with the expanded path.
    """
    return _clone_with_path(self, os.path.expanduser(self.path))

as_posix ¶

as_posix() -> str

Return the path with forward slashes.

Source code in stdl/fs.py

def as_posix(self) -> str:
    """Return the path with forward slashes."""
    return self.path.replace("\\", "/")

as_uri ¶

as_uri() -> str

Return the path as a file URI.

Returns:

Name	Type	Description
`str`	`str`	The file URI (e.g., 'file:///path/to/file').

Raises:

Type	Description
`ValueError`	If the path is not absolute.

Source code in stdl/fs.py

def as_uri(self) -> str:
    """
    Return the path as a file URI.

    Returns:
        str: The file URI (e.g., 'file:///path/to/file').

    Raises:
        ValueError: If the path is not absolute.
    """
    if not self.is_absolute:
        raise ValueError("relative path can't be expressed as a file URI")

    return "file://" + pathname2url(os.path.abspath(self.path))

match ¶

match(pattern: str) -> bool

Check if the path matches the given glob pattern.

Parameters:

Name	Type	Description	Default
`pattern`	`str`	A glob pattern to match against.	required

Returns:

Name	Type	Description
`bool`	`bool`	True if the path matches the pattern.

Source code in stdl/fs.py

def match(self, pattern: str) -> bool:
    """
    Check if the path matches the given glob pattern.

    Args:
        pattern: A glob pattern to match against.

    Returns:
        bool: True if the path matches the pattern.
    """
    return fnmatch.fnmatch(self.basename, pattern)

with_name ¶

with_name(name: str) -> Self

Return a new path with the name changed.

Source code in stdl/fs.py

def with_name(self, name: str) -> Self:
    """Return a new path with the name changed."""
    parent = os.path.dirname(self.path)
    if parent:
        return _clone_with_path(self, os.path.join(parent, name))

    return _clone_with_path(self, name)

exists_async `async` ¶

exists_async() -> bool

Async version of exists.

Source code in stdl/fs.py

async def exists_async(self) -> bool:
    """Async version of exists."""
    return await run_fs_sync(lambda: self.exists)

size_async `async` ¶

size_async() -> int

Async version of size.

Source code in stdl/fs.py

async def size_async(self) -> int:
    """Async version of size."""
    return await run_fs_sync(lambda: self.size)

size_readable_async `async` ¶

size_readable_async() -> str

Async version of size_readable.

Source code in stdl/fs.py

async def size_readable_async(self) -> str:
    """Async version of size_readable."""
    return await run_fs_sync(lambda: self.size_readable)

remove_async `async` ¶

remove_async() -> Self

Async version of remove().

Source code in stdl/fs.py

async def remove_async(self) -> Self:
    """Async version of remove()."""
    return await _run_fs_sync_shielded(self.remove)

delete_async `async` ¶

delete_async() -> Self

Async version of delete().

Source code in stdl/fs.py

async def delete_async(self) -> Self:
    """Async version of delete()."""
    return await self.remove_async()

move_to ¶

move_to(target: PathBase, *, mkdir: bool = False, overwrite: bool = True) -> Self

Move the path to a new target.

Parameters:

Name	Type	Description	Default
`target`	`PathBase`	The destination path object.	required
`mkdir`	`bool`	Whether to create missing parent directories. Defaults to False.	`False`
`overwrite`	`bool`	Whether to overwrite files if they already exist. Defaults to True.	`True`

Returns:

Type	Description
`Self`	A new path object pointing at the moved path.

Source code in stdl/fs.py

def move_to(self, target: PathBase, *, mkdir: bool = False, overwrite: bool = True) -> Self:
    """
    Move the path to a new target.

    Args:
        target: The destination path object.
        mkdir: Whether to create missing parent directories. Defaults to False.
        overwrite: Whether to overwrite files if they already exist. Defaults to True.

    Returns:
        A new path object pointing at the moved path.
    """
    raise NotImplementedError

copy_to ¶

copy_to(target: PathBase, *, mkdir: bool = False, overwrite: bool = True) -> Self

Copy the path to a new target.

Parameters:

Name	Type	Description	Default
`target`	`PathBase`	The destination path object.	required
`mkdir`	`bool`	Whether to create missing parent directories. Defaults to False.	`False`
`overwrite`	`bool`	Whether to overwrite files if they already exist. Defaults to True.	`True`

Returns:

Type	Description
`Self`	A new path object pointing at the copied path.

Source code in stdl/fs.py

def copy_to(
    self,
    target: PathBase,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> Self:
    """
    Copy the path to a new target.

    Args:
        target: The destination path object.
        mkdir: Whether to create missing parent directories. Defaults to False.
        overwrite: Whether to overwrite files if they already exist. Defaults to True.

    Returns:
        A new path object pointing at the copied path.
    """
    raise NotImplementedError

get_xattr ¶

get_xattr(name: str, group: str = 'user') -> str

Retrieve the value of an extended attribute.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the extended attribute.	required
`group`	`str`	The group of the extended attribute. Defaults to "user".	`'user'`

Returns:

Type	Description
`str`	The value of the extended attribute.

Source code in stdl/fs.py

def get_xattr(self, name: str, group: str = "user") -> str:
    """
    Retrieve the value of an extended attribute.

    Args:
        name: The name of the extended attribute.
        group: The group of the extended attribute. Defaults to "user".

    Returns:
        The value of the extended attribute.
    """
    return os.getxattr(self.path, f"{group}.{name}").decode()

get_xattr_async `async` ¶

get_xattr_async(name: str, group: str = 'user') -> str

Async version of get_xattr().

Source code in stdl/fs.py

async def get_xattr_async(self, name: str, group: str = "user") -> str:
    """Async version of get_xattr()."""
    return await run_fs_sync(self.get_xattr, name, group)

set_xattr ¶

set_xattr(value: str | bytes, name: str, group: str = 'user') -> Self

Set an extended attribute.

Parameters:

Name	Type	Description	Default
`value`	`str \| bytes`	The value of the extended attribute.	required
`name`	`str`	The name of the extended attribute.	required
`group`	`str`	The group of the extended attribute. Defaults to "user".	`'user'`

Source code in stdl/fs.py

def set_xattr(self, value: str | bytes, name: str, group: str = "user") -> Self:
    """
    Set an extended attribute.

    Args:
        value: The value of the extended attribute.
        name: The name of the extended attribute.
        group: The group of the extended attribute. Defaults to "user".
    """
    if isinstance(value, str):
        value = value.encode()
    os.setxattr(self.path, f"{group}.{name}", value)

    return self

set_xattr_async `async` ¶

set_xattr_async(value: str | bytes, name: str, group: str = 'user') -> Self

Async version of set_xattr().

Source code in stdl/fs.py

async def set_xattr_async(self, value: str | bytes, name: str, group: str = "user") -> Self:
    """Async version of set_xattr()."""
    return await _run_fs_sync_shielded(self.set_xattr, value, name, group)

remove_xattr ¶

remove_xattr(name: str, group: str = 'user') -> Self

Remove an extended attribute.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the extended attribute.	required
`group`	`str`	The group of the extended attribute. Defaults to "user".	`'user'`

Source code in stdl/fs.py

def remove_xattr(self, name: str, group: str = "user") -> Self:
    """
    Remove an extended attribute.

    Args:
        name: The name of the extended attribute.
        group: The group of the extended attribute. Defaults to "user".
    """
    os.removexattr(self.path, f"{group}.{name}")

    return self

remove_xattr_async `async` ¶

remove_xattr_async(name: str, group: str = 'user') -> Self

Async version of remove_xattr().

Source code in stdl/fs.py

async def remove_xattr_async(self, name: str, group: str = "user") -> Self:
    """Async version of remove_xattr()."""
    return await _run_fs_sync_shielded(self.remove_xattr, name, group)

File ¶

File(path: str | PathLike, encoding: str = 'utf-8', *, abs: bool = False)

Bases: PathBase

Initialize a File object.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike`	File path.	required
`encoding`	`str`	The file's encoding.	`'utf-8'`
`abs`	`bool`	Whether to use the absolute path.	`False`

Source code in stdl/fs.py

def __init__(
    self,
    path: str | PathLike,
    encoding: str = "utf-8",
    *,
    abs: bool = False,  # noqa: A002
) -> None:
    """
    Initialize a File object.

    Args:
        path (str | PathLike): File path.
        encoding (str, optional): The file's encoding.
        abs (bool): Whether to use the absolute path.
    """
    super().__init__(path, abs=abs)

    self.encoding = encoding

exists `property` ¶

exists: bool

Whether the file exists.

parent `property` ¶

parent: Directory

The parent directory.

dirname `property` ¶

dirname: str

The file's directory name.

ext `property` ¶

ext: str

The file's extension (without the dot). Returns empty string if the file has no extension.

suffix `property` ¶

suffix: str

The file extension WITH the dot (e.g., '.txt'). Empty string if no extension.

stem `property` ¶

stem: str

The file's stem (base name without extension).

size `property` ¶

size: int

The file's size in bytes.

size_readable `property` ¶

size_readable: str

The file's size in a human-readable format if readable is set to True.

create ¶

create() -> File

Create an empty file if it doesn't exist.

Source code in stdl/fs.py

def create(self) -> File:
    """Create an empty file if it doesn't exist."""
    if self.exists:
        return self

    open(self.path, "a", encoding=self.encoding).close()

    return self

create_async `async` ¶

create_async() -> File

Async version of create().

Source code in stdl/fs.py

async def create_async(self) -> File:
    """Async version of create()."""
    return await _run_fs_sync_shielded(self.create)

touch ¶

touch(*, mkdir: bool = False) -> File

Create the file if needed and update its timestamps.

Source code in stdl/fs.py

def touch(self, *, mkdir: bool = False) -> File:
    """Create the file if needed and update its timestamps."""
    parent = os.path.dirname(self.path)
    if parent and not os.path.isdir(parent):
        if mkdir:
            os.makedirs(parent, exist_ok=True)
        else:
            raise FileNotFoundError(f"No such directory: '{parent}'")

    with open(self.path, "a", encoding=self.encoding):
        pass

    os.utime(self.path, None)

    return self

touch_async `async` ¶

touch_async(*, mkdir: bool = False) -> File

Async version of touch().

Source code in stdl/fs.py

async def touch_async(self, *, mkdir: bool = False) -> File:
    """Async version of touch()."""
    return await _run_fs_sync_shielded(self.touch, mkdir=mkdir)

remove ¶

remove() -> File

Remove the file.

Source code in stdl/fs.py

def remove(self) -> File:
    """Remove the file."""
    if not self.exists:
        return self

    os.remove(self.path)

    return self

clear ¶

clear() -> File

Clear the contents of a file if it exists.

Source code in stdl/fs.py

def clear(self) -> File:
    """Clear the contents of a file if it exists."""
    if not self.exists:
        return self

    open(self.path, "w", encoding=self.encoding).close()

    return self

clear_async `async` ¶

clear_async() -> File

Async version of clear().

Source code in stdl/fs.py

async def clear_async(self) -> File:
    """Async version of clear()."""
    return await _run_fs_sync_shielded(self.clear)

read ¶

read(mode: Literal['r'] = 'r') -> str

read(mode: Literal['rb']) -> bytes

read(mode: Literal['r', 'rb'] = 'r') -> str | bytes

Read the contents of a file.

Parameters:

Name	Type	Description	Default
`mode`	`Literal['r', 'rb']`	The mode to open the file in. Use 'r' for text, 'rb' for binary.	`'r'`

Returns:

Type	Description
`str \| bytes`	str \| bytes: The file contents as string (text mode) or bytes (binary mode).

Source code in stdl/fs.py

def read(self, mode: Literal["r", "rb"] = "r") -> str | bytes:
    """
    Read the contents of a file.

    Args:
        mode: The mode to open the file in. Use 'r' for text, 'rb' for binary.

    Returns:
        str | bytes: The file contents as string (text mode) or bytes (binary mode).
    """
    if mode == "rb":
        with open(self.path, mode) as f:
            return f.read()

    with open(self.path, mode, encoding=self.encoding) as f:
        return f.read()

read_async `async` ¶

read_async(mode: Literal['r'] = 'r') -> str

read_async(mode: Literal['rb']) -> bytes

read_async(mode: Literal['r', 'rb'] = 'r') -> str | bytes

Async version of read().

Source code in stdl/fs.py

async def read_async(self, mode: Literal["r", "rb"] = "r") -> str | bytes:
    """Async version of read()."""
    return await run_fs_sync(self.read, mode)

write ¶

write(data: str, *, mode: Literal['w'] = 'w', newline: bool = True) -> File

write(data: bytes, *, mode: Literal['wb'], newline: bool = True) -> File

write(data: str | bytes, *, mode: Literal['w', 'wb'] = 'w', newline: bool = True) -> File

Write data to a file, overwriting any existing data.

Parameters:

Name	Type	Description	Default
`data`	`str \| bytes`	The data to write.	required
`mode`	`Literal['w', 'wb']`	The mode to open the file in. Use 'w' for text, 'wb' for binary.	`'w'`
`newline`	`bool`	Whether to add a newline at the end. Ignored in binary mode.	`True`

Returns:

Name	Type	Description
`File`	`File`	The File object for method chaining.

Source code in stdl/fs.py

def write(
    self,
    data: str | bytes,
    *,
    mode: Literal["w", "wb"] = "w",
    newline: bool = True,
) -> File:
    """
    Write data to a file, overwriting any existing data.

    Args:
        data: The data to write.
        mode: The mode to open the file in. Use 'w' for text, 'wb' for binary.
        newline: Whether to add a newline at the end. Ignored in binary mode.

    Returns:
        File: The File object for method chaining.
    """
    self._write(data, mode, newline=newline)

    return self

write_async `async` ¶

write_async(data: str, *, mode: Literal['w'] = 'w', newline: bool = True) -> File

write_async(data: bytes, *, mode: Literal['wb'], newline: bool = True) -> File

write_async(data: str | bytes, *, mode: Literal['w', 'wb'] = 'w', newline: bool = True) -> File

Async version of write().

Source code in stdl/fs.py

async def write_async(
    self,
    data: str | bytes,
    *,
    mode: Literal["w", "wb"] = "w",
    newline: bool = True,
) -> File:
    """Async version of write()."""
    return await _run_fs_sync_shielded(self.write, data, mode=mode, newline=newline)

append ¶

append(data: str, *, mode: Literal['a'] = 'a', newline: bool = True) -> File

append(data: bytes, *, mode: Literal['ab'], newline: bool = True) -> File

append(data: str | bytes, *, mode: Literal['a', 'ab'] = 'a', newline: bool = True) -> File

Append data to a file.

Parameters:

Name	Type	Description	Default
`data`	`str \| bytes`	The data to append.	required
`mode`	`Literal['a', 'ab']`	The mode to open the file in. Use 'a' for text, 'ab' for binary.	`'a'`
`newline`	`bool`	Whether to add a newline at the end. Ignored in binary mode.	`True`

Returns:

Name	Type	Description
`File`	`File`	The File object for method chaining.

Source code in stdl/fs.py

def append(
    self,
    data: str | bytes,
    *,
    mode: Literal["a", "ab"] = "a",
    newline: bool = True,
) -> File:
    """
    Append data to a file.

    Args:
        data: The data to append.
        mode: The mode to open the file in. Use 'a' for text, 'ab' for binary.
        newline: Whether to add a newline at the end. Ignored in binary mode.

    Returns:
        File: The File object for method chaining.
    """
    self._write(data, mode, newline=newline)

    return self

append_async `async` ¶

append_async(data: str, *, mode: Literal['a'] = 'a', newline: bool = True) -> File

append_async(data: bytes, *, mode: Literal['ab'], newline: bool = True) -> File

append_async(data: str | bytes, *, mode: Literal['a', 'ab'] = 'a', newline: bool = True) -> File

Async version of append().

Source code in stdl/fs.py

async def append_async(
    self,
    data: str | bytes,
    *,
    mode: Literal["a", "ab"] = "a",
    newline: bool = True,
) -> File:
    """Async version of append()."""
    return await _run_fs_sync_shielded(self.append, data, mode=mode, newline=newline)

open ¶

open(mode: Literal['r'] = 'r', encoding: str | None = None, **kwargs: OpenKwarg) -> IO[str]

open(mode: Literal['rb'], encoding: None = None, **kwargs: OpenKwarg) -> IO[bytes]

open(mode: str, encoding: str | None = None, **kwargs: OpenKwarg) -> IO[Any]

open(mode: str = 'r', encoding: str | None = None, **kwargs: OpenKwarg) -> IO[Any]

Open the file and return a file handle.

Parameters:

Name	Type	Description	Default
`mode`	`str`	The mode to open the file in.	`'r'`
`encoding`	`str \| None`	The encoding to use. Defaults to self.encoding for text modes.	`None`
`**kwargs`	`OpenKwarg`	Additional arguments passed to the built-in open().	`{}`

Returns:

Name	Type	Description
`IO`	`IO[Any]`	A file handle.

Source code in stdl/fs.py

def open(self, mode: str = "r", encoding: str | None = None, **kwargs: OpenKwarg) -> IO[Any]:
    """
    Open the file and return a file handle.

    Args:
        mode: The mode to open the file in.
        encoding: The encoding to use. Defaults to self.encoding for text modes.
        **kwargs: Additional arguments passed to the built-in open().

    Returns:
        IO: A file handle.
    """
    if "b" in mode:
        return open(self.path, mode, **kwargs)

    if encoding is None:
        encoding = self.encoding

    return open(self.path, mode, encoding=encoding, **kwargs)

open_async `async` ¶

open_async(mode: Literal['r'] = 'r', encoding: str | None = None, **kwargs: OpenKwarg) -> AsyncFileHandle[str]

open_async(mode: Literal['rb'], encoding: None = None, **kwargs: OpenKwarg) -> AsyncFileHandle[bytes]

open_async(mode: str, encoding: str | None = None, **kwargs: OpenKwarg) -> AsyncFileHandle[str] | AsyncFileHandle[bytes]

open_async(mode: str = 'r', encoding: str | None = None, **kwargs: OpenKwarg) -> AsyncFileHandle[str] | AsyncFileHandle[bytes]

Async version of open().

Source code in stdl/fs.py

async def open_async(
    self, mode: str = "r", encoding: str | None = None, **kwargs: OpenKwarg
) -> AsyncFileHandle[str] | AsyncFileHandle[bytes]:
    """Async version of open()."""
    if "b" not in mode and encoding is None:
        encoding = self.encoding
    anyio = _load_anyio()
    with anyio.CancelScope(shield=True):
        handle = await anyio.open_file(self.path, mode, encoding=encoding, **kwargs)

    return AsyncFileHandle(handle)

write_iter ¶

write_iter(data: Iterable[Any], sep: str = '\n') -> File

Write data from an iterable to a file, overwriting any existing data.

Parameters:

Name	Type	Description	Default
`data`	`Iterable`	The data to write.	required
`sep`	`str`	The separator to use between items.	`'\n'`

Returns:

Name	Type	Description
`File`	`File`	The File object for method chaining.

Source code in stdl/fs.py

def write_iter(self, data: Iterable[Any], sep: str = "\n") -> File:
    """
    Write data from an iterable to a file, overwriting any existing data.

    Args:
        data (Iterable): The data to write.
        sep (str, optional): The separator to use between items.

    Returns:
        File: The File object for method chaining.
    """
    self._write_iter(data, "w", sep=sep)

    return self

write_iter_async `async` ¶

write_iter_async(data: Iterable[Any], sep: str = '\n') -> File

Async version of write_iter().

Source code in stdl/fs.py

async def write_iter_async(self, data: Iterable[Any], sep: str = "\n") -> File:
    """Async version of write_iter()."""
    return await _run_fs_sync_shielded(self.write_iter, data, sep=sep)

append_iter ¶

append_iter(data: Iterable[Any], sep: str = '\n') -> File

Append data from an iterable to a file.

Parameters:

Name	Type	Description	Default
`data`	`Iterable`	The data to append.	required
`sep`	`str`	The separator to use between items.	`'\n'`

Returns:

Name	Type	Description
`File`	`File`	The File object for method chaining.

Source code in stdl/fs.py

def append_iter(self, data: Iterable[Any], sep: str = "\n") -> File:
    """
    Append data from an iterable to a file.

    Args:
        data (Iterable): The data to append.
        sep (str, optional): The separator to use between items.

    Returns:
        File: The File object for method chaining.
    """
    self._write_iter(data, "a", sep=sep)

    return self

append_iter_async `async` ¶

append_iter_async(data: Iterable[Any], sep: str = '\n') -> File

Async version of append_iter().

Source code in stdl/fs.py

async def append_iter_async(self, data: Iterable[Any], sep: str = "\n") -> File:
    """Async version of append_iter()."""
    return await _run_fs_sync_shielded(self.append_iter, data, sep=sep)

readlines ¶

readlines() -> list[str]

Equivalent to TextIOWrapper.readlines().

Source code in stdl/fs.py

def readlines(self) -> list[str]:
    """Equivalent to TextIOWrapper.readlines()."""
    with open(self.path, encoding=self.encoding) as f:
        return f.readlines()

readlines_async `async` ¶

readlines_async() -> list[str]

Async version of readlines().

Source code in stdl/fs.py

async def readlines_async(self) -> list[str]:
    """Async version of readlines()."""
    return await run_fs_sync(self.readlines)

splitlines ¶

splitlines() -> list[str]

Equivalent to File.read().splitlines().

Source code in stdl/fs.py

def splitlines(self) -> list[str]:
    """Equivalent to File.read().splitlines()."""
    data = self.read(mode="r")
    if isinstance(data, bytes):
        data = data.decode(self.encoding)

    return data.splitlines()

splitlines_async `async` ¶

splitlines_async() -> list[str]

Async version of splitlines().

Source code in stdl/fs.py

async def splitlines_async(self) -> list[str]:
    """Async version of splitlines()."""
    return await run_fs_sync(self.splitlines)

move_to ¶

move_to(target: File | Directory, *, mkdir: bool = False, overwrite: bool = True) -> File

Move the file to a new target.

Parameters:

Name	Type	Description	Default
`target`	`File \| Directory`	The destination `File` or `Directory`.	required
`mkdir`	`bool`	Whether to create missing parent directories. Defaults to False.	`False`
`overwrite`	`bool`	Whether to overwrite the destination if it already exists. Defaults to True.	`True`

Returns:

Name	Type	Description
`File`	`File`	A new File object pointing at the moved path.

Source code in stdl/fs.py

def move_to(
    self,
    target: File | Directory,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> File:
    """
    Move the file to a new target.

    Args:
        target: The destination `File` or `Directory`.
        mkdir: Whether to create missing parent directories. Defaults to False.
        overwrite: Whether to overwrite the destination if it already exists. Defaults to True.

    Returns:
        File: A new File object pointing at the moved path.
    """
    if isinstance(target, Directory):
        dest_path = os.path.join(target.path, self.basename)
    elif isinstance(target, File):
        dest_path = target.path
    else:
        raise TypeError("File.move_to() target must be a File or Directory")

    if _prepare_destination_path(
        dest_path,
        mkdir=mkdir,
        overwrite=overwrite,
        source_path=self.path,
        expected_type="file",
    ):
        return _clone_with_path(self, dest_path)

    shutil.move(self.path, dest_path)

    return _clone_with_path(self, dest_path)

move_to_async `async` ¶

move_to_async(target: File | Directory, *, mkdir: bool = False, overwrite: bool = True) -> File

Async version of move_to().

Source code in stdl/fs.py

async def move_to_async(
    self,
    target: File | Directory,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> File:
    """Async version of move_to()."""
    return await _run_fs_sync_shielded(self.move_to, target, mkdir=mkdir, overwrite=overwrite)

copy_to ¶

copy_to(target: File | Directory, *, mkdir: bool = False, overwrite: bool = True) -> File

Copy the file to a new target.

Parameters:

Name	Type	Description	Default
`target`	`File \| Directory`	The destination `File` or `Directory`.	required
`mkdir`	`bool`	Whether to create missing parent directories. Defaults to False.	`False`
`overwrite`	`bool`	Whether to overwrite the destination if it already exists. Defaults to True.	`True`

Returns:

Name	Type	Description
`File`	`File`	A new File object pointing at the copied path.

Source code in stdl/fs.py

def copy_to(
    self,
    target: File | Directory,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> File:
    """
    Copy the file to a new target.

    Args:
        target: The destination `File` or `Directory`.
        mkdir: Whether to create missing parent directories. Defaults to False.
        overwrite: Whether to overwrite the destination if it already exists. Defaults to True.

    Returns:
        File: A new File object pointing at the copied path.
    """
    if isinstance(target, Directory):
        dest_path = os.path.join(target.path, self.basename)
    elif isinstance(target, File):
        dest_path = target.path
    else:
        raise TypeError("File.copy_to() target must be a File or Directory")

    if _prepare_destination_path(
        dest_path,
        mkdir=mkdir,
        overwrite=overwrite,
        source_path=self.path,
        expected_type="file",
        remove_existing=False,
    ):
        return _clone_with_path(self, dest_path)

    _copy_file_to_destination(self.path, dest_path)

    return _clone_with_path(self, dest_path)

copy_to_async `async` ¶

copy_to_async(target: File | Directory, *, mkdir: bool = False, overwrite: bool = True) -> File

Async version of copy_to().

Source code in stdl/fs.py

async def copy_to_async(
    self,
    target: File | Directory,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> File:
    """Async version of copy_to()."""
    return await _run_fs_sync_shielded(self.copy_to, target, mkdir=mkdir, overwrite=overwrite)

with_dir ¶

with_dir(directory: str) -> File

Change the directory of the file object. This will not move the actual file to that directory. Use File.move_to for that.

Source code in stdl/fs.py

def with_dir(self, directory: str) -> File:
    """
    Change the directory of the file object. This will not move the actual file to that directory.
    Use File.move_to for that.
    """
    return _clone_with_path(self, os.path.join(directory, self.basename))

with_ext ¶

with_ext(ext: str) -> File

Change the extension of the file and return the new File object.

Parameters:

Name	Type	Description	Default
`ext`	`str`	The new extension of the file.	required

Returns:

Name	Type	Description
`File`	`File`	The File object with the new extension.

Source code in stdl/fs.py

def with_ext(self, ext: str) -> File:
    """
    Change the extension of the file and return the new File object.

    Args:
        ext (str): The new extension of the file.

    Returns:
        File: The File object with the new extension.
    """
    if not ext.startswith("."):
        ext = f".{ext}"

    return _clone_with_path(self, os.path.join(self.dirname, f"{self.stem}{ext}"))

with_suffix ¶

with_suffix(suffix: str) -> File

Add a suffix to the file's name and return the new File object.

Source code in stdl/fs.py

def with_suffix(self, suffix: str) -> File:
    """Add a suffix to the file's name and return the new File object."""
    filename = f"{self.stem}{suffix}{self.suffix}"
    return _clone_with_path(self, os.path.join(self.dirname, filename))

with_prefix ¶

with_prefix(prefix: str) -> File

Add a prefix to the file's name and return the new File object.

Source code in stdl/fs.py

def with_prefix(self, prefix: str) -> File:
    """Add a prefix to the file's name and return the new File object."""
    filename = f"{prefix}{self.stem}{self.suffix}"
    return _clone_with_path(self, os.path.join(self.dirname, filename))

with_stem ¶

with_stem(stem: str) -> File

Return a new File with the stem changed, keeping the suffix.

Source code in stdl/fs.py

def with_stem(self, stem: str) -> File:
    """Return a new File with the stem changed, keeping the suffix."""
    suffix = self.suffix
    return _clone_with_path(self, os.path.join(self.dirname, f"{stem}{suffix}"))

rename ¶

rename(name: str) -> File

Rename the file and return the new File object.

Source code in stdl/fs.py

def rename(self, name: str) -> File:
    """Rename the file and return the new File object."""
    new_path = os.path.join(self.dirname, name)
    os.rename(self.path, new_path)

    return _clone_with_path(self, new_path)

link ¶

link(target: str, follow_symlinks: bool = True) -> File

Create a hard link and return a File for the created link path.

Source code in stdl/fs.py

def link(self, target: str, follow_symlinks: bool = True) -> File:
    """Create a hard link and return a File for the created link path."""
    try:
        os.link(self.path, target, follow_symlinks=follow_symlinks)
    except NotImplementedError:
        if not follow_symlinks:
            raise

        os.link(self.path, target)

    return _clone_with_path(self, target)

link_async `async` ¶

link_async(target: str, follow_symlinks: bool = True) -> File

Async version of link().

Source code in stdl/fs.py

async def link_async(self, target: str, follow_symlinks: bool = True) -> File:
    """Async version of link()."""
    return await _run_fs_sync_shielded(self.link, target, follow_symlinks=follow_symlinks)

symlink ¶

symlink(target: str) -> File

Create a symbolic link and return a File for the created link path.

Source code in stdl/fs.py

def symlink(self, target: str) -> File:
    """Create a symbolic link and return a File for the created link path."""
    os.symlink(_symlink_source_path(self.path, target), target)

    return _clone_with_path(self, target)

symlink_async `async` ¶

symlink_async(target: str) -> File

Async version of symlink().

Source code in stdl/fs.py

async def symlink_async(self, target: str) -> File:
    """Async version of symlink()."""
    return await _run_fs_sync_shielded(self.symlink, target)

rand `classmethod` ¶

rand(prefix: str = 'file', ext: str = '') -> File

Create a new random file with a specified prefix and extension.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	The prefix for the file name.	`'file'`
`ext`	`str`	The extension for the file name.	`''`

Returns:

Name	Type	Description
`File`	`File`	A new File object with a random name.

Source code in stdl/fs.py

@classmethod
def rand(cls, prefix: str = "file", ext: str = "") -> File:
    """
    Create a new random file with a specified prefix and extension.

    Args:
        prefix (str, optional): The prefix for the file name.
        ext (str, optional): The extension for the file name.

    Returns:
        File: A new File object with a random name.
    """
    return File(rand_filename(prefix, ext))

Directory ¶

Directory(path: str | PathLike, *, abs: bool = False)

Bases: PathBase

Initialize a Directory object.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike`	Directory path.	required
`abs`	`bool`	Whether to use the absolute path.	`False`

Source code in stdl/fs.py

def __init__(
    self,
    path: str | PathLike,
    *,
    abs: bool = False,  # noqa: A002
) -> None:
    """
    Initialize a Directory object.

    Args:
        path (str | PathLike): Directory path.
        abs (bool): Whether to use the absolute path.
    """
    super().__init__(path, abs=abs)

size `property` ¶

size: int

Total size of the directory and all its contents in bytes.

size_readable `property` ¶

size_readable: str

Total size of the directory and all its contents in human-readable form.

exists `property` ¶

exists: bool

Check if the directory exists.

parent `property` ¶

parent: Directory

The parent directory.

create ¶

create(mode: int = 511, exist_ok: bool = True) -> Directory

Create directory (including parents).

Source code in stdl/fs.py

def create(self, mode: int = 0o777, exist_ok: bool = True) -> Directory:
    """Create directory (including parents)."""
    os.makedirs(self.path, mode=mode, exist_ok=exist_ok)

    return self

create_async `async` ¶

create_async(mode: int = 511, exist_ok: bool = True) -> Directory

Async version of create().

Source code in stdl/fs.py

async def create_async(self, mode: int = 0o777, exist_ok: bool = True) -> Directory:
    """Async version of create()."""
    return await _run_fs_sync_shielded(self.create, mode, exist_ok)

clear ¶

clear() -> Directory

Remove all children while keeping the directory itself.

Source code in stdl/fs.py

def clear(self) -> Directory:
    """Remove all children while keeping the directory itself."""
    if not self.exists:
        return self

    with os.scandir(self.path) as entries:
        for entry in entries:
            _remove_existing_path(entry.path)

    return self

clear_async `async` ¶

clear_async() -> Directory

Async version of clear().

Source code in stdl/fs.py

async def clear_async(self) -> Directory:
    """Async version of clear()."""
    return await _run_fs_sync_shielded(self.clear)

move_to ¶

move_to(target: Directory, *, mkdir: bool = False, overwrite: bool = True) -> Directory

Move this directory into another directory, preserving its basename.

Parameters:

Name	Type	Description	Default
`target`	`Directory`	The destination parent directory.	required
`mkdir`	`bool`	Whether to create missing parent directories. Defaults to False.	`False`
`overwrite`	`bool`	Whether to overwrite if target path already exists. Defaults to True.	`True`

Returns:

Name	Type	Description
`Directory`	`Directory`	A new Directory object pointing at the moved path.

Source code in stdl/fs.py

def move_to(
    self,
    target: Directory,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> Directory:
    """
    Move this directory into another directory, preserving its basename.

    Args:
        target: The destination parent directory.
        mkdir: Whether to create missing parent directories. Defaults to False.
        overwrite: Whether to overwrite if target path already exists. Defaults to True.

    Returns:
        Directory: A new Directory object pointing at the moved path.
    """
    if not isinstance(target, Directory):
        raise TypeError("Directory.move_to() target must be a Directory")

    dest = os.path.join(target.path, self.basename)
    _ensure_directory_target_is_not_inside_source(self.path, dest)
    if _prepare_destination_path(
        dest,
        mkdir=mkdir,
        overwrite=overwrite,
        source_path=self.path,
        expected_type="directory",
    ):
        return _clone_with_path(self, dest)

    shutil.move(self.path, dest)

    return _clone_with_path(self, dest)

move_to_async `async` ¶

move_to_async(target: Directory, *, mkdir: bool = False, overwrite: bool = True) -> Directory

Async version of move_to().

Source code in stdl/fs.py

async def move_to_async(
    self,
    target: Directory,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> Directory:
    """Async version of move_to()."""
    return await _run_fs_sync_shielded(self.move_to, target, mkdir=mkdir, overwrite=overwrite)

copy_to ¶

copy_to(target: Directory, *, mkdir: bool = False, overwrite: bool = True) -> Directory

Copy this directory into another directory, preserving its basename.

Parameters:

Name	Type	Description	Default
`target`	`Directory`	The destination parent directory.	required
`mkdir`	`bool`	Whether to create missing parent directories. Defaults to False.	`False`
`overwrite`	`bool`	Whether to overwrite if target path already exists. Defaults to True.	`True`

Returns:

Name	Type	Description
`Directory`	`Directory`	A new Directory object pointing at the copied path.

Source code in stdl/fs.py

def copy_to(
    self,
    target: Directory,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> Directory:
    """
    Copy this directory into another directory, preserving its basename.

    Args:
        target: The destination parent directory.
        mkdir: Whether to create missing parent directories. Defaults to False.
        overwrite: Whether to overwrite if target path already exists. Defaults to True.

    Returns:
        Directory: A new Directory object pointing at the copied path.
    """
    if not isinstance(target, Directory):
        raise TypeError("Directory.copy_to() target must be a Directory")

    dest = os.path.join(target.path, self.basename)
    _ensure_directory_target_is_not_inside_source(self.path, dest)
    if _prepare_destination_path(
        dest,
        mkdir=mkdir,
        overwrite=overwrite,
        source_path=self.path,
        expected_type="directory",
        remove_existing=False,
    ):
        return _clone_with_path(self, dest)

    _copy_directory_to_destination(self.path, dest, overwrite=overwrite)

    return _clone_with_path(self, dest)

copy_to_async `async` ¶

copy_to_async(target: Directory, *, mkdir: bool = False, overwrite: bool = True) -> Directory

Async version of copy_to().

Source code in stdl/fs.py

async def copy_to_async(
    self,
    target: Directory,
    *,
    mkdir: bool = False,
    overwrite: bool = True,
) -> Directory:
    """Async version of copy_to()."""
    return await _run_fs_sync_shielded(self.copy_to, target, mkdir=mkdir, overwrite=overwrite)

remove ¶

remove() -> Directory

Remove the directory and all its contents.

Source code in stdl/fs.py

def remove(self) -> Directory:
    """Remove the directory and all its contents."""
    if not self.exists:
        return self

    shutil.rmtree(self.path)

    return self

remove_async `async` ¶

remove_async() -> Directory

Async version of remove().

Source code in stdl/fs.py

async def remove_async(self) -> Directory:
    """Async version of remove()."""
    return await _run_fs_sync_shielded(self.remove)

yield_files ¶

yield_files(ext: str | tuple[str, ...] | None = None, glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True) -> Generator[File, None, None]

Yield files in the directory.

All filter parameters use AND logic - files must match all specified filters.

Parameters:

Name	Type	Description	Default
`ext`	`str \| tuple[str, ...] \| None`	If provided, only yield files with provided extensions.	`None`
`glob`	`str \| Pattern[str] \| None`	Glob pattern as string or compiled regex from fnmatch.translate().	`None`
`regex`	`str \| Pattern[str] \| None`	Regex pattern as string or compiled re.Pattern.	`None`
`recursive`	`bool`	Whether to search recursively.	`True`

Yields:

Type	Description
`File`	Generator[File, None, None]: The files in the directory.

Source code in stdl/fs.py

def yield_files(
    self,
    ext: str | tuple[str, ...] | None = None,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
) -> Generator[File, None, None]:
    """
    Yield files in the directory.

    All filter parameters use AND logic - files must match all specified filters.

    Args:
        ext: If provided, only yield files with provided extensions.
        glob: Glob pattern as string or compiled regex from fnmatch.translate().
        regex: Regex pattern as string or compiled re.Pattern.
        recursive: Whether to search recursively.

    Yields:
        Generator[File, None, None]: The files in the directory.
    """
    glob_pattern = re.compile(fnmatch.translate(glob)) if isinstance(glob, str) else glob
    regex_pattern = re.compile(regex) if isinstance(regex, str) else regex
    for file_path in yield_files_in(self.path, ext=ext, recursive=recursive):
        basename = os.path.basename(file_path)
        if glob_pattern and not glob_pattern.match(basename):
            continue

        if regex_pattern and not regex_pattern.search(basename):
            continue

        yield File(file_path)

yield_files_async `async` ¶

yield_files_async(ext: str | tuple[str, ...] | None = None, glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True) -> AsyncGenerator[File, None]

Async version of yield_files().

Source code in stdl/fs.py

async def yield_files_async(
    self,
    ext: str | tuple[str, ...] | None = None,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
) -> AsyncGenerator[File, None]:
    """Async version of yield_files()."""
    async for file in iterate_fs(
        lambda: self.yield_files(ext=ext, glob=glob, regex=regex, recursive=recursive)
    ):
        yield file

get_files ¶

get_files(ext: str | tuple[str, ...] | None = None, glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True) -> list[File]

Get files in the directory.

All filter parameters use AND logic - files must match all specified filters.

Parameters:

Name	Type	Description	Default
`ext`	`str \| tuple[str, ...] \| None`	If provided, only yield files with provided extensions.	`None`
`glob`	`str \| Pattern[str] \| None`	Glob pattern as string or compiled regex from fnmatch.translate().	`None`
`regex`	`str \| Pattern[str] \| None`	Regex pattern as string or compiled re.Pattern.	`None`
`recursive`	`bool`	Whether to search recursively.	`True`

Returns:

Type	Description
`list[File]`	list[File]: The files in the directory.

Source code in stdl/fs.py

def get_files(
    self,
    ext: str | tuple[str, ...] | None = None,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
) -> list[File]:
    """
    Get files in the directory.

    All filter parameters use AND logic - files must match all specified filters.

    Args:
        ext: If provided, only yield files with provided extensions.
        glob: Glob pattern as string or compiled regex from fnmatch.translate().
        regex: Regex pattern as string or compiled re.Pattern.
        recursive: Whether to search recursively.

    Returns:
        list[File]: The files in the directory.
    """
    return list(self.yield_files(ext=ext, glob=glob, regex=regex, recursive=recursive))

get_files_async `async` ¶

get_files_async(ext: str | tuple[str, ...] | None = None, glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True) -> list[File]

Async version of get_files().

Source code in stdl/fs.py

async def get_files_async(
    self,
    ext: str | tuple[str, ...] | None = None,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
) -> list[File]:
    """Async version of get_files()."""
    return await run_fs_sync(
        self.get_files, ext=ext, glob=glob, regex=regex, recursive=recursive
    )

read_files_async `async` ¶

read_files_async(*, mode: Literal['r'] = 'r', ext: str | tuple[str, ...] | None = None, glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True, concurrency: int = 8) -> list[tuple[File, str]]

read_files_async(*, mode: Literal['rb'], ext: str | tuple[str, ...] | None = None, glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True, concurrency: int = 8) -> list[tuple[File, bytes]]

read_files_async(*, mode: Literal['r', 'rb'] = 'r', ext: str | tuple[str, ...] | None = None, glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True, concurrency: int = 8) -> list[tuple[File, str]] | list[tuple[File, bytes]]

Read matching files concurrently.

Parameters:

Name	Type	Description	Default
`mode`	`Literal['r', 'rb']`	The mode to read files in. Use "r" for text and "rb" for binary.	`'r'`
`ext`	`str \| tuple[str, ...] \| None`	If provided, only read files with provided extensions.	`None`
`glob`	`str \| Pattern[str] \| None`	Glob pattern as string or compiled regex from fnmatch.translate().	`None`
`regex`	`str \| Pattern[str] \| None`	Regex pattern as string or compiled re.Pattern.	`None`
`recursive`	`bool`	Whether to search recursively.	`True`
`concurrency`	`int`	Maximum number of concurrent read operations.	`8`

Returns:

Type	Description
`list[tuple[File, str]] \| list[tuple[File, bytes]]`	list[tuple[File, str]] \| list[tuple[File, bytes]]: Matching files and their contents.

Source code in stdl/fs.py

async def read_files_async(
    self,
    *,
    mode: Literal["r", "rb"] = "r",
    ext: str | tuple[str, ...] | None = None,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
    concurrency: int = 8,
) -> list[tuple[File, str]] | list[tuple[File, bytes]]:
    """
    Read matching files concurrently.

    Args:
        mode: The mode to read files in. Use "r" for text and "rb" for binary.
        ext: If provided, only read files with provided extensions.
        glob: Glob pattern as string or compiled regex from fnmatch.translate().
        regex: Regex pattern as string or compiled re.Pattern.
        recursive: Whether to search recursively.
        concurrency: Maximum number of concurrent read operations.

    Returns:
        list[tuple[File, str]] | list[tuple[File, bytes]]: Matching files and their contents.
    """
    files = await self.get_files_async(ext=ext, glob=glob, regex=regex, recursive=recursive)

    if mode == "rb":

        async def read_file_binary(file: File) -> tuple[File, bytes]:
            return file, await file.read_async(mode="rb")

        return await _map_limited_async(files, read_file_binary, concurrency=concurrency)

    async def read_file_text(file: File) -> tuple[File, str]:
        return file, await file.read_async(mode="r")

    return await _map_limited_async(files, read_file_text, concurrency=concurrency)

remove_files_async `async` ¶

remove_files_async(*, ext: str | tuple[str, ...] | None = None, glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True, concurrency: int = 8) -> list[File]

Remove matching files concurrently.

Parameters:

Name	Type	Description	Default
`ext`	`str \| tuple[str, ...] \| None`	If provided, only remove files with provided extensions.	`None`
`glob`	`str \| Pattern[str] \| None`	Glob pattern as string or compiled regex from fnmatch.translate().	`None`
`regex`	`str \| Pattern[str] \| None`	Regex pattern as string or compiled re.Pattern.	`None`
`recursive`	`bool`	Whether to search recursively.	`True`
`concurrency`	`int`	Maximum number of concurrent remove operations.	`8`

Returns:

Type	Description
`list[File]`	list[File]: The removed files in snapshot order.

Source code in stdl/fs.py

async def remove_files_async(
    self,
    *,
    ext: str | tuple[str, ...] | None = None,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
    concurrency: int = 8,
) -> list[File]:
    """
    Remove matching files concurrently.

    Args:
        ext: If provided, only remove files with provided extensions.
        glob: Glob pattern as string or compiled regex from fnmatch.translate().
        regex: Regex pattern as string or compiled re.Pattern.
        recursive: Whether to search recursively.
        concurrency: Maximum number of concurrent remove operations.

    Returns:
        list[File]: The removed files in snapshot order.
    """
    files = await self.get_files_async(ext=ext, glob=glob, regex=regex, recursive=recursive)

    async def remove_file(file: File) -> File:
        await file.remove_async()
        return file

    return await _map_limited_async(files, remove_file, concurrency=concurrency)

yield_subdirs ¶

yield_subdirs(glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True) -> Generator[Directory, None, None]

Yield subdirectories in the directory.

All filter parameters use AND logic - directories must match all specified filters.

Parameters:

Name	Type	Description	Default
`glob`	`str \| Pattern[str] \| None`	Glob pattern as string or compiled regex from fnmatch.translate().	`None`
`regex`	`str \| Pattern[str] \| None`	Regex pattern as string or compiled re.Pattern.	`None`
`recursive`	`bool`	Whether to search recursively.	`True`

Yields:

Type	Description
`Directory`	Generator[Directory, None, None]: The subdirectories in the directory.

Source code in stdl/fs.py

def yield_subdirs(
    self,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
) -> Generator[Directory, None, None]:
    """
    Yield subdirectories in the directory.

    All filter parameters use AND logic - directories must match all specified filters.

    Args:
        glob: Glob pattern as string or compiled regex from fnmatch.translate().
        regex: Regex pattern as string or compiled re.Pattern.
        recursive: Whether to search recursively.

    Yields:
        Generator[Directory, None, None]: The subdirectories in the directory.
    """
    glob_pattern = re.compile(fnmatch.translate(glob)) if isinstance(glob, str) else glob
    regex_pattern = re.compile(regex) if isinstance(regex, str) else regex
    for dir_path in yield_dirs_in(self.path, recursive=recursive):
        basename = os.path.basename(dir_path)
        if glob_pattern and not glob_pattern.match(basename):
            continue

        if regex_pattern and not regex_pattern.search(basename):
            continue

        yield Directory(dir_path)

yield_subdirs_async `async` ¶

yield_subdirs_async(glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True) -> AsyncGenerator[Directory, None]

Async version of yield_subdirs().

Source code in stdl/fs.py

async def yield_subdirs_async(
    self,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
) -> AsyncGenerator[Directory, None]:
    """Async version of yield_subdirs()."""
    async for directory in iterate_fs(
        lambda: self.yield_subdirs(glob=glob, regex=regex, recursive=recursive)
    ):
        yield directory

get_subdirs ¶

get_subdirs(glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True) -> list[Directory]

Get subdirectories in the directory.

All filter parameters use AND logic - directories must match all specified filters.

Parameters:

Name	Type	Description	Default
`glob`	`str \| Pattern[str] \| None`	Glob pattern as string or compiled regex from fnmatch.translate().	`None`
`regex`	`str \| Pattern[str] \| None`	Regex pattern as string or compiled re.Pattern.	`None`
`recursive`	`bool`	Whether to search recursively.	`True`

Returns:

Type	Description
`list[Directory]`	list[Directory]: The subdirectories in the directory.

Source code in stdl/fs.py

def get_subdirs(
    self,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
) -> list[Directory]:
    """
    Get subdirectories in the directory.

    All filter parameters use AND logic - directories must match all specified filters.

    Args:
        glob: Glob pattern as string or compiled regex from fnmatch.translate().
        regex: Regex pattern as string or compiled re.Pattern.
        recursive: Whether to search recursively.

    Returns:
        list[Directory]: The subdirectories in the directory.
    """
    return list(self.yield_subdirs(glob=glob, regex=regex, recursive=recursive))

get_subdirs_async `async` ¶

get_subdirs_async(glob: str | Pattern[str] | None = None, regex: str | Pattern[str] | None = None, recursive: bool = True) -> list[Directory]

Async version of get_subdirs().

Source code in stdl/fs.py

async def get_subdirs_async(
    self,
    glob: str | re.Pattern[str] | None = None,
    regex: str | re.Pattern[str] | None = None,
    recursive: bool = True,
) -> list[Directory]:
    """Async version of get_subdirs()."""
    return await run_fs_sync(self.get_subdirs, glob=glob, regex=regex, recursive=recursive)

yield_entries ¶

yield_entries(*, abs: bool = False) -> Generator[File | Directory, None, None]

Yield immediate child files and directories in a single pass.

Source code in stdl/fs.py

def yield_entries(self, *, abs: bool = False) -> Generator[File | Directory, None, None]:  # noqa: A002
    """Yield immediate child files and directories in a single pass."""
    with os.scandir(self.path) as entries:
        for entry in entries:
            entry_path = os.path.abspath(entry.path) if abs else entry.path
            if entry.is_dir():
                yield Directory(entry_path)
            elif entry.is_file():
                yield File(entry_path)

yield_entries_async `async` ¶

yield_entries_async(*, abs: bool = False) -> AsyncGenerator[File | Directory, None]

Async version of yield_entries().

Source code in stdl/fs.py

async def yield_entries_async(
    self,
    *,
    abs: bool = False,  # noqa: A002
) -> AsyncGenerator[File | Directory, None]:
    """Async version of yield_entries()."""
    async for entry in iterate_fs(lambda: self.yield_entries(abs=abs)):
        yield entry

get_entries ¶

get_entries(*, abs: bool = False) -> list[File | Directory]

Return immediate child files and directories.

Source code in stdl/fs.py

def get_entries(self, *, abs: bool = False) -> list[File | Directory]:  # noqa: A002
    """Return immediate child files and directories."""
    return list(self.yield_entries(abs=abs))

get_entries_async `async` ¶

get_entries_async(*, abs: bool = False) -> list[File | Directory]

Async version of get_entries().

Source code in stdl/fs.py

async def get_entries_async(
    self,
    *,
    abs: bool = False,  # noqa: A002
) -> list[File | Directory]:
    """Async version of get_entries()."""
    return await run_fs_sync(self.get_entries, abs=abs)

walk ¶

walk(*, follow_symlinks: bool = False, abs: bool = False) -> Generator[tuple[Directory, list[Directory], list[File]], None, None]

Recursively walk the directory tree in top-down order.

Yields:

Type	Description
`Directory`	tuple[Directory, list[Directory], list[File]]: The current root, its child directories,
`list[Directory]`	and its child files.

Source code in stdl/fs.py

def walk(
    self,
    *,
    follow_symlinks: bool = False,
    abs: bool = False,  # noqa: A002
) -> Generator[tuple[Directory, list[Directory], list[File]], None, None]:
    """
    Recursively walk the directory tree in top-down order.

    Yields:
        tuple[Directory, list[Directory], list[File]]: The current root, its child directories,
        and its child files.
    """
    yield from _walk_directory(self, follow_symlinks=follow_symlinks, abs=abs, visited=None)

walk_async `async` ¶

walk_async(*, follow_symlinks: bool = False, abs: bool = False) -> AsyncGenerator[tuple[Directory, list[Directory], list[File]], None]

Async version of walk().

Source code in stdl/fs.py

async def walk_async(
    self,
    *,
    follow_symlinks: bool = False,
    abs: bool = False,  # noqa: A002
) -> AsyncGenerator[tuple[Directory, list[Directory], list[File]], None]:
    """Async version of walk()."""
    async for entry in iterate_fs(lambda: self.walk(follow_symlinks=follow_symlinks, abs=abs)):
        yield entry

pickle_load ¶

pickle_load(filepath: str | PathLike) -> PickleDataT

Loads a pickled file.

Source code in stdl/fs.py

def pickle_load(filepath: str | PathLike) -> PickleDataT:
    """Loads a pickled file."""
    with open(filepath, "rb") as f:
        return pickle.load(f)  # noqa: S301

pickle_load_async `async` ¶

pickle_load_async(filepath: str | PathLike) -> PickleDataT

Async version of pickle_load().

Source code in stdl/fs.py

async def pickle_load_async(filepath: str | PathLike) -> PickleDataT:
    """Async version of pickle_load()."""
    return await run_fs_sync(pickle_load, filepath)

pickle_dump ¶

pickle_dump(data: PickleDataT, filepath: str | PathLike) -> None

Dumps an object to the specified filepath.

Source code in stdl/fs.py

def pickle_dump(data: PickleDataT, filepath: str | PathLike) -> None:
    """Dumps an object to the specified filepath."""
    with open(filepath, "wb") as f:
        pickle.dump(data, f)

pickle_dump_async `async` ¶

pickle_dump_async(data: PickleDataT, filepath: str | PathLike) -> None

Async version of pickle_dump().

Source code in stdl/fs.py

async def pickle_dump_async(data: PickleDataT, filepath: str | PathLike) -> None:
    """Async version of pickle_dump()."""
    await _run_fs_sync_shielded(pickle_dump, data, filepath)

json_load ¶

json_load(path: str | PathLike, encoding: str = 'utf-8') -> dict[Any, Any] | list[dict[Any, Any]]

Load a JSON file from the given path.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike`	The path of the JSON file to load.	required
`encoding`	`str`	The encoding of the file. Defaults to "utf-8".	`'utf-8'`

Returns:

Type	Description
`dict[Any, Any] \| list[dict[Any, Any]]`	dict \| list[dict]: The JSON data loaded from the file.

Source code in stdl/fs.py

def json_load(
    path: str | PathLike, encoding: str = "utf-8"
) -> dict[Any, Any] | list[dict[Any, Any]]:
    """
    Load a JSON file from the given path.

    Args:
        path (str | PathLike): The path of the JSON file to load.
        encoding (str, optional): The encoding of the file. Defaults to "utf-8".

    Returns:
        dict | list[dict]: The JSON data loaded from the file.
    """
    with open(os.fspath(path), encoding=encoding) as f:
        return json.load(f)

json_load_async `async` ¶

json_load_async(path: str | PathLike, encoding: str = 'utf-8') -> dict[Any, Any] | list[dict[Any, Any]]

Async version of json_load().

Source code in stdl/fs.py

async def json_load_async(
    path: str | PathLike, encoding: str = "utf-8"
) -> dict[Any, Any] | list[dict[Any, Any]]:
    """Async version of json_load()."""
    return await run_fs_sync(json_load, path, encoding)

json_append ¶

json_append(data: JsonValue, filepath: str | PathLike, encoding: str = 'utf-8', default: Callable[[Any], str] = str, indent: int = 4) -> None

Append data to a JSON file.

Parameters:

Name	Type	Description	Default
`data`	`dict \| list[dict]`	The data to be appended	required
`filepath`	`str \| PathLike`	The path of the JSON file	required
`encoding`	`str`	The encoding of the file. Defaults to "utf-8".	`'utf-8'`
`default`		A function that gets called for objects that can't otherwise be serialized. See json.dump() documentation for more information.	required
`indent`	`int`	The number of spaces to use for indentation. Defaults to 4.	`4`

Source code in stdl/fs.py

def json_append(
    data: JsonValue,
    filepath: str | PathLike,
    encoding: str = "utf-8",
    default: Callable[[Any], str] = str,
    indent: int = 4,
) -> None:
    """
    Append data to a JSON file.

    Args:
        data (dict | list[dict]): The data to be appended
        filepath (str | PathLike): The path of the JSON file
        encoding (str, optional): The encoding of the file. Defaults to "utf-8".
        default : A function that gets called for objects that can't otherwise be serialized.
                  See json.dump() documentation for more information.
        indent (int, optional): The number of spaces to use for indentation. Defaults to 4.
    """
    file = File(filepath)
    path = os.fspath(filepath)

    if not file.exists or file.size == 0:
        json_dump([data], filepath, encoding=encoding, indent=indent, default=default)
        return

    with open(path, "a+", encoding=encoding) as f:
        f.seek(0)
        first_char = f.read(1)
        if first_char == "[":
            f.seek(0, os.SEEK_END)
            f.seek(f.tell() - 2, os.SEEK_SET)
            f.truncate()
            f.write(",\n")
            json.dump(data, f, indent=indent, default=default)
            f.write("]\n")
        elif first_char == "{":
            file_data = first_char + f.read()
            f.seek(0)
            f.truncate()
            f.seek(0)
            f.write("[\n")
            f.write(file_data)
            f.write(",\n")
            json.dump(data, f, indent=indent, default=default)
            f.write("]\n")
        else:
            raise ValueError(f"Cannot parse '{path}' as JSON.")

json_append_async `async` ¶

json_append_async(data: JsonValue, filepath: str | PathLike, encoding: str = 'utf-8', default: Callable[[Any], str] = str, indent: int = 4) -> None

Async version of json_append().

Source code in stdl/fs.py

async def json_append_async(
    data: JsonValue,
    filepath: str | PathLike,
    encoding: str = "utf-8",
    default: Callable[[Any], str] = str,
    indent: int = 4,
) -> None:
    """Async version of json_append()."""
    await _run_fs_sync_shielded(
        json_append, data, filepath, encoding=encoding, default=default, indent=indent
    )

json_dump ¶

json_dump(data: JsonValue, path: str | PathLike, encoding: str = 'utf-8', default: Callable[[Any], str] = str, indent: int = 4) -> None

Dumps data to a JSON file.

Parameters:

Name	Type	Description	Default
`data`	`JsonValue`	data to be dumped	required
`path`	`str \| PathLike`	path to the output file	required
`encoding`	`str`	encoding of the output file. Default: 'utf-8'	`'utf-8'`
`default`	`Callable[[Any], str]`	A function that gets called on objects that cannot be serialized. Default: str	`str`
`indent`	`int`	number of spaces to use when indenting the output json. Default: 4	`4`

Source code in stdl/fs.py

def json_dump(
    data: JsonValue,
    path: str | PathLike,
    encoding: str = "utf-8",
    default: Callable[[Any], str] = str,
    indent: int = 4,
) -> None:
    """
    Dumps data to a JSON file.

    Args:
        data: data to be dumped
        path (str | PathLike): path to the output file
        encoding (str): encoding of the output file. Default: 'utf-8'
        default: A function that gets called on objects that cannot be serialized. Default: str
        indent (int): number of spaces to use when indenting the output json. Default: 4
    """
    with open(os.fspath(path), "w", encoding=encoding) as f:
        json.dump(data, f, indent=indent, default=default)

json_dump_async `async` ¶

json_dump_async(data: JsonValue, path: str | PathLike, encoding: str = 'utf-8', default: Callable[[Any], str] = str, indent: int = 4) -> None

Async version of json_dump().

Source code in stdl/fs.py

async def json_dump_async(
    data: JsonValue,
    path: str | PathLike,
    encoding: str = "utf-8",
    default: Callable[[Any], str] = str,
    indent: int = 4,
) -> None:
    """Async version of json_dump()."""
    await _run_fs_sync_shielded(
        json_dump, data, path, encoding=encoding, default=default, indent=indent
    )

yaml_load ¶

yaml_load(path: str | PathLike, encoding: str = 'utf-8') -> dict[Any, Any] | list[dict[Any, Any]]

Load a YAML file from the given path.

Parameters:

Name	Type	Description	Default
`path`	`str \| PathLike`	The path of the YAML file to load.	required
`encoding`	`str`	The encoding of the file. Defaults to "utf-8".	`'utf-8'`

Returns:

Type	Description
`dict[Any, Any] \| list[dict[Any, Any]]`	dict \| list[dict]: The YAML data loaded from the file.

Source code in stdl/fs.py

def yaml_load(
    path: str | PathLike, encoding: str = "utf-8"
) -> dict[Any, Any] | list[dict[Any, Any]]:
    """
    Load a YAML file from the given path.

    Args:
        path (str | PathLike): The path of the YAML file to load.
        encoding (str, optional): The encoding of the file. Defaults to "utf-8".

    Returns:
        dict | list[dict]: The YAML data loaded from the file.
    """
    with open(path, encoding=encoding) as f:
        return yaml.safe_load(f)

yaml_load_async `async` ¶

yaml_load_async(path: str | PathLike, encoding: str = 'utf-8') -> dict[Any, Any] | list[dict[Any, Any]]

Async version of yaml_load().

Source code in stdl/fs.py

async def yaml_load_async(
    path: str | PathLike, encoding: str = "utf-8"
) -> dict[Any, Any] | list[dict[Any, Any]]:
    """Async version of yaml_load()."""
    return await run_fs_sync(yaml_load, path, encoding)

yaml_dump ¶

yaml_dump(data: YamlValue, path: str | PathLike, encoding: str = 'utf-8') -> None

Dumps data to a YAML file.

Parameters:

Name	Type	Description	Default
`data`	`YamlValue`	data to be dumped	required
`path`	`Pathlike`	path to the output file	required
`encoding`	`str`	encoding of the output file. Default: 'utf-8'	`'utf-8'`

Source code in stdl/fs.py

def yaml_dump(data: YamlValue, path: str | PathLike, encoding: str = "utf-8") -> None:
    """
    Dumps data to a YAML file.

    Args:
        data: data to be dumped
        path (Pathlike): path to the output file
        encoding (str): encoding of the output file. Default: 'utf-8'
    """
    with open(path, "w", encoding=encoding) as f:
        yaml.safe_dump(data, f)

yaml_dump_async `async` ¶

yaml_dump_async(data: YamlValue, path: str | PathLike, encoding: str = 'utf-8') -> None

Async version of yaml_dump().

Source code in stdl/fs.py

async def yaml_dump_async(data: YamlValue, path: str | PathLike, encoding: str = "utf-8") -> None:
    """Async version of yaml_dump()."""
    await _run_fs_sync_shielded(yaml_dump, data, path, encoding)

toml_load_async `async` ¶

toml_load_async(path: str | PathLike, encoding: str = 'utf-8') -> dict[str, Any]

Async version of toml_load().

Source code in stdl/fs.py

async def toml_load_async(path: str | PathLike, encoding: str = "utf-8") -> dict[str, Any]:
    """Async version of toml_load()."""
    return await run_fs_sync(toml_load, path, encoding)

toml_dump_async `async` ¶

toml_dump_async(data: dict[str, Any], path: str | PathLike, encoding: str = 'utf-8') -> None

Async version of toml_dump().

Source code in stdl/fs.py

async def toml_dump_async(
    data: dict[str, Any], path: str | PathLike, encoding: str = "utf-8"
) -> None:
    """Async version of toml_dump()."""
    await _run_fs_sync_shielded(toml_dump, data, path, encoding)

get_dir_size ¶

get_dir_size(directory: str | PathLike, *, readable: Literal[False] = False) -> int

get_dir_size(directory: str | PathLike, *, readable: Literal[True]) -> str

get_dir_size(directory: str | PathLike, *, readable: bool = False) -> str | int

Calculates total size of a directory.

Parameters:

Name	Type	Description	Default
`directory`	`(str, Path)`	target directory	required
`readable`	`bool`	Return the size in human-readable format	`False`

Source code in stdl/fs.py

def get_dir_size(directory: str | PathLike, *, readable: bool = False) -> str | int:
    """
    Calculates total size of a directory.

    Args:
        directory (str, Path): target directory
        readable (bool, optional): Return the size in human-readable format
    """
    total_size = 0
    for dirpath, _, filenames in os.walk(directory):
        for f in filenames:
            fp = os.path.join(dirpath, f)
            # Skip if it's symbolic link.
            if not os.path.islink(fp):
                total_size += os.path.getsize(fp)

    if readable:
        return bytes_readable(total_size)

    return total_size

move_files ¶

move_files(files: list[str | PathLike], directory: str | PathLike, *, mkdir: bool = False) -> None

Moves files to a specified directory.

Parameters:

Name	Type	Description	Default
`files`	`list[str \| PathLike]`	List of files to be moved	required
`directory`	`str \| PathLike`	target directory	required
`mkdir`	`bool`	whether to create the directory if it doesn't exist. Default: False	`False`

Raises:

Type	Description
`FileNotFoundError`	if the target directory does not exist and mkdir is False.

Source code in stdl/fs.py

def move_files(
    files: list[str | PathLike], directory: str | PathLike, *, mkdir: bool = False
) -> None:
    """
    Moves files to a specified directory.

    Args:
        files (list[str | PathLike]): List of files to be moved
        directory (str | PathLike): target directory
        mkdir (bool): whether to create the directory if it doesn't exist. Default: False

    Raises:
        FileNotFoundError : if the target directory does not exist and mkdir is False.
    """
    directory = os.fspath(directory)
    if exists(directory):
        if mkdir:
            os.makedirs(directory, exist_ok=True)
        else:
            raise FileNotFoundError(f"{directory} is not a directory")

    for file in files:
        os.rename(os.fspath(file), f"{directory}{SEP}{os.path.basename(file)}")

rand_filename ¶

rand_filename(prefix: str = 'file', ext: str = '', include_datetime: bool = False) -> str

Generates a random filename with the given prefix and extension. Optionally includes current date and time in the filename.

Parameters:

Name	Type	Description	Default
`prefix`	`str`	Filename prefix.	`'file'`
`ext`	`str`	Filename extension.	`''`
`include_datetime`	`bool`	Whether to include date and time.	`False`

Returns:

Name	Type	Description
`str`	`str`	The generated random filename.

Source code in stdl/fs.py

def rand_filename(prefix: str = "file", ext: str = "", include_datetime: bool = False) -> str:
    """
    Generates a random filename with the given prefix and extension.
    Optionally includes current date and time in the filename.

    Args:
        prefix (str, optional): Filename prefix.
        ext (str, optional): Filename extension.
        include_datetime (bool, optional): Whether to include date and time.

    Returns:
        str: The generated random filename.
    """
    if ext and not ext.startswith("."):
        ext = f".{ext}"
    num = str(secrets.randbelow(9_000_000_000) + 1_000_000_000).zfill(10)
    if include_datetime:
        creation_time = (
            datetime.now(tz=timezone.utc).astimezone().strftime("%Y-%m-%d.%H-%M-%S-%f")[:-3]
        )
        filename = f"{prefix}.{num}.{creation_time}{ext}"
    else:
        filename = f"{prefix}.{num}{ext}"

    return safe_filename(filename)

bytes_readable ¶

bytes_readable(size_bytes: int) -> str

Convert bytes to a human-readable string.

Parameters:

Name	Type	Description	Default
`size_bytes`	`int`	The number of bytes	required

Returns: str : The number of bytes in a human-readable format

Source code in stdl/fs.py

def bytes_readable(size_bytes: int) -> str:
    """
    Convert bytes to a human-readable string.

    Args:
        size_bytes (int): The number of bytes
    Returns:
        str : The number of bytes in a human-readable format
    """
    if size_bytes < 0:
        raise ValueError(size_bytes)

    if size_bytes == 0:
        return "0B"

    i = math.floor(math.log(size_bytes, 1024))
    p = math.pow(1024, i)
    s = round(size_bytes / p, 2)
    size_name = ("B", "KB", "MB", "GB", "TB", "PB", "EB", "ZB", "YB")

    return f"{s} {size_name[i]}"

readable_size_to_bytes ¶

readable_size_to_bytes(size: str, kb_size: Literal[1000, 1024] = 1024) -> int

Convert human-readable string to bytes.

Parameters:

Name	Type	Description	Default
`size`	`str`	The number of bytes in human-readable format	required
`kb_size`	`int`	The byte size of a kilobyte (1000 or 1024). Defaults to 1024.	`1024`

Returns:

Name	Type	Description
`int`	`int`	The number of bytes

Source code in stdl/fs.py

def readable_size_to_bytes(size: str, kb_size: Literal[1000, 1024] = 1024) -> int:
    """
    Convert human-readable string to bytes.

    Args:
        size (str): The number of bytes in human-readable format
        kb_size (int, optional): The byte size of a kilobyte (1000 or 1024). Defaults to 1024.

    Returns:
        int: The number of bytes
    """
    if kb_size not in (1000, 1024):
        raise ValueError(f"Invalid kb_size: {kb_size}. Must be 1000 or 1024.")

    size = size.upper().replace(" ", "")
    if size.isdigit():
        return int(size)

    match = re.match(r"^(\d+(\.\d+)?)(B|KB|MB|GB|TB)$", size)
    if not match:
        raise ValueError(f"Invalid size format: {size}")

    number, unit = float(match.group(1)), match.group(3)

    units = {"B": 1, "KB": kb_size, "MB": kb_size**2, "GB": kb_size**3, "TB": kb_size**4}

    if unit not in units:
        raise ValueError(f"Invalid unit: {unit}")

    return int(number * units[unit])

windows_has_drive ¶

windows_has_drive(letter: str) -> bool

Check if a drive letter exists on Windows. Will always return False if the platform is not Windows.

Parameters:

Name	Type	Description	Default
`letter`	`str`	The letter of the drive.	required

Returns:

Name	Type	Description
`bool`	`bool`	Whether the drive exists.

Source code in stdl/fs.py

def windows_has_drive(letter: str) -> bool:
    """
    Check if a drive letter exists on Windows.
    Will always return False if the platform is not Windows.

    Args:
        letter (str): The letter of the drive.

    Returns:
        bool: Whether the drive exists.
    """
    if sys.platform != "win32":
        return False

    return os.path.exists(f"{letter}:{SEP}")

is_wsl ¶

is_wsl() -> bool

Check if the current platform is Windows Subsystem for Linux (WSL).

Source code in stdl/fs.py

def is_wsl() -> bool:
    """Check if the current platform is Windows Subsystem for Linux (WSL)."""
    return sys.platform == "linux" and "microsoft" in platform.platform()

mkdir ¶

mkdir(path: str | PathLike, mode: int = 511, exist_ok: bool = True) -> None

Creates a directory.

Parameters:

Name	Type	Description	Default
`path`	`str \| Path`	The path of the directory to create.	required
`mode`	`int`	The mode to set for the directory. Defaults to 511 (octal 0777).	`511`
`exist_ok`	`bool`	Whether to raise an exception if the directory already exists. Defaults to True.	`True`

Source code in stdl/fs.py

def mkdir(path: str | PathLike, mode: int = 511, exist_ok: bool = True) -> None:
    """
    Creates a directory.

    Args:
        path (str | Path): The path of the directory to create.
        mode (int, optional): The mode to set for the directory. Defaults to 511 (octal 0777).
        exist_ok (bool, optional): Whether to raise an exception if the directory already exists. Defaults to True.
    """
    os.makedirs(os.fspath(path), exist_ok=exist_ok, mode=mode)

mkdirs ¶

mkdirs(dest: str | PathLike, names: list[str]) -> None

Creates directories inside a destination directory.

Parameters:

Name	Type	Description	Default
`dest`	`str \| Path`	The destination directory.	required
`names`	`list[str]`	A list of directory names to be created in the destination directory.	required

Source code in stdl/fs.py

def mkdirs(dest: str | PathLike, names: list[str]) -> None:
    """
    Creates directories inside a destination directory.

    Args:
        dest (str | Path): The destination directory.
        names (list[str]): A list of directory names to be created in the destination directory.
    """
    if not exists(dest):
        mkdir(dest)

    for i in names:
        if not exists(i):
            path = f"{dest}{SEP}{i}"
            mkdir(path)

yield_files_in ¶

yield_files_in(directory: str | PathLike, ext: str | tuple[str, ...] | None = None, *, recursive: bool = True, abs: bool = True) -> Generator[str, None, None]

Yields the paths of files in a directory.

This function searches for files in a directory and yields their paths. If the ext parameter is provided, only files with that extension are yielded. The ext parameter is case-insensitive. If the recursive parameter is set to True, the function will search for files in subdirectories recursively.

Parameters:

Name	Type	Description	Default
`directory`	`str \| Path`	The directory to search.	required
`ext`	`str \| tuple[str, ...]`	If provided, only yield files with provided extensions.	`None`
`recursive`	`bool`	Whether to search recursively.	`True`
`abs`	`bool`	Whether to convert paths to absolute paths.	`True`

Yields:

Type	Description
`str`	Generator[str, None, None]: The absolute paths of the files in the directory, matching the provided extension.

Source code in stdl/fs.py

def yield_files_in(
    directory: str | PathLike,
    ext: str | tuple[str, ...] | None = None,
    *,
    recursive: bool = True,
    abs: bool = True,  # noqa: A002
) -> Generator[str, None, None]:
    """
    Yields the paths of files in a directory.

    This function searches for files in a directory and yields their paths.
    If the `ext` parameter is provided, only files with that extension are yielded. The `ext` parameter is case-insensitive.
    If the `recursive` parameter is set to `True`, the function will search for files in subdirectories recursively.

    Args:
        directory (str | Path): The directory to search.
        ext (str | tuple[str, ...], optional): If provided, only yield files with provided extensions.
        recursive (bool, optional): Whether to search recursively.
        abs (bool, optional): Whether to convert paths to absolute paths.

    Yields:
        Generator[str, None, None]: The absolute paths of the files in the directory, matching the provided extension.
    """
    normalized_ext = _normalized_ext(ext)

    for file_path in _iter_file_paths(directory, recursive=recursive):
        if normalized_ext is not None and not file_path.lower().endswith(normalized_ext):
            continue

        yield os.path.abspath(file_path) if abs else file_path

get_files_in ¶

get_files_in(directory: str | PathLike, ext: str | tuple[str, ...] | None = None, *, recursive: bool = True, abs: bool = True) -> list[str]

Returns the paths of files in a directory.

This function searches for files in a directory and yields their paths. If the ext parameter is provided, only files with that extension are returned. The ext parameter is case-insensitive. If the recursive parameter is set to True, the function will search for files in subdirectories recursively.

Parameters:

Name	Type	Description	Default
`directory`	`str \| Path`	The directory to search.	required
`ext`	`str \| tuple[str, ...]`	If provided, only yield files with provided extensions. Defaults to None.	`None`
`recursive`	`bool`	Whether to search recursively. Defaults to True.	`True`
`abs`	`bool`	Whether to convert paths to absolute paths.	`True`

Returns:

Type	Description
`list[str]`	list[str]: The absolute path of the files in the directory, matching the provided extension.

Source code in stdl/fs.py

def get_files_in(
    directory: str | PathLike,
    ext: str | tuple[str, ...] | None = None,
    *,
    recursive: bool = True,
    abs: bool = True,  # noqa: A002
) -> list[str]:
    """
    Returns the paths of files in a directory.

    This function searches for files in a directory and yields their paths.
    If the `ext` parameter is provided, only files with that extension are returned. The `ext` parameter is case-insensitive.
    If the `recursive` parameter is set to `True`, the function will search for files in subdirectories recursively.

    Args:
        directory (str | Path): The directory to search.
        ext (str | tuple[str, ...], optional): If provided, only yield files with provided extensions. Defaults to None.
        recursive (bool, optional): Whether to search recursively. Defaults to True.
        abs (bool, optional): Whether to convert paths to absolute paths.

    Returns:
        list[str]: The absolute path of the files in the directory, matching the provided extension.
    """
    return list(yield_files_in(directory, ext, recursive=recursive, abs=abs))

yield_dirs_in ¶

yield_dirs_in(directory: str | PathLike, *, recursive: bool = True, abs: bool = True) -> Generator[str, None, None]

Yields paths to all directories in the specified directory. Yielded paths are converted to absolute paths.

Parameters:

Name	Type	Description	Default
`directory`	`str \| Path`	The directory to search.	required
`recursive`	`bool`	Whether to search recursively.	`True`
`abs`	`bool`	Whether to convert paths to absolute paths.	`True`

Yields:

Type	Description
`str`	Generator[str, None, None]: The paths of the directories that are found during travelsal.

Source code in stdl/fs.py

def yield_dirs_in(
    directory: str | PathLike,
    *,
    recursive: bool = True,
    abs: bool = True,  # noqa: A002
) -> Generator[str, None, None]:
    """
    Yields paths to all directories in the specified directory.
    Yielded paths are converted to absolute paths.

    Args:
        directory (str | Path): The directory to search.
        recursive (bool, optional): Whether to search recursively.
        abs (bool, optional): Whether to convert paths to absolute paths.


    Yields:
        Generator[str, None, None]: The paths of the directories that are found during travelsal.
    """
    visited = {_dir_identity(os.fspath(directory))}
    queue: Queue[str | PathLike[str]] = Queue()
    queue.put(directory)
    while not queue.empty():
        next_dir = queue.get()
        with os.scandir(next_dir) as entries:
            for entry in sorted(entries, key=lambda entry: os.fspath(entry.name)):
                if entry.is_dir():
                    entry_identity = _entry_dir_identity(entry)
                    if recursive and entry_identity not in visited:
                        visited.add(entry_identity)
                        queue.put(entry.path)

                    yield os.path.abspath(entry.path) if abs else entry.path

get_dirs_in ¶

get_dirs_in(directory: str | PathLike, *, recursive: bool = True, abs: bool = True) -> list[str]

Returns all directories in the specified directory. Returned paths are converted to absolute paths.

Parameters:

Name	Type	Description	Default
`directory`	`str \| Path`	The directory to search.	required
`recursive`	`bool`	Whether to search recursively.	`True`
`abs`	`bool`	Whether to convert paths to absolute paths.	`True`

Returns:

Type	Description
`list[str]`	list[str]: The paths of the directories that are found during travelsal.

Source code in stdl/fs.py

def get_dirs_in(
    directory: str | PathLike,
    *,
    recursive: bool = True,
    abs: bool = True,  # noqa: A002
) -> list[str]:
    """
    Returns all directories in the specified directory.
    Returned paths are converted to absolute paths.

    Args:
        directory (str | Path): The directory to search.
        recursive (bool, optional): Whether to search recursively.
        abs (bool, optional): Whether to convert paths to absolute paths.

    Returns:
        list[str]: The paths of the directories that are found during travelsal.
    """
    return list(yield_dirs_in(directory, recursive=recursive, abs=abs))

ensure_paths_exist ¶

ensure_paths_exist(*args: str | PathLike | Iterable[str | PathLike]) -> None

Ensures that the specified paths exist.

Parameters:

Name	Type	Description	Default
`*args`		one or more strings representing the paths to check. Can also include an Iterable of paths.	required

Raises:

Type	Description
`FileNotFoundError`	if one of the provided paths does not exist.

Source code in stdl/fs.py

def ensure_paths_exist(*args: str | PathLike | Iterable[str | PathLike]) -> None:
    """
    Ensures that the specified paths exist.

    Args:
        *args : one or more strings representing the paths to check. Can also include an Iterable of paths.

    Raises:
        FileNotFoundError : if one of the provided paths does not exist.
    """

    def check_path(path: str | PathLike) -> None:
        if not os.path.exists(os.fspath(path)):
            raise FileNotFoundError(f"Path does not exist: '{path}'")

    for path in args:
        if isinstance(path, (str, bytes, PathLike)):
            check_path(path)
        elif isinstance(path, Iterable):
            for i in path:
                check_path(i)

ensure_paths_dont_exist ¶

ensure_paths_dont_exist(*args: str | PathLike | Iterable[str | PathLike]) -> None

Ensures that the specified paths don't exist.

Parameters:

Name	Type	Description	Default
`*args`		one or more strings representing the paths to check. Can also include an Iterable of paths.	required

Raises:

Type	Description
`FileNotFoundError`	if one of the provided paths exists.

Source code in stdl/fs.py

def ensure_paths_dont_exist(*args: str | PathLike | Iterable[str | PathLike]) -> None:
    """
    Ensures that the specified paths don't exist.

    Args:
        *args : one or more strings representing the paths to check. Can also include an Iterable of paths.

    Raises:
        FileNotFoundError : if one of the provided paths exists.
    """

    def check_path(path: str | PathLike) -> None:
        if os.path.exists(os.fspath(path)):
            raise FileExistsError(f"Path already exists: '{path}'")

    for path in args:
        if isinstance(path, (str, bytes, PathLike)):
            check_path(path)
        elif isinstance(path, Iterable):
            for i in path:
                check_path(i)

exec_cmd ¶

exec_cmd(cmd: list[str] | str, timeout: float | None = None, shell: bool = False, capture_output: bool = True, check: bool = False, cwd: str | None = None, stdin: IO | None = None, stdout: IO | None = None, stderr: IO | None = None, input: str | None | bytes = None, env: dict[str, str] | None = None, text: bool = True, *args: SubprocessRunArg, **kwargs: SubprocessRunKwarg) -> CompletedCommand

Wrapper for subprocess.run with nicer default arguments.

Parameters:

Name	Type	Description	Default
`cmd`	`str \| list[str]`	command to run. A list of strings or a single string.	required
`timeout`	`float`	the time after which the command is killed.	`None`
`shell`	`bool`	whether or not to run the command in a shell.	`False`
`capture_output`	`bool`	whether or not to capture the output to stdout and stderr.	`True`
`check`	`bool`	whether or not to raise an exception on a non-zero exit code.	`False`
`cwd`	`str`	current working directory to run the command in.	`None`
`stdin`	`IO`	file object to read stdin from.	`None`
`stdout`	`IO`	file object to write stdout to.	`None`
`stderr`	`IO`	file object to write stderr to.	`None`
`input`	`str \| bytes`	input to send to the command.	`None`
`env`	`dict`	environment variables to pass to the new process.	`None`
`text`	`bool`	whether or not to return output as text or bytes.	`True`
`*args`		additional arguments to pass to subprocess.run.	required
`**kwargs`		additional keyword arguments to pass to subprocess.run.	required

Returns:

Type	Description
`CompletedCommand`	subprocess.CompletedProcess : the completed process.

Source code in stdl/fs.py

def exec_cmd(
    cmd: list[str] | str,
    timeout: float | None = None,
    shell: bool = False,
    capture_output: bool = True,
    check: bool = False,
    cwd: str | None = None,
    stdin: IO | None = None,
    stdout: IO | None = None,
    stderr: IO | None = None,
    input: str | None | bytes = None,
    env: dict[str, str] | None = None,
    text: bool = True,
    *args: SubprocessRunArg,
    **kwargs: SubprocessRunKwarg,
) -> CompletedCommand:
    """
    Wrapper for subprocess.run with nicer default arguments.

    Args:
        cmd (str | list[str]): command to run. A list of strings or a single string.
        timeout (float, optional): the time after which the command is killed.
        shell (bool, optional): whether or not to run the command in a shell.
        capture_output (bool, optional): whether or not to capture the output to stdout and stderr.
        check (bool, optional): whether or not to raise an exception on a non-zero exit code.
        cwd (str, optional): current working directory to run the command in.
        stdin (IO, optional): file object to read stdin from.
        stdout (IO, optional): file object to write stdout to.
        stderr (IO, optional): file object to write stderr to.
        input (str | bytes): input to send to the command.
        env (dict, optional): environment variables to pass to the new process.
        text (bool): whether or not to return output as text or bytes.
        *args : additional arguments to pass to subprocess.run.
        **kwargs : additional keyword arguments to pass to subprocess.run.

    Returns:
        subprocess.CompletedProcess : the completed process.
    """
    if isinstance(cmd, str):
        cmd = shlex.split(cmd)

    start_time = time.time()

    result = subprocess.run(  # noqa: S603
        cmd,
        *args,
        timeout=timeout,
        shell=shell,
        check=check,
        stdin=stdin,
        stdout=stdout,
        stderr=stderr,
        capture_output=capture_output,
        text=text,
        input=input,
        env=env,
        cwd=cwd,
        **kwargs,
    )

    return CompletedCommand(
        result.args,
        result.returncode,
        time.time() - start_time,
        result.stdout,
        result.stderr,
    )

read_piped ¶

read_piped() -> str

Reads piped input from stdin.

Source code in stdl/fs.py

def read_piped() -> str:
    """Reads piped input from stdin."""
    if sys.stdin.isatty():
        return ""

    return sys.stdin.read().strip()

start_file ¶

start_file(path: str | PathLike) -> None

Open the file with your OS's default application.

This function determines the current operating system and uses the appropriate command to open the specified file with the default application. It supports Windows, macOS, and Linux, including Windows Subsystem for Linux (WSL).

Source code in stdl/fs.py

def start_file(path: str | PathLike) -> None:
    """
    Open the file with your OS's default application.

    This function determines the current operating system and uses the appropriate command
    to open the specified file with the default application. It supports Windows, macOS,
    and Linux, including Windows Subsystem for Linux (WSL).
    """
    path = os.fspath(path)
    if is_wsl():
        exec_cmd(f"cmd.exe /C start '{path}'", check=True)
    elif sys.platform == "win32":
        os.startfile(path)  # noqa: S606
    elif sys.platform == "darwin":
        exec_cmd(f"open {path}", check=True)
    elif sys.platform == "linux":
        exec_cmd(f"xdg-open '{path}'", check=True)
    else:
        raise NotImplementedError(f"Unsupported platform: {sys.platform}")

fs

CompletedCommand ¶

stdout_lines property ¶

stderr_lines property ¶

PathBase ¶

path property ¶

nodes property ¶

created property ¶

modified property ¶

accessed property ¶

basename property ¶

abspath property ¶

parents property ¶

is_absolute property ¶

is_symlink property ¶

exists property ¶

parent property ¶

resolve ¶

resolve_async async ¶

created_async async ¶

modified_async async ¶

accessed_async async ¶

ctime_async async ¶

mtime_async async ¶

atime_async async ¶

is_symlink_async async ¶

rename ¶

rename_async async ¶

chmod ¶

chmod_async async ¶

chown ¶

chown_async async ¶

should_exist ¶

should_exist_async async ¶

should_not_exist ¶

should_not_exist_async async ¶

to_path ¶

to_str ¶

stat ¶

stat_async async ¶

samefile ¶

samefile_async async ¶

readlink ¶

readlink_async async ¶

owner ¶

owner_async async ¶

group ¶

group_async async ¶

relative_to ¶

expanduser ¶

as_posix ¶

as_uri ¶

match ¶

with_name ¶

exists_async async ¶

size_async async ¶

size_readable_async async ¶

remove_async async ¶

delete_async async ¶

move_to ¶

copy_to ¶

get_xattr ¶

get_xattr_async async ¶

set_xattr ¶

set_xattr_async async ¶

remove_xattr ¶

remove_xattr_async async ¶

File ¶

exists property ¶

parent property ¶

dirname property ¶

ext property ¶

suffix property ¶

stem property ¶

size property ¶

size_readable property ¶

create ¶

create_async async ¶

touch ¶

touch_async async ¶

stdout_lines `property` ¶

stderr_lines `property` ¶

path `property` ¶

nodes `property` ¶

created `property` ¶

modified `property` ¶

accessed `property` ¶

basename `property` ¶

abspath `property` ¶

parents `property` ¶

is_absolute `property` ¶

is_symlink `property` ¶

exists `property` ¶

parent `property` ¶

resolve_async `async` ¶

created_async `async` ¶

modified_async `async` ¶

accessed_async `async` ¶

ctime_async `async` ¶

mtime_async `async` ¶

atime_async `async` ¶

is_symlink_async `async` ¶

rename_async `async` ¶

chmod_async `async` ¶

chown_async `async` ¶

should_exist_async `async` ¶

should_not_exist_async `async` ¶

stat_async `async` ¶

samefile_async `async` ¶

readlink_async `async` ¶

owner_async `async` ¶

group_async `async` ¶

exists_async `async` ¶

size_async `async` ¶

size_readable_async `async` ¶

remove_async `async` ¶

delete_async `async` ¶

get_xattr_async `async` ¶

set_xattr_async `async` ¶

remove_xattr_async `async` ¶

exists `property` ¶

parent `property` ¶

dirname `property` ¶

ext `property` ¶

suffix `property` ¶

stem `property` ¶

size `property` ¶

size_readable `property` ¶

create_async `async` ¶

touch_async `async` ¶

clear_async `async` ¶

read_async `async` ¶

write_async `async` ¶

append_async `async` ¶

open_async `async` ¶

write_iter_async `async` ¶

append_iter_async `async` ¶

readlines_async `async` ¶

splitlines_async `async` ¶

move_to_async `async` ¶

copy_to_async `async` ¶

link_async `async` ¶

symlink_async `async` ¶

rand `classmethod` ¶

size `property` ¶

size_readable `property` ¶

exists `property` ¶

parent `property` ¶

create_async `async` ¶

clear_async `async` ¶

move_to_async `async` ¶

copy_to_async `async` ¶

remove_async `async` ¶

yield_files_async `async` ¶

get_files_async `async` ¶

read_files_async `async` ¶

remove_files_async `async` ¶

yield_subdirs_async `async` ¶

get_subdirs_async `async` ¶

yield_entries_async `async` ¶