sh

shell utils

Classes:

• Done –

  Completed subprocess

• DoneDict –

  Completed subprocess as a typed-dict; see Done

• DoneError –

  Error raised when a process returns a non-zero/not-ok exit status

• HrTime –

  High resolution time split into whole seconds and nanoseconds

• HrTimeDict –

  High resolution time as a typed-dict; see HrTime

• Stdio –

  Standard-io enum object; values are the standard file descriptors

• FlagMeta –

  Metaclass turning attribute access on Flag into flags

• Flag –

  Namespace that turns any attribute into the corresponding cli flag

• LIN –

  Linux (and Mac) shell commands/methods container

• WIN –

  Windows shell commands/methods container

Functions:

• echo –

  Print/echo function

• chmod –

  Change the access permissions of a file

• copy_file –

  Copy a file given a source-path and a destination-path

• cp –

  Copy the directory/file src to the directory/file dest

• dir_exists –

  Return True if the given path exists; False otherwise; alias for isdir

• dir_exists_async –

  Return True if the directory exists; False otherwise

• dirpath_gen –

  Yield all dirpaths as pathlib.Path objects beneath a dirpath

• dirs_gen –

  Yield directory-paths beneath a dirpath (defaults to os.getcwd())

• exists –

  Return True if the given path exists; False otherwise

• extension –

  Return the extension for a fspath

• file_exists –

  Return True if the given path exists; False otherwise; alias for isfile

• file_exists_async –

  Return True if the file exists; False otherwise

• file_lines_gen –

  Yield lines from a given fspath

• filecmp –

  Compare 2 files for equality given their filepaths

• filepath_gen –

  Yield all filepaths as pathlib.Path objects beneath a dirpath

• filepath_mtimedelta_sec –

  Return the seconds since the file(path) was last modified

• files_dirs_gen –

  Return a files_gen() and a dirs_gen() in one swell-foop

• files_gen –

  Yield file-paths beneath a given dirpath (defaults to os.getcwd())

• filesize –

  Return the size of the given file(path) in bytes

• filesize_async –

  Return the size of the file at the given fspath

• fspath –

  Alias for os._fspath; returns fspath string for any type of path

• glob –

  Return an iterator of fspaths matching the given glob pattern

• is_dir –

  Return True if the given path is a directory; alias for isdir

• is_dir_async –

  Return True if the given path is a file; False otherwise

• is_file –

  Return True if the given path is a file; alias for isfile

• is_file_async –

  Return True if the given path is a file; False otherwise

• is_link –

  Return True if the given path is a link; alias for islink

• is_link_async –

  Return True if the given path is a link; False otherwise

• isdir –

  Return True if the given path is a directory; False otherwise

• isdir_async –

  Return True if the given path is a file; False otherwise

• isfile –

  Return True if the given path is a file; False otherwise

• isfile_async –

  Return True if the given path is a file; False otherwise

• islink –

  Return True if the given path is a link; False otherwise

• islink_async –

  Return True if the given path is a link; False otherwise

• listdir_async –

  Async version of os.listdir

• listdir_gen –

  Return an iterator of strings from DirEntries

• lstat_async –

  Async version of os.lstat

• mkdir –

  Make directory at given fspath

• mkdirp –

  Make directory and parents

• move –

  Move file(s) like on the command line

• path_gen –

  Yield all filepaths as pathlib.Path objects beneath a dirpath

• read_bytes –

  Read bytes from a fspath

• read_bytes_async –

  (ASYNC) Load/Read bytes from a fspath

• read_bytes_gen –

  Yield bytes from a given fspath

• read_bytes_gen_async –

  Yield (asynchronously) bytes from a given fspath

• read_json –

  Load/Read-&-parse json data given a fspath

• read_json_async –

  Load/Read-&-parse json data given a fspath

• read_str –

  Load/Read a string given a fspath

• read_str_async –

  (ASYNC) Load/Read a string given a fspath

• rm_gen –

  Remove files & directories in the style of the shell

• rmdir –

  Remove directory at given fspath

• rmfile –

  Remove a file at given fspath

• safepath –

  Check if a file/dir path is save/unused; returns an unused path.

• scandir –

  Typed version of os.scandir

• scandir_gen –

  Return an iterator of os.DirEntry objects

• scandir_list –

  Return a list of os.DirEntry objects

• sep_join –

  Join iterable of strings on the current platform os.path.sep value

• sep_lstrip –

  Left-strip a string of the current platform's os.path.sep value

• sep_rstrip –

  Right-strip a string of the current platform's os.path.sep value

• sep_split –

  Split a string on the current platform os.path.sep value

• sep_strip –

  Strip a string of the current platform's os.path.sep value

• shebang –

  Get the shebang string given a fspath; Returns None if no shebang

• stat –

  Return the os.stat_result object for a given fspath

• stat_async –

  Async version of os.stat

• touch –

  Create an empty file given a fspath

• walk_gen –

  Yield all paths beneath a given dirpath (defaults to os.getcwd())

• write_bytes –

  Write/Save bytes to a fspath

• write_bytes_async –

  (ASYNC) Write/Save bytes to a fspath

• write_bytes_gen –

  Write/Save bytes to a fspath

• write_bytes_gen_async –

  Write/save bytes to a filepath from an (async)iterable/iterator of bytes

• write_json –

  Save/Write json-serial-ize-able data to a fspath

• write_json_async –

  Save/Write json-serial-ize-able data to a fspath

• write_str –

  Save/Write a string to fspath

• write_str_async –

  (ASYNC) Save/Write a string to fspath

• mkenv –

  Return the environment dict to run a subprocess with

• decode_stdio_bytes –

  Return Stdio bytes from stdout/stderr as a string

• pstdout –

  Get the STDOUT as a string from a subprocess

• pstderr –

  Get the STDERR as a string from a subprocess

• pstdout_pstderr –

  Get the STDOUT and STDERR as strings from a subprocess

• flatten_args –

  Flatten possibly nested iterables of sequences to a list of strings

• do –

  Run a subprocess synchronously

• shell –

  Run a subprocess synchronously in current shell

• do_asyncify –

  Run a subprocess asynchronously using asyncified version of do

• do_async –

  Run a subprocess and await its completion

• doa –

  Run a subprocess and await its completion

• pwd –

  Return present-working-directory path string; alias for os.getcwd

• dirname –

  Return dirname/parent-dir of given path; alias of os.path.dirname

• basename –

  Return the basename of given path; alias of os.path.dirname

• cd –

  Change directory to given dirpath; alias for os.chdir

• export –

  Export/Set an environment variable

• setenv –

  Export/Set an environment variable

• shplit –

  Typed alias for shlex.split

• quote –

  Typed alias for shlex.quote

• q –

  Typed alias for shlex.quote

• which –

  Return the result of shutil.which

• where –

  Return the result of shutil.which; alias of shellfish.sh.which

• which_lru –

  Return the result of shutil.which and cache the results

• tree –

  Create a directory tree string given a directory path

• ls –

  List files and dirs given a dirpath (defaults to pwd)

• ls_files –

  List the files in a given directory path

• ls_dirs –

  List the directories in a given directory path

• ls_files_dirs –

  List the files and directories given directory path

• ls_async –

  List files and dirs given a dirpath (defaults to pwd)

• rm –

  Remove files & directories in the style of the shell

• mv –

  Move file(s) like on the command line

• source –

  Execute/run a python file given a fspath and put globals in globasl

Done

Bases: _ShellfishBaseModel

Completed subprocess

Returned by shellfish.sh.do (and friends) once a process has finished.

Examples:

>>> done = Done(
...     args=["echo", "hello"],
...     returncode=0,
...     stdout="hello\nworld\n",
...     stderr="",
...     ti=0.0,
...     tf=0.5,
...     dt=0.5,
... )
>>> done.returncode
0
>>> done.lines
['hello', 'world']
>>> done.grep("world")
['world']
>>> done.check()  # does not raise; returncode is 0

Methods:

• model_post_init –

  Pydantic post-init hook; defers to __post_init__

• hrdt_dict –

  Return the high resolution run-time as a typed-dict

• stdout_lines –

  Return stdout split into lines

• stderr_lines –

  Return stderr split into lines

• done_dict –

  Return Done object as typed-dict

• check –

  Check returncode and stderr

• sys_print –

  Write self.stdout to sys.stdout and self.stderr to sys.stderr

• write_stdout –

  Write stdout as a string to a fspath

• completed_process –

  Return subprocess.CompletedProcess object

• write_stderr –

  Write stderr as a string to a fspath

• json_parse_stdout –

  Return json parsed stdout

• json_parse_stderr –

  Return json parsed stderr

• json_parse –

  Return json parsed stdout (or stderr)

• parse_json –

  Alias for json_parse

• grep –

  Return lines in stdout that contain the given string

Attributes:

• args (list[str]) –

  Command args the process was run with

• returncode (int) –

  Exit status of the process

• stdout (str) –

  Standard output (stdout) of the process

• stderr (str) –

  Standard error (stderr) of the process

• ti (float) –

  Time the process started (seconds since epoch)

• tf (float) –

  Time the process finished (seconds since epoch)

• dt (float) –

  Time the process took to run (seconds; tf - ti)

• hrdt (HrTime | None) –

  High resolution dt, if the runner provided one

• stdin (str | None) –

  Standard input (stdin) written to the process, if any

• async_proc (bool) –

  True if the process was run asynchronously

• dryrun (bool) –

  True if the process was not actually run (dryrun)

• verbose (bool) –

  Echo stdout/stderr to the parent process on init; excluded from dumps

• lines (list[str]) –

  Stdout split into lines without line-endings

args instance-attribute

args: list[str]

Command args the process was run with

returncode instance-attribute

returncode: int

Exit status of the process

stdout instance-attribute

stdout: str

Standard output (stdout) of the process

stderr instance-attribute

stderr: str

Standard error (stderr) of the process

ti instance-attribute

ti: float

Time the process started (seconds since epoch)

tf instance-attribute

tf: float

Time the process finished (seconds since epoch)

dt instance-attribute

dt: float

Time the process took to run (seconds; tf - ti)

hrdt class-attribute instance-attribute

hrdt: HrTime | None = None

High resolution dt, if the runner provided one

stdin class-attribute instance-attribute

stdin: str | None = None

Standard input (stdin) written to the process, if any

async_proc class-attribute instance-attribute

async_proc: bool = False

True if the process was run asynchronously

dryrun class-attribute instance-attribute

dryrun: bool = Field(False)

True if the process was not actually run (dryrun)

verbose class-attribute instance-attribute

verbose: bool = Field(False, exclude=True)

Echo stdout/stderr to the parent process on init; excluded from dumps

lines property

lines: list[str]

Stdout split into lines without line-endings

model_post_init

model_post_init(_context: Any) -> None

Pydantic post-init hook; defers to __post_init__

hrdt_dict

hrdt_dict() -> HrTimeDict

Return the high resolution run-time as a typed-dict

Falls back to converting dt to HrTime when the runner did not provide an hrdt.

stdout_lines

stdout_lines(*, keepends: bool = False) -> list[str]

Return stdout split into lines

Parameters:

• keepends

  (bool, default: False ) –

  Keep the line-ending characters on each line

Returns:

• list[str] –

  List of stdout lines

stderr_lines

stderr_lines(*, keepends: bool = False) -> list[str]

Return stderr split into lines

Parameters:

• keepends

  (bool, default: False ) –

  Keep the line-ending characters on each line

Returns:

• list[str] –

  List of stderr lines

done_dict

done_dict() -> DoneDict

Return Done object as typed-dict

check

check(
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
) -> None

Check returncode and stderr

Parameters:

• ok_code

  (int | list[int] | tuple[int, ...] | set[int], default: 0 ) –

  Return code (or collection of return codes) considered ok

Raises:

• DoneError –

  If the return code is not ok

sys_print

sys_print() -> None

Write self.stdout to sys.stdout and self.stderr to sys.stderr

write_stdout

write_stdout(
    filepath: FsPath, *, append: bool = False
) -> None

Write stdout as a string to a fspath

Parameters:

• filepath

  (FsPath) –

  Filepath to write stdout to

• append

  (bool, default: False ) –

  Append to the file instead of overwriting it

completed_process

completed_process() -> CompletedProcess[str]

Return subprocess.CompletedProcess object

write_stderr

write_stderr(
    filepath: FsPath, *, append: bool = False
) -> None

Write stderr as a string to a fspath

Parameters:

• filepath

  (FsPath) –

  Filepath of location to write stderr

• append

  (bool, default: False ) –

  Append to the file instead of overwriting it

json_parse_stdout

json_parse_stdout(
    *,
    jsonc: bool = False,
    jsonl: bool = False,
    ndjson: bool = False,
) -> Any

Return json parsed stdout

Parameters:

• jsonc

  (bool, default: False ) –

  Parse stdout as jsonc (json with comments)

• jsonl

  (bool, default: False ) –

  Parse stdout as jsonl (json-lines)

• ndjson

  (bool, default: False ) –

  Parse stdout as ndjson (newline delimited json)

Returns:

• Any –

  The parsed stdout

json_parse_stderr

json_parse_stderr(
    *,
    jsonc: bool = False,
    jsonl: bool = False,
    ndjson: bool = False,
) -> Any

Return json parsed stderr

Parameters:

• jsonc

  (bool, default: False ) –

  Parse stderr as jsonc (json with comments)

• jsonl

  (bool, default: False ) –

  Parse stderr as jsonl (json-lines)

• ndjson

  (bool, default: False ) –

  Parse stderr as ndjson (newline delimited json)

Returns:

• Any –

  The parsed stderr

json_parse

json_parse(
    *,
    stderr: bool = False,
    jsonc: bool = False,
    jsonl: bool = False,
    ndjson: bool = False,
) -> Any

Return json parsed stdout (or stderr)

Parameters:

• stderr

  (bool, default: False ) –

  Parse stderr instead of stdout

• jsonc

  (bool, default: False ) –

  Parse as jsonc (json with comments)

• jsonl

  (bool, default: False ) –

  Parse as jsonl (json-lines)

• ndjson

  (bool, default: False ) –

  Parse as ndjson (newline delimited json)

Returns:

• Any –

  The parsed stdout, or the parsed stderr if stderr is True

parse_json

parse_json(
    *,
    stderr: bool = False,
    jsonc: bool = False,
    jsonl: bool = False,
    ndjson: bool = False,
) -> Any

Alias for json_parse

(bc I keep flip-flopping the fn name)

Parameters:

• stderr

  (bool, default: False ) –

  Parse stderr instead of stdout

• jsonc

  (bool, default: False ) –

  Parse as jsonc (json with comments)

• jsonl

  (bool, default: False ) –

  Parse as jsonl (json-lines)

• ndjson

  (bool, default: False ) –

  Parse as ndjson (newline delimited json)

Returns:

• Any –

  The parsed stdout, or the parsed stderr if stderr is True

grep

grep(string: str) -> list[str]

Return lines in stdout that contain the given string

Parameters:

• string

  (str) –

  String to search for

Returns:

• list[str] –

  list[str]: List of strings of stdout lines containing the given search string

DoneDict

Bases: TypedDict

Completed subprocess as a typed-dict; see Done

Attributes:

• args (list[str]) –

  Command args the process was run with

• returncode (int) –

  Exit status of the process

• stdout (str) –

  Standard output (stdout) of the process

• stderr (str) –

  Standard error (stderr) of the process

• ti (float) –

  Time the process started (seconds since epoch)

• tf (float) –

  Time the process finished (seconds since epoch)

• dt (float) –

  Time the process took to run (seconds; tf - ti)

• hrdt (HrTimeDict | None) –

  High resolution dt, if the runner provided one

• stdin (str | None) –

  Standard input (stdin) written to the process, if any

• async_proc (bool) –

  True if the process was run asynchronously

• verbose (bool) –

  True if stdout/stderr were echoed to the parent process' stdout/stderr

args instance-attribute

args: list[str]

Command args the process was run with

returncode instance-attribute

returncode: int

Exit status of the process

stdout instance-attribute

stdout: str

Standard output (stdout) of the process

stderr instance-attribute

stderr: str

Standard error (stderr) of the process

ti instance-attribute

ti: float

Time the process started (seconds since epoch)

tf instance-attribute

tf: float

Time the process finished (seconds since epoch)

dt instance-attribute

dt: float

Time the process took to run (seconds; tf - ti)

hrdt instance-attribute

hrdt: HrTimeDict | None

High resolution dt, if the runner provided one

stdin instance-attribute

stdin: str | None

Standard input (stdin) written to the process, if any

async_proc instance-attribute

async_proc: bool

True if the process was run asynchronously

verbose instance-attribute

verbose: bool

True if stdout/stderr were echoed to the parent process' stdout/stderr

DoneError

DoneError(done: Done)

Bases: SubprocessError

Error raised when a process returns a non-zero/not-ok exit status

Raised by Done.check.

Examples:

>>> done = Done(
...     args=["sh", "-c", "exit 1"],
...     returncode=1,
...     stdout="",
...     stderr="uh oh\n",
...     ti=0.0,
...     tf=0.1,
...     dt=0.1,
... )
>>> try:
...     done.check()
... except DoneError as e:
...     (e.returncode, e.cmd, e.stderr)
(1, ['sh', '-c', 'exit 1'], 'uh oh\n')

Parameters:

• done

  (Done) –

  Done object with a non-zero/not-ok returncode

Methods:

• error_msg –

  Return the error message string for this error's returncode

Attributes:

• done (Done) –

  The Done object that produced this error

• returncode (int) –

  Exit status of the process

• stdout (str) –

  Standard output (stdout) of the process

• stderr (str) –

  Standard error (stderr) of the process

• cmd (list[str]) –

  Command args the process was run with

• output (str) –

  Alias for stdout; mirrors subprocess.CalledProcessError.output

done instance-attribute

done: Done = done

The Done object that produced this error

returncode instance-attribute

returncode: int = done.returncode

Exit status of the process

stdout instance-attribute

stdout: str = done.stdout

Standard output (stdout) of the process

stderr instance-attribute

stderr: str = done.stderr

Standard error (stderr) of the process

cmd instance-attribute

cmd: list[str] = done.args

Command args the process was run with

output property writable

output: str

Alias for stdout; mirrors subprocess.CalledProcessError.output

error_msg

error_msg() -> str

Return the error message string for this error's returncode

HrTime

Bases: _ShellfishBaseModel

High resolution time split into whole seconds and nanoseconds

Examples:

>>> HrTime.from_seconds(1.5)
HrTime(secs=1, nanos=500000000)
>>> HrTime.from_seconds(1.5).hrdt_dict()
{'secs': 1, 'nanos': 500000000}

Methods:

• from_seconds –

  Return HrTime object from seconds

• hrdt_dict –

  Return this HrTime as a typed-dict

Attributes:

• secs (int) –

  Whole seconds

• nanos (int) –

  Nanoseconds remainder (0 <= nanos < 1_000_000_000)

• sec (int) –

  Deprecated alias for secs

• ns (int) –

  Deprecated alias for nanos

secs class-attribute instance-attribute

secs: int = Field(
    validation_alias=AliasChoices("sec", "secs", "s")
)

Whole seconds

nanos class-attribute instance-attribute

nanos: int = Field(
    validation_alias=AliasChoices("ns", "nsecs", "nanos")
)

Nanoseconds remainder (0 <= nanos < 1_000_000_000)

sec property

sec: int

Deprecated alias for secs

ns property

ns: int

Deprecated alias for nanos

from_seconds classmethod

from_seconds(seconds: float) -> HrTime

Return HrTime object from seconds

Parameters:

• seconds

  (float) –

  number of seconds

Returns:

• HrTime –

  HrTime object for the given number of seconds

hrdt_dict

hrdt_dict() -> HrTimeDict

Return this HrTime as a typed-dict

HrTimeDict

Bases: TypedDict

High resolution time as a typed-dict; see HrTime

Attributes:

• secs (int) –

  Whole seconds

• nanos (int) –

  Nanoseconds remainder (0 <= nanos < 1_000_000_000)

secs instance-attribute

secs: int

Whole seconds

nanos instance-attribute

nanos: int

Nanoseconds remainder (0 <= nanos < 1_000_000_000)

Stdio

Bases: IntEnum

Standard-io enum object; values are the standard file descriptors

Examples:

>>> from shellfish.stdio import Stdio
>>> Stdio.stdout
<Stdio.stdout: 1>
>>> int(Stdio.stderr)
2

Attributes:

• stdin –

  Standard input (fd 0)

• stdout –

  Standard output (fd 1)

• stderr –

  Standard error (fd 2)

stdin class-attribute instance-attribute

stdin = 0

Standard input (fd 0)

stdout class-attribute instance-attribute

stdout = 1

Standard output (fd 1)

stderr class-attribute instance-attribute

stderr = 2

Standard error (fd 2)

FlagMeta

Bases: type

Metaclass turning attribute access on Flag into flags

Methods:

• attr2flag –

  Return an attribute name as a cli flag (underscores to dashes)

attr2flag cached staticmethod

attr2flag(string: str) -> str

Return an attribute name as a cli flag (underscores to dashes)

Parameters:

• string

  (str) –

  Attribute name to convert

Returns:

• str –

  The attribute name with underscores replaced by dashes

Flag

Namespace that turns any attribute into the corresponding cli flag

Leading underscores become leading dashes and inner underscores become dashes, so Flag.__dry_run is '--dry-run'. Not meant to be instantiated.

Examples:

>>> Flag.__help
'--help'
>>> Flag._v
'-v'
>>> Flag.__dry_run
'--dry-run'

LIN

Bases: LIN

Linux (and Mac) shell commands/methods container

Extends shellfish.osfs.LIN with rsync/sync helpers. All members are static methods; the class is used as a namespace, not instantiated.

Methods:

• rsync_args –

  Return args for rsync command on linux/mac

• rsync –

  Run an rsync subprocess

• link_dir –

  Make a directory symlink

• link_dirs –

  Make multiple directory symlinks

• link_file –

  Make a file symlink

• link_files –

  Make multiple file symlinks

• unlink_dir –

  Unlink a directory symlink given a path to the symlink

• unlink_dirs –

  Unlink directory symlinks given the paths the links

• unlink_file –

  Unlink a file symlink given a path to the symlink

• unlink_files –

  Unlink directory symlinks given the paths the links

rsync_args staticmethod

rsync_args(
    src: str,
    dest: str,
    *,
    delete: bool = False,
    dry_run: bool = False,
    exclude: Iterable[str] | None = None,
    include: Iterable[str] | None = None,
) -> list[str]

Return args for rsync command on linux/mac

Parameters:

• src

  (str) –

  path to remote (raid) tdir

• dest

  (str) –

  path to local tdir

• delete

  (bool, default: False ) –

  Flag that will do a 'hard sync'

• exclude

  (Iterable[str] | None, default: None ) –

  Strings/patterns to exclude

• include

  (Iterable[str] | None, default: None ) –

  Strings/patterns to include

• dry_run

  (bool, default: False ) –

  Perform operation as a dry run

Returns:

• list[str] –

  subprocess return code from rsync

Rsync return codes::

- 0 == Success
- 1 == Syntax or usage error
- 2 == Protocol incompatibility
- 3 == Errors selecting input/output files, dirs
- 4 == Requested  action not supported: an attempt was made to
  manipulate 64-bit files on a platform that cannot support them;
  or an option was specified that is supported by the client and
  not the server.
- 5 == Error starting client-server protocol
- 6 == Daemon unable to append to log-file
- 10 == Error in socket I/O
- 11 == Error in file I/O
- 12 == Error in rsync protocol data stream
- 13 == Errors with program diagnostics
- 14 == Error in IPC code
- 20 == Received SIGUSR1 or SIGINT
- 21 == Some error returned by waitpid()
- 22 == Error allocating core memory buffers
- 23 == Partial transfer due to error
- 24 == Partial transfer due to vanished source files
- 25 == The --max-delete limit stopped deletions
- 30 == Timeout in data send2viewserver/receive
- 35 == Timeout waiting for daemon connection

rsync staticmethod

rsync(
    src: str,
    dest: str,
    *,
    delete: bool = False,
    mkdirs: bool = False,
    dry_run: bool = False,
    exclude: Iterable[str] | None = None,
    include: Iterable[str] | None = None,
) -> Done

Run an rsync subprocess

Parameters:

• mkdirs

  (bool, default: False ) –

  Make destination directories if they do not already exist; defaults to False.

• src

  (str) –

  Source directory path

• dest

  (str) –

  Destination directory path

• delete

  (bool, default: False ) –

  Delete files/directories in destination if they do exist in source

• exclude

  (Iterable[str] | None, default: None ) –

  Strings/patterns to exclude

• include

  (Iterable[str] | None, default: None ) –

  Strings/patterns to include

• dry_run

  (bool, default: False ) –

  Perform operation as a dry run

Returns:

• Done ( Done ) –

  Done object containing the info for the rsync run

Rsync return codes::

- 0 == Success
- 1 == Syntax or usage error
- 2 == Protocol incompatibility
- 3 == Errors selecting input/output files, dirs
- 4 == Requested  action not supported: an attempt was made to
  manipulate 64-bit files on a platform that cannot support them;
  or an option was specified that is supported by the client and
  not the server.
- 5 == Error starting client-server protocol
- 6 == Daemon unable to append to log-file
- 10 == Error in socket I/O
- 11 == Error in file I/O
- 12 == Error in rsync protocol data stream
- 13 == Errors with program diagnostics
- 14 == Error in IPC code
- 20 == Received SIGUSR1 or SIGINT
- 21 == Some error returned by waitpid()
- 22 == Error allocating core memory buffers
- 23 == Partial transfer due to error
- 24 == Partial transfer due to vanished source files
- 25 == The --max-delete limit stopped deletions
- 30 == Timeout in data send2viewserver/receive
- 35 == Timeout waiting for daemon connection

link_dir staticmethod

link_dir(
    linkpath: str,
    targetpath: str,
    *,
    exist_ok: bool = False,
) -> None

Make a directory symlink

Parameters:

• linkpath

  (str) –

  Path to the link to be made

• targetpath

  (str) –

  Path to the target of the link to be made

• exist_ok

  (bool, default: False ) –

  Allow link to exist

link_dirs staticmethod

link_dirs(
    link_target_tuples: list[tuple[str, str]],
    *,
    exist_ok: bool = False,
) -> None

Make multiple directory symlinks

Parameters:

• link_target_tuples

  (list[tuple[str, str]]) –

  Iterable of tuples of the form: (link, target) or a dictionary mapping with key => value pairs of the form link => target.

• exist_ok

  (bool, default: False ) –

  Allow link to exist

link_file staticmethod

link_file(
    linkpath: str,
    targetpath: str,
    *,
    exist_ok: bool = False,
) -> None

Make a file symlink

Parameters:

• linkpath

  (str) –

  Path to the link to be made

• targetpath

  (str) –

  Path to the target of the link to be made

• exist_ok

  (bool, default: False ) –

  Allow links to already exist

link_files staticmethod

link_files(
    link_target_tuples: list[tuple[str, str]],
    *,
    exist_ok: bool = False,
) -> None

Make multiple file symlinks

Parameters:

• exist_ok

  (bool, default: False ) –

  Allow links to already exist

• link_target_tuples

  (list[tuple[str, str]]) –

  Iterable of tuples of the form: (link, target) or a dictionary mapping with key => value pairs of the form link => target.

unlink_dir staticmethod

unlink_dir(link: str) -> None

Unlink a directory symlink given a path to the symlink

Parameters:

• link

  (str) –

  path to the symlink

unlink_dirs staticmethod

unlink_dirs(links: Iterable[str]) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (Iterable[str]) –

  Iterable of paths to links

unlink_file staticmethod

unlink_file(link: str) -> None

Unlink a file symlink given a path to the symlink

Parameters:

• link

  (str) –

  path to the symlink

unlink_files staticmethod

unlink_files(links: Iterable[str]) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (Iterable[str]) –

  Iterable of paths to links

WIN

Bases: WIN

Windows shell commands/methods container

Extends shellfish.osfs.WIN with robocopy/sync helpers. All members are static methods; the class is used as a namespace, not instantiated.

Methods:

• robocopy_args –

  Return list of robocopy command args

• robocopy –

  Robocopy wrapper function (crude in that it opens a subprocess)

• link_dir –

  Make a directory symlink

• link_dirs –

  Make multiple directory symlinks

• link_file –

  Make a file symlink

• link_files –

  Make multiple file symlinks

• unlink_dir –

  Unlink a directory symlink given a path to the symlink

• unlink_dirs –

  Unlink directory symlinks given the paths the links

• unlink_file –

  Unlink a file symlink given a path to the symlink

• unlink_files –

  Unlink directory symlinks given the paths the links

robocopy_args staticmethod

robocopy_args(
    src: str,
    dest: str,
    *,
    delete: bool = False,
    exclude_files: list[str] | None = None,
    exclude_dirs: list[str] | None = None,
    dry_run: bool = False,
) -> list[str]

Return list of robocopy command args

Parameters:

• src

  (str) –

  path to source directory

• dest

  (str) –

  path to destination directory

• delete

  (bool, default: False ) –

  Delete files in the destination directory if they do not exist in the source directory

• exclude_files

  (list[str] | None, default: None ) –

  Strings/patterns with which to exclude files

• exclude_dirs

  (list[str] | None, default: None ) –

  Strings/patterns with which to exclude directories

• dry_run

  (bool, default: False ) –

  Do the operation as a dry run

Returns:

• list[str] –

  subprocess return code from robocopy

Robocopy return codes::

0. No files were copied. No failure was encountered. No files were
   mismatched. The files already exist in the destination
   directory; therefore, the copy operation was skipped.
1. All files were copied successfully.
2. There are some additional files in the destination directory
   that are not present in the source directory. No files were
   copied.
3. Some files were copied. Additional files were present. No
   failure was encountered.
5. Some files were copied. Some files were mismatched. No failure
   was encountered.
6. Additional files and mismatched files exist. No files were
   copied and no failures were encountered. This means that the
   files already exist in the destination directory.
7. Files were copied, a file mismatch was present, and additional
   files were present.
8. Several files did not copy.

robocopy staticmethod

robocopy(
    src: str,
    dest: str,
    *,
    mkdirs: bool = True,
    delete: bool = False,
    exclude_files: Iterable[str] | None = None,
    exclude_dirs: Iterable[str] | None = None,
    dry_run: bool = False,
) -> Done

Robocopy wrapper function (crude in that it opens a subprocess)

Parameters:

• src

  (str) –

  path to source directory

• dest

  (str) –

  path to destination directory

• delete

  (bool, default: False ) –

  Delete files in the destination directory if they do not exist in the source directory

• exclude_files

  (Iterable[str] | None, default: None ) –

  Strings/patterns with which to exclude files

• exclude_dirs

  (Iterable[str] | None, default: None ) –

  Strings/patterns with which to exclude directories

• dry_run

  (bool, default: False ) –

  Do the operation as a dry run

• mkdirs

  (bool, default: True ) –

  Flag to make destinaation directories if they do not already exist

Returns:

• Done –

  subprocess return code from robocopy

Robocopy return codes::

0. No files were copied. No failure was encountered. No files were
   mismatched. The files already exist in the destination
   directory; therefore, the copy operation was skipped.
1. All files were copied successfully.
2. There are some additional files in the destination directory
   that are not present in the source directory. No files were
   copied.
3. Some files were copied. Additional files were present. No
   failure was encountered.
5. Some files were copied. Some files were mismatched. No failure
   was encountered.
6. Additional files and mismatched files exist. No files were
   copied and no failures were encountered. This means that the
   files already exist in the destination directory.
7. Files were copied, a file mismatch was present, and additional
   files were present.
8. Several files did not copy.

link_dir staticmethod

link_dir(
    linkpath: str,
    targetpath: str,
    *,
    exist_ok: bool = False,
) -> None

Make a directory symlink

Parameters:

• linkpath

  (str) –

  Path to the link to be made

• targetpath

  (str) –

  Path to the target of the link to be made

• exist_ok

  (bool, default: False ) –

  If True, do not raise an exception if the link exists

link_dirs staticmethod

link_dirs(
    link_target_tuples: list[tuple[str, str]],
    *,
    exist_ok: bool = False,
) -> None

Make multiple directory symlinks

Parameters:

• link_target_tuples

  (list[tuple[str, str]]) –

  Iterable of tuples of the form: (link, target) or a dictionary mapping with key => value pairs of the form link => target.

• exist_ok

  (bool, default: False ) –

  If True, do not raise an exception if the link(s) exist

link_file staticmethod

link_file(
    linkpath: str,
    targetpath: str,
    *,
    exist_ok: bool = False,
) -> None

Make a file symlink

Parameters:

• linkpath

  (str) –

  Path to the link to be made

• targetpath

  (str) –

  Path to the target of the link to be made

• exist_ok

  (bool, default: False ) –

  If True, don't raise an exception if the link exists

link_files staticmethod

link_files(
    link_target_tuples: list[tuple[str, str]],
    *,
    exist_ok: bool = False,
) -> None

Make multiple file symlinks

Parameters:

• link_target_tuples

  (list[tuple[str, str]]) –

  Iterable of tuples of the form: (link, target) or a dictionary mapping with key => value pairs of the form link => target.

• exist_ok

  (bool, default: False ) –

  If True, don't raise an exception if the link exists

unlink_dir staticmethod

unlink_dir(link: str) -> None

Unlink a directory symlink given a path to the symlink

Parameters:

• link

  (str) –

  path to the symlink

unlink_dirs staticmethod

unlink_dirs(links: Iterable[str]) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (Iterable[str]) –

  Iterable of paths to links

unlink_file staticmethod

unlink_file(link: str) -> None

Unlink a file symlink given a path to the symlink

Parameters:

• link

  (str) –

  path to the symlink

unlink_files staticmethod

unlink_files(links: Iterable[str]) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (Iterable[str]) –

  Iterable of paths to links

echo

echo(
    *objects: Any,
    sep: str = " ",
    end: str = "\n",
    file: IO[str] | None = None,
    flush: bool = False,
) -> None

Print/echo function

This function is basically the print function, and exists so that one can deliberately print without using the built-in print function which is not allowed by the dgpy-libs ruff rules.

Parameters:

• *objects

  (Any, default: () ) –

  Item(s) to print/echo

• sep

  (str, default: ' ' ) –

  Separator to print with

• end

  (str, default: '\n' ) –

  End of print suffix; defaults to \n

• file

  (IO[str] | None, default: None ) –

  File like object to write to if not stdout

• flush

  (bool, default: False ) –

  Flush the file after writing

Examples:

>>> echo("shellfish")
shellfish

chmod

chmod(fspath: FsPath, mode: int) -> None

Change the access permissions of a file

Parameters:

• fspath

  (FsPath) –

  Path to file to chmod

• mode

  (int) –

  Permissions mode as an int

copy_file

copy_file(
    src: FsPath,
    dest: FsPath,
    *,
    dryrun: bool = False,
    mkdirp: bool = False,
) -> tuple[str, str]

Copy a file given a source-path and a destination-path

Parameters:

• src

  (str) –

  Source fspath

• dest

  (str) –

  Destination fspath

• dryrun

  (bool, default: False ) –

  Do not copy file if True just return the src and dest

• mkdirp

  (bool, default: False ) –

  Create parent directories if they do not exist

cp

cp(
    src: FsPath,
    dest: FsPath,
    *,
    force: bool = True,
    recursive: bool = False,
    r: bool = False,
    f: bool = True,
) -> None

Copy the directory/file src to the directory/file dest

Parameters:

• src

  (str) –

  Source directory/file to copy

• dest

  (FsPath) –

  Destination directory/file to copy

• force

  (bool, default: True ) –

  Force the copy (like -f flag for cp in shell)

• recursive

  (bool, default: False ) –

  Recursive copy (like -r flag for cp in shell)

• r

  (bool, default: False ) –

  alias for recursive

• f

  (bool, default: True ) –

  alias for force

Raises:

• ValueError –

  If src is a directory and recursive and r are both False

dir_exists

dir_exists(fspath: FsPath) -> bool

Return True if the given path exists; False otherwise; alias for isdir

dir_exists_async async

dir_exists_async(fspath: FsPath) -> bool

Return True if the directory exists; False otherwise

dirpath_gen

dirpath_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = False,
    topdown: bool = True,
    onerror: Callable[[OSError], Any] | None = None,
    followlinks: bool = False,
    check: bool = True,
) -> Iterator[Path]

Yield all dirpaths as pathlib.Path objects beneath a dirpath

dirs_gen

dirs_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = True,
    topdown: bool = True,
    onerror: Callable[[OSError], Any] | None = None,
    followlinks: bool = False,
    check: bool = True,
) -> Iterator[str]

Yield directory-paths beneath a dirpath (defaults to os.getcwd())

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  Directory path to walk down/through.

• abspath

  (bool, default: True ) –

  Yield the absolute path

• onerror

  (Callable[[OSError], Any] | None, default: None ) –

  Function called on OSError

• topdown

  (bool, default: True ) –

  Not applicable

• followlinks

  (bool, default: False ) –

  Follow links

• check

  (bool, default: True ) –

  Check that dir exists

Returns:

• Iterator[str] –

  Generator object that yields directory paths (absolute or relative)

Examples:

>>> tmpdir = 'dirs_gen.doctest'
>>> from os import makedirs; _makedirs(tmpdir, exist_ok=True)
>>> filepath_parts = [
...     ("dir", "file1.txt"),
...     ("dir", "file2.txt"),
...     ("dir", "file3.txt"),
...     ("dir", "dir2", "file1.txt"),
...     ("dir", "dir2", "file2.txt"),
...     ("dir", "dir2", "file3.txt"),
...     ("dir", "dir2a", "file1.txt"),
...     ("dir", "dir2a", "file2.txt"),
...     ("dir", "dir2a", "file3.txt"),
... ]
>>> from shellfish.fs import touch
>>> expected_dirs = []
>>> expected_files = []
>>> for f in filepath_parts:
...     fspath = path.join(*f)
...     fspath = path.join(tmpdir, fspath)
...     dirpath = path.dirname(fspath)
...     expected_files.append(fspath)
...     expected_dirs.append(dirpath)
...     _makedirs(dirpath, exist_ok=True)
...     touch(fspath)
>>> expected_dirs = list(sorted(set(expected_dirs)))
>>> from pprint import pprint
>>> expected_files = [el.replace('\\', '/') for el in expected_files]
>>> pprint(expected_files)
['dirs_gen.doctest/dir/file1.txt',
 'dirs_gen.doctest/dir/file2.txt',
 'dirs_gen.doctest/dir/file3.txt',
 'dirs_gen.doctest/dir/dir2/file1.txt',
 'dirs_gen.doctest/dir/dir2/file2.txt',
 'dirs_gen.doctest/dir/dir2/file3.txt',
 'dirs_gen.doctest/dir/dir2a/file1.txt',
 'dirs_gen.doctest/dir/dir2a/file2.txt',
 'dirs_gen.doctest/dir/dir2a/file3.txt']
>>> expected_dirs = [el.replace('\\', '/') for el in expected_dirs]
>>> pprint(expected_dirs)
['dirs_gen.doctest/dir',
 'dirs_gen.doctest/dir/dir2',
 'dirs_gen.doctest/dir/dir2a']
>>> _files = list(files_gen(tmpdir))
>>> _dirs = list(dirs_gen(tmpdir))
>>> files_n_dirs_list = list(sorted(_files + _dirs))
>>> files_n_dirs_list = [el.replace('\\', '/') for el in files_n_dirs_list]
>>> pprint(files_n_dirs_list)
['dirs_gen.doctest',
 'dirs_gen.doctest/dir',
 'dirs_gen.doctest/dir/dir2',
 'dirs_gen.doctest/dir/dir2/file1.txt',
 'dirs_gen.doctest/dir/dir2/file2.txt',
 'dirs_gen.doctest/dir/dir2/file3.txt',
 'dirs_gen.doctest/dir/dir2a',
 'dirs_gen.doctest/dir/dir2a/file1.txt',
 'dirs_gen.doctest/dir/dir2a/file2.txt',
 'dirs_gen.doctest/dir/dir2a/file3.txt',
 'dirs_gen.doctest/dir/file1.txt',
 'dirs_gen.doctest/dir/file2.txt',
 'dirs_gen.doctest/dir/file3.txt']
>>> expected = sorted(set(expected_files + expected_dirs + [tmpdir]))
>>> expected = [el.replace('\\', '/') for el in expected]
>>> pprint(expected)
['dirs_gen.doctest',
 'dirs_gen.doctest/dir',
 'dirs_gen.doctest/dir/dir2',
 'dirs_gen.doctest/dir/dir2/file1.txt',
 'dirs_gen.doctest/dir/dir2/file2.txt',
 'dirs_gen.doctest/dir/dir2/file3.txt',
 'dirs_gen.doctest/dir/dir2a',
 'dirs_gen.doctest/dir/dir2a/file1.txt',
 'dirs_gen.doctest/dir/dir2a/file2.txt',
 'dirs_gen.doctest/dir/dir2a/file3.txt',
 'dirs_gen.doctest/dir/file1.txt',
 'dirs_gen.doctest/dir/file2.txt',
 'dirs_gen.doctest/dir/file3.txt']
>>> files_n_dirs_list == expected
True
>>> from shutil import rmtree
>>> rmtree(tmpdir)

exists

exists(fspath: FsPath) -> bool

Return True if the given path exists; False otherwise

extension

extension(fspath: str, *, period: bool = False) -> str

Return the extension for a fspath

Examples:

>>> from shellfish.fs import extension
>>> extension("foo.bar")
'bar'
>>> extension("foo.tar.gz")
'tar.gz'
>>> extension("foo.tar.gz", period=True)
'.tar.gz'

file_exists

file_exists(fspath: FsPath) -> bool

Return True if the given path exists; False otherwise; alias for isfile

file_exists_async async

file_exists_async(fspath: FsPath) -> bool

Return True if the file exists; False otherwise

file_lines_gen

file_lines_gen(
    filepath: FsPath, *, keepends: bool = True
) -> Iterable[str]

Yield lines from a given fspath

Parameters:

• filepath

  (FsPath) –

  File to yield lines from

• keepends

  (bool, default: True ) –

  Flag to keep the ends of the file lines

Yields:

• Iterable[str] –

  Lines from the given fspath

Examples:

>>> string = '\n'.join(str(i) for i in range(1, 10))
>>> string
'1\n2\n3\n4\n5\n6\n7\n8\n9'
>>> fspath = "file_lines_gen.doctest.txt"
>>> from shellfish.fs import write_str
>>> write_str(fspath, string)
17
>>> for file_line in file_lines_gen(fspath):
...     file_line
'1\n'
'2\n'
'3\n'
'4\n'
'5\n'
'6\n'
'7\n'
'8\n'
'9'
>>> for file_line in file_lines_gen(fspath, keepends=False):
...     file_line
'1'
'2'
'3'
'4'
'5'
'6'
'7'
'8'
'9'
>>> import os; os.remove(fspath)

filecmp

filecmp(
    left: FsPath,
    right: FsPath,
    *,
    shallow: bool = True,
    blocksize: int = 65536,
) -> bool

Compare 2 files for equality given their filepaths

Parameters:

• left

  (FsPath) –

  Filepath 1

• right

  (FsPath) –

  Filepath 2

• shallow

  (bool, default: True ) –

  Check only size and modification time if True

• blocksize

  (int, default: 65536 ) –

  Chunk size to read files

Returns:

• bool –

  True if files are equal, False otherwise

filepath_gen

filepath_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = False,
    topdown: bool = True,
    onerror: Callable[[OSError], Any] | None = None,
    followlinks: bool = False,
    check: bool = True,
) -> Iterator[Path]

Yield all filepaths as pathlib.Path objects beneath a dirpath

filepath_mtimedelta_sec

filepath_mtimedelta_sec(filepath: FsPath) -> float

Return the seconds since the file(path) was last modified

files_dirs_gen

files_dirs_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = True,
    topdown: bool = True,
    onerror: Callable[[OSError], Any] | None = None,
    followlinks: bool = False,
    check: bool = True,
) -> tuple[Iterator[str], Iterator[str]]

Return a files_gen() and a dirs_gen() in one swell-foop

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  Directory path to walk down/through.

• abspath

  (bool, default: True ) –

  Yield the absolute path

• onerror

  (Callable[[OSError], Any] | None, default: None ) –

  Function called on OSError

• topdown

  (bool, default: True ) –

  Not applicable

• followlinks

  (bool, default: False ) –

  Follow links

• check

  (bool, default: True ) –

  Check if dirpath is a directory

Returns:

• tuple[Iterator[str], Iterator[str]] –

  A tuple of two generators (files_gen(), dirs_gen())

Examples:

>>> tmpdir = 'files_dirs_gen.doctest'
>>> from os import makedirs; _makedirs(tmpdir, exist_ok=True)
>>> filepath_parts = [
...     ("dir", "file1.txt"),
...     ("dir", "file2.txt"),
...     ("dir", "file3.txt"),
...     ("dir", "dir2", "file1.txt"),
...     ("dir", "dir2", "file2.txt"),
...     ("dir", "dir2", "file3.txt"),
...     ("dir", "dir2a", "file1.txt"),
...     ("dir", "dir2a", "file2.txt"),
...     ("dir", "dir2a", "file3.txt"),
... ]
>>> from shellfish.fs import touch
>>> expected_dirs = []
>>> expected_files = []
>>> for f in filepath_parts:
...     fspath = path.join(*f)
...     fspath = path.join(tmpdir, fspath)
...     dirpath = path.dirname(fspath)
...     expected_files.append(fspath)
...     expected_dirs.append(dirpath)
...     _makedirs(dirpath, exist_ok=True)
...     touch(fspath)
>>> expected_dirs = list(sorted(set(expected_dirs)))
>>> from pprint import pprint
>>> expected_files = [el.replace('\\', '/') for el in expected_files]
>>> pprint(expected_files)
['files_dirs_gen.doctest/dir/file1.txt',
 'files_dirs_gen.doctest/dir/file2.txt',
 'files_dirs_gen.doctest/dir/file3.txt',
 'files_dirs_gen.doctest/dir/dir2/file1.txt',
 'files_dirs_gen.doctest/dir/dir2/file2.txt',
 'files_dirs_gen.doctest/dir/dir2/file3.txt',
 'files_dirs_gen.doctest/dir/dir2a/file1.txt',
 'files_dirs_gen.doctest/dir/dir2a/file2.txt',
 'files_dirs_gen.doctest/dir/dir2a/file3.txt']
>>> expected_dirs = [el.replace('\\', '/') for el in expected_dirs]
>>> pprint(expected_dirs)
['files_dirs_gen.doctest/dir',
 'files_dirs_gen.doctest/dir/dir2',
 'files_dirs_gen.doctest/dir/dir2a']
>>> _files, _dirs = files_dirs_gen(tmpdir)
>>> _files = list(_files)
>>> _dirs = list(_dirs)
>>> files_n_dirs_list = list(sorted(set(_files + _dirs)))
>>> files_n_dirs_list = [el.replace('\\', '/') for el in files_n_dirs_list]
>>> pprint(files_n_dirs_list)
['files_dirs_gen.doctest',
 'files_dirs_gen.doctest/dir',
 'files_dirs_gen.doctest/dir/dir2',
 'files_dirs_gen.doctest/dir/dir2/file1.txt',
 'files_dirs_gen.doctest/dir/dir2/file2.txt',
 'files_dirs_gen.doctest/dir/dir2/file3.txt',
 'files_dirs_gen.doctest/dir/dir2a',
 'files_dirs_gen.doctest/dir/dir2a/file1.txt',
 'files_dirs_gen.doctest/dir/dir2a/file2.txt',
 'files_dirs_gen.doctest/dir/dir2a/file3.txt',
 'files_dirs_gen.doctest/dir/file1.txt',
 'files_dirs_gen.doctest/dir/file2.txt',
 'files_dirs_gen.doctest/dir/file3.txt']
>>> expected = sorted(set(expected_files + expected_dirs + [tmpdir]))
>>> expected = [el.replace('\\', '/') for el in expected]
>>> pprint(expected)
['files_dirs_gen.doctest',
 'files_dirs_gen.doctest/dir',
 'files_dirs_gen.doctest/dir/dir2',
 'files_dirs_gen.doctest/dir/dir2/file1.txt',
 'files_dirs_gen.doctest/dir/dir2/file2.txt',
 'files_dirs_gen.doctest/dir/dir2/file3.txt',
 'files_dirs_gen.doctest/dir/dir2a',
 'files_dirs_gen.doctest/dir/dir2a/file1.txt',
 'files_dirs_gen.doctest/dir/dir2a/file2.txt',
 'files_dirs_gen.doctest/dir/dir2a/file3.txt',
 'files_dirs_gen.doctest/dir/file1.txt',
 'files_dirs_gen.doctest/dir/file2.txt',
 'files_dirs_gen.doctest/dir/file3.txt']
>>> files_n_dirs_list == expected
True
>>> from shutil import rmtree
>>> rmtree(tmpdir)

files_gen

files_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = True,
    topdown: bool = True,
    onerror: Callable[[OSError], Any] | None = None,
    followlinks: bool = False,
    check: bool = True,
) -> Iterator[str]

Yield file-paths beneath a given dirpath (defaults to os.getcwd())

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  Directory path to walk down/through.

• abspath

  (bool, default: True ) –

  Yield the absolute path

• onerror

  (Callable[[OSError], Any] | None, default: None ) –

  Function called on OSError

• topdown

  (bool, default: True ) –

  Not applicable

• followlinks

  (bool, default: False ) –

  Follow links

• check

  (bool, default: True ) –

  Check that dir exists

Returns:

• Iterator[str] –

  Generator object that yields file-paths (absolute or relative)

Examples:

>>> tmpdir = 'files_gen.doctest'
>>> from os import makedirs; _makedirs(tmpdir, exist_ok=True)
>>> filepath_parts = [
...     ("dir", "file1.txt"),
...     ("dir", "file2.txt"),
...     ("dir", "file3.txt"),
...     ("dir", "dir2", "file1.txt"),
...     ("dir", "dir2", "file2.txt"),
...     ("dir", "dir2", "file3.txt"),
...     ("dir", "dir2a", "file1.txt"),
...     ("dir", "dir2a", "file2.txt"),
...     ("dir", "dir2a", "file3.txt"),
... ]
>>> from shellfish.fs import touch
>>> expected_files = []
>>> for f in filepath_parts:
...     fspath = path.join(*f)
...     fspath = path.join(tmpdir, fspath)
...     dirpath = path.dirname(fspath)
...     expected_files.append(fspath)
...     _makedirs(dirpath, exist_ok=True)
...     touch(fspath)
>>> from pprint import pprint
>>> expected_files = [el.replace('\\', '/') for el in expected_files]
>>> pprint(expected_files)
['files_gen.doctest/dir/file1.txt',
 'files_gen.doctest/dir/file2.txt',
 'files_gen.doctest/dir/file3.txt',
 'files_gen.doctest/dir/dir2/file1.txt',
 'files_gen.doctest/dir/dir2/file2.txt',
 'files_gen.doctest/dir/dir2/file3.txt',
 'files_gen.doctest/dir/dir2a/file1.txt',
 'files_gen.doctest/dir/dir2a/file2.txt',
 'files_gen.doctest/dir/dir2a/file3.txt']
>>> files_list = list(sorted(set(files_gen(tmpdir))))
>>> files_list = [el.replace('\\', '/') for el in files_list]
>>> pprint(files_list)
['files_gen.doctest/dir/dir2/file1.txt',
 'files_gen.doctest/dir/dir2/file2.txt',
 'files_gen.doctest/dir/dir2/file3.txt',
 'files_gen.doctest/dir/dir2a/file1.txt',
 'files_gen.doctest/dir/dir2a/file2.txt',
 'files_gen.doctest/dir/dir2a/file3.txt',
 'files_gen.doctest/dir/file1.txt',
 'files_gen.doctest/dir/file2.txt',
 'files_gen.doctest/dir/file3.txt']
>>> pprint(list(sorted(set(expected_files))))
['files_gen.doctest/dir/dir2/file1.txt',
 'files_gen.doctest/dir/dir2/file2.txt',
 'files_gen.doctest/dir/dir2/file3.txt',
 'files_gen.doctest/dir/dir2a/file1.txt',
 'files_gen.doctest/dir/dir2a/file2.txt',
 'files_gen.doctest/dir/dir2a/file3.txt',
 'files_gen.doctest/dir/file1.txt',
 'files_gen.doctest/dir/file2.txt',
 'files_gen.doctest/dir/file3.txt']
>>> list(sorted(set(files_list))) == list(sorted(set(expected_files)))
True
>>> from shutil import rmtree
>>> rmtree(tmpdir)

filesize

filesize(fspath: FsPath) -> int

Return the size of the given file(path) in bytes

Parameters:

• fspath

  (FsPath) –

  Filepath as a string or pathlib.Path object

Returns:

• int ( int ) –

  size of the fspath in bytes

filesize_async async

filesize_async(fspath: FsPath) -> int

Return the size of the file at the given fspath

Examples:

>>> from asyncio import run as aiorun
>>> from pathlib import Path
>>> from tempfile import TemporaryDirectory
>>> with TemporaryDirectory() as tmpdir:
...     tmpdir = Path(tmpdir)
...     fpath = tmpdir / "test.txt"
...     written = fpath.write_text("hello world")
...     aiorun(filesize_async(fpath))
11

fspath

fspath(fspath: FsPath) -> str

Alias for os._fspath; returns fspath string for any type of path

glob

glob(
    pattern: str,
    *,
    recursive: bool = False,
    r: bool = False,
) -> Iterator[str]

Return an iterator of fspaths matching the given glob pattern

Parameters:

• pattern

  (str) –

  Glob pattern

• recursive

  (bool, default: False ) –

  Recursively search directories if True

• r

  (bool, default: False ) –

  Recursively search directories if True (Alias for recursive)

Returns:

• Iterator[str] –

  Iterator[str]: Iterator of fspaths matching the glob pattern

is_dir

is_dir(fspath: FsPath) -> bool

Return True if the given path is a directory; alias for isdir

is_dir_async async

is_dir_async(fspath: FsPath) -> bool

Return True if the given path is a file; False otherwise

is_file

is_file(fspath: FsPath) -> bool

Return True if the given path is a file; alias for isfile

is_file_async async

is_file_async(fspath: FsPath) -> bool

Return True if the given path is a file; False otherwise

is_link

is_link(fspath: FsPath) -> bool

Return True if the given path is a link; alias for islink

is_link_async async

is_link_async(fspath: FsPath) -> bool

Return True if the given path is a link; False otherwise

isdir

isdir(fspath: FsPath) -> bool

Return True if the given path is a directory; False otherwise

isdir_async async

isdir_async(fspath: FsPath) -> bool

Return True if the given path is a file; False otherwise

isfile

isfile(fspath: FsPath) -> bool

Return True if the given path is a file; False otherwise

isfile_async async

isfile_async(fspath: FsPath) -> bool

Return True if the given path is a file; False otherwise

islink

islink(fspath: FsPath) -> bool

Return True if the given path is a link; False otherwise

islink_async async

islink_async(fspath: FsPath) -> bool

Return True if the given path is a link; False otherwise

listdir_async async

listdir_async(fspath: FsPath) -> list[str]

Async version of os.listdir

listdir_gen

listdir_gen(
    fspath: FsPath = ".",
    *,
    abspath: bool = False,
    follow_symlinks: bool = True,
    files: bool = True,
    dirs: bool = True,
    symlinks: bool = False,
    files_only: bool = False,
    dirs_only: bool = False,
    symlinks_only: bool = False,
) -> Iterator[Path]

Return an iterator of strings from DirEntries

Examples:

>>> tmpdir = 'listdir_gen.doctest'
>>> from shellfish import sh
>>> from os import makedirs, path, chdir
>>> from shutil import rmtree
>>> _makedirs(tmpdir, exist_ok=True)
>>> pwd = sh.pwd()
>>> sh.cd(tmpdir)
>>> filepath_parts = [
...     ("dir", "file1.txt"),
...     ("dir", "file2.txt"),
...     ("dir", "file3.txt"),
...     ("dir", "data1.json"),
...     ("dir", "dir2", "file1.txt"),
...     ("dir", "dir2", "file2.txt"),
...     ("dir", "dir2", "file3.txt"),
...     ("dir", "dir2a", "file1.txt"),
...     ("dir", "dir2a", "file2.txt"),
...     ("dir", "dir2a", "file3.txt"),
... ]
>>> from shellfish.fs import touch
>>> expected_files = []
>>> for f in filepath_parts:
...     fspath = path.join(*f)
...     fspath = path.join(tmpdir, fspath)
...     dirpath = path.dirname(fspath)
...     expected_files.append(fspath)
...     _makedirs(dirpath, exist_ok=True)
...     touch(fspath)
>>> dirpath = path.join(tmpdir, 'dir')
>>> dirpath.replace("\\", "/")
'listdir_gen.doctest/dir'
>>> sorted(listdir_gen(dirpath, dirs=False, symlinks=False))
['data1.json', 'file1.txt', 'file2.txt', 'file3.txt']
>>> abspaths = sorted(listdir_gen(dirpath, abspath=True, dirs=False, symlinks=False))
>>> for abspath in [p.replace("\\", "/") for p in abspaths]:
...    print(abspath)
listdir_gen.doctest/dir/data1.json
listdir_gen.doctest/dir/file1.txt
listdir_gen.doctest/dir/file2.txt
listdir_gen.doctest/dir/file3.txt
>>> sh.cd(pwd)
>>> import os
>>> if path.exists(tmpdir):
...     rmtree(tmpdir)
>>> path.isdir(tmpdir)
False

lstat_async async

lstat_async(fspath: FsPath) -> stat_result

Async version of os.lstat

mkdir

mkdir(
    fspath: FsPath,
    *,
    parents: bool = False,
    p: bool = False,
    exist_ok: bool = False,
) -> None

Make directory at given fspath

Parameters:

• fspath

  (FsPath) –

  Directory path to create

• parents

  (bool, default: False ) –

  Make parent dirs if True; do not make parent dirs if False

• p

  (bool, default: False ) –

  Make parent dirs if True; do not make parent dirs if False (alias of parents)

• exist_ok

  (bool, default: False ) –

  Throw error if directory exists and exist_ok is False

Returns:

• None –

  None

mkdirp

mkdirp(fspath: FsPath) -> None

Make directory and parents

move

move(src: FsPath, dest: FsPath) -> None

Move file(s) like on the command line

Parameters:

• src

  (FsPath) –

  source file(s)

• dest

  (FsPath) –

  destination path

path_gen

path_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = False,
    topdown: bool = True,
    onerror: Callable[[OSError], Any] | None = None,
    followlinks: bool = False,
    check: bool = True,
) -> Iterator[Path]

Yield all filepaths as pathlib.Path objects beneath a dirpath

read_bytes

read_bytes(filepath: FsPath) -> bytes

Read bytes from a fspath

Parameters:

• filepath

  (FsPath) –

  fspath read as bytes

Returns:

• bytes –

  bytes from the fspath

Examples:

>>> from shellfish.fs import read_bytes, write_bytes
>>> fspath = "rbytes.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> write_bytes(fspath, bites_to_save)
20
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> read_bytes(fspath)
b'These are some bytes'
>>> import os; os.remove(fspath)

read_bytes_async async

read_bytes_async(filepath: FsPath) -> bytes

(ASYNC) Load/Read bytes from a fspath

Parameters:

• filepath

  (FsPath) –

  fspath read as bytes

Returns:

• bytes –

  bytes from the fspath

Examples:

>>> from shellfish.fs._async import read_bytes_async, write_bytes_async
>>> from asyncio import run as aiorun
>>> fspath = "rbytes_async.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> aiorun(write_bytes_async(fspath, bites_to_save))
20
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> aiorun(read_bytes_async(fspath))
b'These are some bytes'
>>> import os; os.remove(fspath)

read_bytes_gen

read_bytes_gen(
    filepath: FsPath, blocksize: int = 65536
) -> Iterable[bytes]

Yield bytes from a given fspath

read_bytes_gen_async async

read_bytes_gen_async(
    filepath: FsPath, blocksize: int = 65536
) -> AsyncIterable[bytes]

Yield (asynchronously) bytes from a given fspath

Parameters:

• filepath

  (FsPath) –

  fspath to read from

• blocksize

  (int, default: 65536 ) –

  size of the block to read

Yields:

• AsyncIterable[bytes] –

  bytes from AsyncIterable[bytes] of the file bytes

Examples:

>>> from os import remove
>>> from asyncio import run
>>> from shellfish.fs._async import write_bytes_gen_async, read_bytes_gen_async
>>> fspath = 'rbytes_gen_async.doctest.txt'
>>> bites_to_save = (b"These are some bytes... ", b"more bytes!")
>>> bites_to_save
(b'These are some bytes... ', b'more bytes!')
>>> run(write_bytes_gen_async(fspath, bites_to_save))
35
>>> async def read():
...     async for b in read_bytes_gen_async(fspath, blocksize=4):
...         print(b)
>>> run(read())
b'Thes'
b'e ar'
b'e so'
b'me b'
b'ytes'
b'... '
b'more'
b' byt'
b'es!'
>>> remove(fspath)
>>> async def async_gen():
...     for b in bites_to_save:
...        yield b
>>> run(write_bytes_gen_async(fspath, bites_to_save))
35
>>> run(read())
b'Thes'
b'e ar'
b'e so'
b'me b'
b'ytes'
b'... '
b'more'
b' byt'
b'es!'
>>> remove(fspath)
>>> class AsyncIterable:
...     def __aiter__(self):
...         return async_gen()
>>> run(write_bytes_gen_async(fspath, AsyncIterable()))
35
>>> run(read())
b'Thes'
b'e ar'
b'e so'
b'me b'
b'ytes'
b'... '
b'more'
b' byt'
b'es!'
>>> remove(fspath)

read_json

read_json(filepath: FsPath) -> Any

Load/Read-&-parse json data given a fspath

Parameters:

• filepath

  (FsPath) –

  Filepath to load/read data from

Returns:

• Any –

  Parsed JSON data

Examples:

Imports:

>>> from shellfish.fs import read_json, write_json

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_dict.doctest.json"
>>> write_json(fspath, data)
19
>>> read_json(fspath)
{'a': 1, 'b': 2, 'c': 3}
>>> import os; os.remove(fspath)

Lists:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> data = list(data.items())
>>> data  # has tuples, but will be saved as strings
[('a', 1), ('b', 2), ('c', 3)]
>>> fspath = "rjson_dict.doctest.json"
>>> write_json(fspath, data)
25
>>> read_json(fspath)
[['a', 1], ['b', 2], ['c', 3]]
>>> os.remove(fspath)

read_json_async async

read_json_async(filepath: FsPath) -> Any

Load/Read-&-parse json data given a fspath

Parameters:

• filepath

  (FsPath) –

  Filepath to load/read data from

Returns:

• Any –

  Parsed JSON data

Examples:

Imports:

>>> from asyncio import run
>>> from shellfish.fs._async import read_json_async, write_json_async

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_async_dict.doctest.json"
>>> run(write_json_async(fspath, data))
19
>>> run(read_json_async(fspath))
{'a': 1, 'b': 2, 'c': 3}
>>> import os; os.remove(fspath)

Lists:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> data = list(data.items())
>>> data  # has tuples, but will be saved as strings
[('a', 1), ('b', 2), ('c', 3)]
>>> fspath = "rjson_async_list.doctest.json"
>>> run(write_json_async(fspath, data))
25
>>> run(read_json_async(fspath))
[['a', 1], ['b', 2], ['c', 3]]

>>> import os; os.remove(fspath)

read_str

read_str(
    filepath: FsPath, *, encoding: str = "utf-8"
) -> str

Load/Read a string given a fspath

Parameters:

• filepath

  (FsPath) –

  Filepath for file to read

• encoding

  (str, default: 'utf-8' ) –

  Encoding to use for reading the file

Returns:

• str ( str ) –

  String read from given fspath

Examples:

>>> from shellfish.fs import read_str, write_str
>>> fspath = "read_str.doctest.txt"
>>> write_str(fspath, r'Check out this string')
21
>>> read_str(fspath)
'Check out this string'
>>> import os; os.remove(fspath)

read_str_async async

read_str_async(
    filepath: FsPath, encoding: str = "utf-8"
) -> str

(ASYNC) Load/Read a string given a fspath

Parameters:

• filepath

  (FsPath) –

  Filepath for file to read

• encoding

  (str, default: 'utf-8' ) –

  File encoding (Default='utf-8')

Returns:

• str ( str ) –

  String read from given fspath

rm_gen

rm_gen(
    fspath: FsPath,
    *,
    force: bool = False,
    recursive: bool = False,
    dryrun: bool = False,
) -> Generator[str, Any, Any]

Remove files & directories in the style of the shell

Parameters:

• fspath

  (FsPath) –

  Path to file or directory to remove

• force

  (bool, default: False ) –

  Force removal of files and directories

• recursive

  (bool, default: False ) –

  Flag to remove recursively (like the -r in rm -r dir)

• dryrun

  (bool, default: False ) –

  Do not remove file if True

Raises:

• ValueError –

  If recursive and r are False and fspath is a directory

rmdir

rmdir(
    fspath: FsPath,
    *,
    force: bool = False,
    recursive: bool = False,
) -> None

Remove directory at given fspath

Parameters:

• fspath

  (FsPath) –

  Directory path to remove

• force

  (bool, default: False ) –

  Force removal of files and directories

• recursive

  (bool, default: False ) –

  Recursively remove all contents if True

Returns:

• None –

  None

rmfile

rmfile(fspath: FsPath, *, dryrun: bool = False) -> str

Remove a file at given fspath

Parameters:

• fspath

  (FsPath) –

  Filepath to remove

• dryrun

  (bool, default: False ) –

  Do not remove file if True

Returns:

• str –

  None

safepath

safepath(fspath: FsPath) -> str

Check if a file/dir path is save/unused; returns an unused path.

Parameters:

• fspath

  (FsPath) –

  file-system path; file or directory path string or Path obj

Returns:

• str ( str ) –

  file/dir path that does not exist and contains the given path

scandir

scandir(
    dirpath: FsPath = ".",
) -> Iterable[DirEntry[AnyStr]]

Typed version of os.scandir

scandir_gen

scandir_gen(
    fspath: FsPath = ".",
    *,
    recursive: bool = False,
    follow_symlinks: bool = True,
    files: bool = True,
    dirs: bool = True,
    symlinks: bool = True,
    files_only: bool = False,
    dirs_only: bool = False,
    symlinks_only: bool = False,
) -> Iterator[DirEntry[str]]

Return an iterator of os.DirEntry objects

Parameters:

• fspath

  (FsPath, default: '.' ) –

  (FsPath): dirpath to look through

• recursive

  (bool, default: False ) –

  recursively scan the directory

• follow_symlinks

  (bool, default: True ) –

  follow symlinks when checking for dirs and files

• files

  (bool, default: True ) –

  include files

• dirs

  (bool, default: True ) –

  include directories

• symlinks

  (bool, default: True ) –

  include symlinks

• dirs_only

  (bool, default: False ) –

  only include directories

• files_only

  (bool, default: False ) –

  only include files

• symlinks_only

  (bool, default: False ) –

  only include symlinks

Returns:

• Iterator[DirEntry[str]] –

  Iterator[DirEntry]: Iterator of os.DirEntry objects

Raises:

• ValueError –

  if any of the kwargs (dirs, files and symlinks) are not True

scandir_list

scandir_list(
    dirpath: FsPath = ".",
) -> list[DirEntry[AnyStr]]

Return a list of os.DirEntry objects

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  Dirpath to scan

Returns:

• list[DirEntry[AnyStr]] –

  list[DirEntry]: List of os.DirEntry objects

sep_join

sep_join(path_strings: Iterator[str]) -> str

Join iterable of strings on the current platform os.path.sep value

sep_lstrip

sep_lstrip(fspath: FsPath) -> str

Left-strip a string of the current platform's os.path.sep value

sep_rstrip

sep_rstrip(fspath: FsPath) -> str

Right-strip a string of the current platform's os.path.sep value

sep_split

sep_split(fspath: FsPath) -> tuple[str, ...]

Split a string on the current platform os.path.sep value

sep_strip

sep_strip(fspath: FsPath) -> str

Strip a string of the current platform's os.path.sep value

shebang

shebang(fspath: FsPath) -> str | None

Get the shebang string given a fspath; Returns None if no shebang

Parameters:

• fspath

  (FsPath) –

  Path to file that might have a shebang

Returns:

• str | None –

  Optional[str]: The shebang string if it exists, None otherwise

Examples:

>>> from inspect import getabsfile
>>> script = 'ashellscript.sh'
>>> with open(script, 'w') as f:
...     f.write('#!/bin/bash\necho "howdy"\n')
25
>>> shebang(script)
'#!/bin/bash'
>>> from os import remove
>>> remove(script)

stat

stat(fspath: FsPath) -> stat_result

Return the os.stat_result object for a given fspath

Parameters:

• fspath

  (FsPath) –

  Path to file or directory

Returns:

• stat_result –

  os.stat_result: stat_result object

stat_async async

stat_async(fspath: FsPath) -> stat_result

Async version of os.stat

touch

touch(fspath: FsPath, *, mkdirp: bool = True) -> None

Create an empty file given a fspath

Parameters:

• fspath

  (FsPath) –

  File-system path for where to make an empty file

• mkdirp

  (bool, default: True ) –

  Make parent directories if they don't exist

walk_gen

walk_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = True,
    topdown: bool = True,
    onerror: Callable[[OSError], Any] | None = None,
    followlinks: bool = False,
    check: bool = True,
) -> Iterator[str]

Yield all paths beneath a given dirpath (defaults to os.getcwd())

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  Directory path to walk down/through.

• abspath

  (bool, default: True ) –

  Yield the absolute path

• onerror

  (Callable[[OSError], Any] | None, default: None ) –

  Function called on OSError

• topdown

  (bool, default: True ) –

  Not applicable

• followlinks

  (bool, default: False ) –

  Follow links

• check

  (bool, default: True ) –

  Check if dirpath exists

Returns:

• Iterator[str] –

  Generator object that yields directory paths (absolute or relative)

Examples:

>>> tmpdir = 'walk_gen.doctest'
>>> from os import makedirs; _makedirs(tmpdir, exist_ok=True)
>>> filepath_parts = [
...     ("dir", "file1.txt"),
...     ("dir", "file2.txt"),
...     ("dir", "file3.txt"),
...     ("dir", "dir2", "file1.txt"),
...     ("dir", "dir2", "file2.txt"),
...     ("dir", "dir2", "file3.txt"),
...     ("dir", "dir2a", "file1.txt"),
...     ("dir", "dir2a", "file2.txt"),
...     ("dir", "dir2a", "file3.txt"),
... ]
>>> from shellfish.fs import touch
>>> expected_dirs = []
>>> expected_files = []
>>> for f in filepath_parts:
...     fspath = path.join(*f).replace('\\', '/')
...     fspath = path.join(tmpdir, fspath).replace('\\', '/')
...     dirpath = path.dirname(fspath)
...     expected_files.append(fspath)
...     expected_dirs.append(dirpath)
...     _makedirs(dirpath, exist_ok=True)
...     touch(fspath)
>>> expected_dirs = [el.replace('\\', '/') for el in sorted(set(expected_dirs))]
>>> from pprint import pprint
>>> pprint(expected_files)
['walk_gen.doctest/dir/file1.txt',
 'walk_gen.doctest/dir/file2.txt',
 'walk_gen.doctest/dir/file3.txt',
 'walk_gen.doctest/dir/dir2/file1.txt',
 'walk_gen.doctest/dir/dir2/file2.txt',
 'walk_gen.doctest/dir/dir2/file3.txt',
 'walk_gen.doctest/dir/dir2a/file1.txt',
 'walk_gen.doctest/dir/dir2a/file2.txt',
 'walk_gen.doctest/dir/dir2a/file3.txt']
>>> pprint(expected_dirs)
['walk_gen.doctest/dir',
 'walk_gen.doctest/dir/dir2',
 'walk_gen.doctest/dir/dir2a']
>>> walk_gen_list = list(sorted(walk_gen(tmpdir)))
>>> walk_gen_list = [el.replace('\\', '/') for el in walk_gen_list]
>>> pprint(walk_gen_list)
['walk_gen.doctest',
 'walk_gen.doctest/dir',
 'walk_gen.doctest/dir/dir2',
 'walk_gen.doctest/dir/dir2/file1.txt',
 'walk_gen.doctest/dir/dir2/file2.txt',
 'walk_gen.doctest/dir/dir2/file3.txt',
 'walk_gen.doctest/dir/dir2a',
 'walk_gen.doctest/dir/dir2a/file1.txt',
 'walk_gen.doctest/dir/dir2a/file2.txt',
 'walk_gen.doctest/dir/dir2a/file3.txt',
 'walk_gen.doctest/dir/file1.txt',
 'walk_gen.doctest/dir/file2.txt',
 'walk_gen.doctest/dir/file3.txt']
>>> expected = sorted(set(expected_files + expected_dirs + [tmpdir]))
>>> pprint(expected)
['walk_gen.doctest',
 'walk_gen.doctest/dir',
 'walk_gen.doctest/dir/dir2',
 'walk_gen.doctest/dir/dir2/file1.txt',
 'walk_gen.doctest/dir/dir2/file2.txt',
 'walk_gen.doctest/dir/dir2/file3.txt',
 'walk_gen.doctest/dir/dir2a',
 'walk_gen.doctest/dir/dir2a/file1.txt',
 'walk_gen.doctest/dir/dir2a/file2.txt',
 'walk_gen.doctest/dir/dir2a/file3.txt',
 'walk_gen.doctest/dir/file1.txt',
 'walk_gen.doctest/dir/file2.txt',
 'walk_gen.doctest/dir/file3.txt']
>>> walk_gen_list == expected
True
>>> from shutil import rmtree
>>> rmtree(tmpdir)

write_bytes

write_bytes(
    filepath: FsPath,
    bites: bytes,
    *,
    append: bool = False,
    chmod: int | None = None,
) -> int

Write/Save bytes to a fspath

The parameter 'bites' is used instead of 'bytes' to not redefine the built-in python bytes object.

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• bites

  (bytes) –

  Bytes to be written

• append

  (bool, default: False ) –

  Append to the file if True, overwrite otherwise; default is False

• chmod

  (Optional[int], default: None ) –

  chmod the file after writing; default is None

Returns:

• int ( int ) –

  Number of bytes written

Examples:

>>> from shellfish.fs import read_bytes, write_bytes
>>> fspath = "wbytes.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> write_bytes(fspath, bites_to_save)
20
>>> read_bytes(fspath)
b'These are some bytes'
>>> import os; os.remove(fspath)

write_bytes_async async

write_bytes_async(
    filepath: FsPath,
    bites: bytes,
    *,
    append: bool = False,
    chmod: int | None = None,
) -> int

(ASYNC) Write/Save bytes to a fspath

The parameter 'bites' is used instead of 'bytes' so as to not redefine the built-in python bytes object.

Parameters:

• append

  (bool, default: False ) –

  Append to the fspath if True; otherwise overwrite

• filepath

  (FsPath) –

  fspath to write to

• bites

  (bytes) –

  Bytes to be written

• chmod

  (int | None, default: None ) –

  chmod the fspath to this mode after writing

Returns:

• int –

  None

Examples:

>>> from shellfish.fs._async import read_bytes_async, write_bytes_async
>>> from asyncio import run as aiorun
>>> fspath = "wbytes_async.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> aiorun(write_bytes_async(fspath, bites_to_save))
20
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> aiorun(read_bytes_async(fspath))
b'These are some bytes'
>>> import os; os.remove(fspath)

write_bytes_gen

write_bytes_gen(
    filepath: FsPath,
    bytes_gen: Iterable[bytes],
    *,
    append: bool = False,
    chmod: int | None = None,
) -> int

Write/Save bytes to a fspath

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• bytes_gen

  (Iterable[bytes]) –

  Bytes to be written

• append

  (bool, default: False ) –

  Append to the file if True, overwrite otherwise; default is False

• chmod

  (Optional[int], default: None ) –

  chmod the file after writing; default is None

Returns:

• int ( int ) –

  Number of bytes written

Examples:

>>> from shellfish.fs import read_bytes, write_bytes
>>> fspath = "wbytes_gen.doctest.txt"
>>> bites_to_save = (b"These are some bytes... ", b"more bytes!")
>>> bites_to_save  # they are bytes!
(b'These are some bytes... ', b'more bytes!')
>>> write_bytes_gen(fspath, (b for b in bites_to_save))
35
>>> read_bytes(fspath)
b'These are some bytes... more bytes!'
>>> import os; os.remove(fspath)

write_bytes_gen_async async

write_bytes_gen_async(
    filepath: FsPath,
    bytes_gen: Iterable[bytes] | AsyncIterable[bytes],
    *,
    append: bool = False,
    chmod: int | None = None,
) -> int

Write/save bytes to a filepath from an (async)iterable/iterator of bytes

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• bytes_gen

  (Iterable[bytes] | AsyncIterable[bytes]) –

  AsyncIterable/Iterator of bytes to write

• append

  (bool, default: False ) –

  Append to the fspath if True; otherwise overwrite

• chmod

  (int | None, default: None ) –

  chmod the fspath if not None

Returns:

• int ( int ) –

  number of bytes written

Examples:

>>> from os import remove
>>> from asyncio import run
>>> from shellfish.fs._async import write_bytes_gen_async, read_bytes_gen_async
>>> fspath = 'wbytes_gen_async.doctest.txt'
>>> bites_to_save = (b"These are some bytes... ", b"more bytes!")
>>> bites_to_save
(b'These are some bytes... ', b'more bytes!')
>>> run(write_bytes_gen_async(fspath, bites_to_save))
35
>>> async def read():
...     async for b in read_bytes_gen_async(fspath, blocksize=4):
...         print(b)
>>> run(read())
b'Thes'
b'e ar'
b'e so'
b'me b'
b'ytes'
b'... '
b'more'
b' byt'
b'es!'
>>> remove(fspath)
>>> async def async_gen():
...     for b in bites_to_save:
...        yield b
>>> run(write_bytes_gen_async(fspath, bites_to_save))
35
>>> run(read())
b'Thes'
b'e ar'
b'e so'
b'me b'
b'ytes'
b'... '
b'more'
b' byt'
b'es!'
>>> remove(fspath)
>>> class AsyncIterable:
...     def __aiter__(self):
...         return async_gen()
>>> run(write_bytes_gen_async(fspath, AsyncIterable()))
35
>>> run(read())
b'Thes'
b'e ar'
b'e so'
b'me b'
b'ytes'
b'... '
b'more'
b' byt'
b'es!'
>>> remove(fspath)

write_json

write_json(
    filepath: FsPath,
    data: Any,
    *,
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Callable[[Any], Any] | None = None,
    chmod: int | None = None,
    append: bool = False,
    **kwargs: Any,
) -> int

Save/Write json-serial-ize-able data to a fspath

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• data

  (Any) –

  json-serial-ize-able data

• fmt

  (bool, default: False ) –

  Indented (2 spaces) or minify data (default=False)

• pretty

  (bool, default: False ) –

  Indented (2 spaces) or minify data (default=False)

• sort_keys

  (bool, default: False ) –

  Sort the data keys if the data is a dictionary.

• append_newline

  (bool, default: False ) –

  Append a newline to the end of the file

• default

  (Callable[[Any], Any] | None, default: None ) –

  default function hook

• chmod

  (Optional[int], default: None ) –

  Optional chmod to set on file

• append

  (bool, default: False ) –

  Append to the file if True, overwrite otherwise; default

• **kwargs

  (Any, default: {} ) –

  Additional keyword arguments to pass to jsonbourne.JSON.dumpb

Returns:

• int ( int ) –

  Number of bytes written

Examples:

Imports:

>>> from shellfish.fs import read_json, write_json

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_dict.doctest.json"
>>> write_json(fspath, data)
19
>>> read_json(fspath)
{'a': 1, 'b': 2, 'c': 3}
>>> import os; os.remove(fspath)

Lists:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> data = list(data.items())
>>> data  # has tuples, but will be saved as strings
[('a', 1), ('b', 2), ('c', 3)]
>>> fspath = "rjson_dict.doctest.json"
>>> write_json(fspath, data)
25
>>> read_json(fspath)
[['a', 1], ['b', 2], ['c', 3]]
>>> os.remove(fspath)

write_json_async async

write_json_async(
    filepath: FsPath,
    data: Any,
    *,
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Callable[[Any], Any] | None = None,
    append: bool = False,
    chmod: int | None = None,
    **kwargs: Any,
) -> int

Save/Write json-serial-ize-able data to a fspath

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• data

  (Any) –

  json-serial-ize-able data

• fmt

  (bool, default: False ) –

  Indented (2 spaces) or minify data (default=False)

• pretty

  (bool, default: False ) –

  Indented (2 spaces) or minify data (default=False)

• sort_keys

  (bool, default: False ) –

  Sort the data keys if the data is a dictionary.

• append_newline

  (bool, default: False ) –

  Sort the data keys if the data is a dictionary.

• default

  (Callable[[Any], Any] | None, default: None ) –

  default function hook

• append

  (bool, default: False ) –

  Append to the fspath if True; default is False

• chmod

  (Optional[int], default: None ) –

  chmod the fspath if not None

• **kwargs

  (Any, default: {} ) –

  Additional keyword arguments to pass to jsonbourne.JSON.dump

Returns:

• int ( int ) –

  Number of bytes written

Examples:

Imports:

>>> from asyncio import run
>>> from shellfish.fs._async import read_json_async, write_json_async

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "wjson_async_dict.doctest.json"
>>> run(write_json_async(fspath, data))
19
>>> run(read_json_async(fspath))
{'a': 1, 'b': 2, 'c': 3}
>>> import os; os.remove(fspath)

Lists:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> data = list(data.items())
>>> data  # has tuples, but will be saved as strings
[('a', 1), ('b', 2), ('c', 3)]
>>> fspath = "wjson_async_list.doctest.json"
>>> run(write_json_async(fspath, data))
25
>>> run(read_json_async(fspath))
[['a', 1], ['b', 2], ['c', 3]]

>>> import os; os.remove(fspath)

write_str

write_str(
    filepath: FsPath,
    string: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    chmod: int | None = None,
) -> int

Save/Write a string to fspath

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• string

  (str) –

  string to be written

• encoding

  (str, default: 'utf-8' ) –

  String encoding to write file with

• append

  (bool, default: False ) –

  Flag to append to file; default = False

• chmod

  (Optional[int], default: None ) –

  Optional chmod to set on file

Returns:

• int –

  None

Examples:

>>> from shellfish.fs import read_str, write_str
>>> fspath = "sstring.doctest.txt"
>>> write_str(fspath, r'Check out this string')
21
>>> read_str(fspath)
'Check out this string'
>>> import os; os.remove(fspath)

write_str_async async

write_str_async(
    filepath: FsPath,
    string: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    chmod: int | None = None,
) -> int

(ASYNC) Save/Write a string to fspath

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• string

  (str) –

  string to be written

• encoding

  (str, default: 'utf-8' ) –

  File encoding (Default='utf-8')

• append

  (bool, default: False ) –

  Append to the fspath if True; default is False

• chmod

  (Optional[int], default: None ) –

  chmod the fspath if not None

Returns:

• int ( int ) –

  number of bytes written

mkenv

mkenv(
    env: dict[str, str], *, extenv: bool = True
) -> dict[str, str]

Return the environment dict to run a subprocess with

Parameters:

• env

  (dict[str, str]) –

  Environment variables for the subprocess

• extenv

  (bool, default: True ) –

  Extend os.environ with env instead of replacing it

Returns:

• dict[str, str] –

  The environment variables dict

decode_stdio_bytes

decode_stdio_bytes(
    stdio_bytes: str | bytes, *, lf: bool = True
) -> str

Return Stdio bytes from stdout/stderr as a string

Parameters:

• stdio_bytes

  (bytes) –

  STDOUT/STDERR bytes

• lf

  (bool, default: True ) –

  Replace \r\n line endings with \n

Returns:

• str ( str ) –

  decoded stdio bytes

pstdout

pstdout(proc: CompletedProcess[AnyStr]) -> str

Get the STDOUT as a string from a subprocess

Parameters:

• proc

  (CompletedProcess[AnyStr]) –

  python subprocess.process object with stdout

Returns:

• str –

  STDOUT for the proc as string

pstderr

pstderr(proc: CompletedProcess[AnyStr]) -> str

Get the STDERR as a string from a subprocess

Parameters:

• proc

  (CompletedProcess[AnyStr]) –

  python subprocess.process object with STDERR

Returns:

• str –

  STDERR for the proc as string

pstdout_pstderr

pstdout_pstderr(
    proc: CompletedProcess[AnyStr],
) -> tuple[str, str]

Get the STDOUT and STDERR as strings from a subprocess

Parameters:

• proc

  (CompletedProcess[AnyStr]) –

  Completed-subprocess

Returns:

• tuple[str, str] –

  Tuple of two strings: (stdout-string, stderr-string)

flatten_args

flatten_args(*args: Any | list[Any]) -> list[str]

Flatten possibly nested iterables of sequences to a list of strings

Examples:

>>> list(flatten_args("cmd", ["uno", "dos", "tres"]))
['cmd', 'uno', 'dos', 'tres']
>>> list(flatten_args("cmd", ["uno", "dos", "tres", ["4444", "five"]]))
['cmd', 'uno', 'dos', 'tres', '4444', 'five']

do

do(
    *popenargs: PopenArgs,
    args: PopenArgs | None = None,
    env: dict[str, str] | None = None,
    extenv: bool = True,
    cwd: FsPath | None = None,
    shell: bool = False,
    check: bool = False,
    tee: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    timeout: float | int | None = None,
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
    dryrun: bool = False,
) -> Done

Run a subprocess synchronously

Parameters:

• *popenargs

  (PopenArgs, default: () ) –

  Args given as *args; Cannot use both *popenargs and args

• args

  (PopenArgs | None, default: None ) –

  Args as strings for the subprocess

• env

  (dict[str, str] | None, default: None ) –

  Environment variables as a dictionary (Default value = None)

• extenv

  (bool, default: True ) –

  Extend the environment with the current environment (Default value = True)

• cwd

  (FsPath | None, default: None ) –

  Current working directory (Default value = None)

• shell

  (bool, default: False ) –

  Run in shell or sub-shell

• check

  (bool, default: False ) –

  Check the outputs (generally useless)

• input

  (STDIN, default: None ) –

  Stdin to give to the subprocess

• tee

  (bool, default: False ) –

  Flag to tee the subprocess stdout and stderr to sys.stdout/stderr

• verbose

  (bool, default: False ) –

  Flag to write the subprocess stdout and stderr to sys.stdout and sys.stderr

• timeout

  (float | int | None, default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (int | list[int] | tuple[int, ...] | set[int], default: 0 ) –

  Return code(s) to check against

• dryrun

  (bool, default: False ) –

  Don't run the subprocess

Returns:

• Done –

  Finished PRun object which is a dictionary, so a dictionary

Raises:

• ValueError –

  if args and *popenargs are both given

shell

shell(
    *popenargs: PopenArgs,
    args: PopenArgs | None = None,
    env: dict[str, str] | None = None,
    shell: bool = True,
    extenv: bool = True,
    cwd: FsPath | None = None,
    check: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    timeout: float | int | None = None,
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
    dryrun: bool = False,
) -> Done

Run a subprocess synchronously in current shell

Parameters:

• *popenargs

  (PopenArgs, default: () ) –

  Args given as *args; Cannot use both *popenargs and args

• args

  (PopenArgs | None, default: None ) –

  Args as strings for the subprocess

• env

  (dict[str, str] | None, default: None ) –

  Environment variables as a dictionary (Default value = None)

• shell

  (bool, default: True ) –

  Run in shell or sub-shell; default is True for shell

• extenv

  (bool, default: True ) –

  Extend the environment with the current environment (Default value = True)

• cwd

  (FsPath | None, default: None ) –

  Current working directory (Default value = None)

• check

  (bool, default: False ) –

  Check the outputs (generally useless)

• input

  (STDIN, default: None ) –

  Stdin to give to the subprocess

• verbose

  (bool, default: False ) –

  Flag to write the subprocess stdout and stderr to sys.stdout and sys.stderr

• timeout

  (float | int | None, default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (int | list[int] | tuple[int, ...] | set[int], default: 0 ) –

  Return code(s) to check if ok

• dryrun

  (bool, default: False ) –

  Don't run the subprocess

Returns:

• Done –

  Finished PRun object which is a dictionary, so a dictionary

do_asyncify async

do_asyncify(
    *popenargs: PopenArgs,
    args: PopenArgs | None = None,
    env: dict[str, str] | None = None,
    extenv: bool = True,
    cwd: str | None = None,
    shell: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    check: bool = False,
    timeout: float | int | None = None,
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
    dryrun: bool = False,
) -> Done

Run a subprocess asynchronously using asyncified version of do

do_async async

do_async(
    *popenargs: PopenArgs,
    args: PopenArgs | None = None,
    env: dict[str, str] | None = None,
    extenv: bool = True,
    cwd: str | None = None,
    shell: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    check: bool = False,
    timeout: float | None = None,
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
    dryrun: bool = False,
) -> Done

Run a subprocess and await its completion

Parameters:

• *popenargs

  (PopenArgs, default: () ) –

  Args given as *args; Cannot use both *popenargs and args

• args

  (PopenArgs | None, default: None ) –

  Args as strings for the subprocess

• check

  (bool, default: False ) –

  Check the result returncode

• env

  (dict[str, str] | None, default: None ) –

  Environment variables as a dictionary (Default value = None)

• extenv

  (bool, default: True ) –

  Extend environment with the current environment (Default value = True)

• cwd

  (str | None, default: None ) –

  Current working directory (Default value = None)

• shell

  (bool, default: False ) –

  Run in shell or sub-shell

• input

  (STDIN, default: None ) –

  Stdin to give to the subprocess

• verbose

  (bool, default: False ) –

  Flag to write the subprocess stdout and stderr to sys.stdout and sys.stderr

• timeout

  (float | None, default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (int | list[int] | tuple[int, ...] | set[int], default: 0 ) –

  Return code(s) that are considered OK (Default value = 0)

• dryrun

  (bool, default: False ) –

  Flag to not run the subprocess but return a Done object

Returns:

• Done –

  Finished PRun object which is a dictionary, so a dictionary

Raises:

• ValueError –

  If both *popenargs and args are given

doa async

doa(
    *popenargs: PopenArgs,
    args: PopenArgs | None = None,
    env: dict[str, str] | None = None,
    extenv: bool = True,
    cwd: str | None = None,
    shell: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    check: bool = False,
    timeout: float | None = None,
    ok_code: int
    | list[int]
    | tuple[int, ...]
    | set[int] = 0,
    dryrun: bool = False,
) -> Done

Run a subprocess and await its completion

Alias for sh.do_async

Parameters:

• *popenargs

  (PopenArgs, default: () ) –

  Args given as *args; Cannot use both *popenargs and args

• args

  (PopenArgs | None, default: None ) –

  Args as strings for the subprocess

• check

  (bool, default: False ) –

  Check the result returncode

• env

  (dict[str, str] | None, default: None ) –

  Environment variables as a dictionary (Default value = None)

• cwd

  (str | None, default: None ) –

  Current working directory (Default value = None)

• shell

  (bool, default: False ) –

  Run in shell or sub-shell

• input

  (STDIN, default: None ) –

  Stdin to give to the subprocess

• verbose

  (bool, default: False ) –

  Flag to write the subprocess stdout and stderr to sys.stdout and sys.stderr

• timeout

  (float | None, default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (int | list[int] | tuple[int, ...] | set[int], default: 0 ) –

  Return code(s) that are considered OK (Default value = 0)

• dryrun

  (bool, default: False ) –

  Flag to not run the subprocess but return a Done object

• extenv

  (bool, default: True ) –

  Extend environment with the current environment (Default value = True)

Returns:

• Done –

  Finished PRun object which is a dictionary, so a dictionary

pwd

pwd() -> str

Return present-working-directory path string; alias for os.getcwd

Returns:

• str ( str ) –

  present working directory as string

Examples:

>>> import os
>>> pwd() == os.getcwd()
True

dirname

dirname(fspath: FsPath) -> str

Return dirname/parent-dir of given path; alias of os.path.dirname

Parameters:

• fspath

  (FsPath) –

  File-system path

Returns:

• str ( str ) –

  basename of path

basename

basename(fspath: FsPath) -> str

Return the basename of given path; alias of os.path.dirname

Parameters:

• fspath

  (FsPath) –

  File-system path

Returns:

• str ( str ) –

  basename of path

cd

cd(dirpath: FsPath) -> None

Change directory to given dirpath; alias for os.chdir

Parameters:

• dirpath

  (FsPath) –

  Directory fspath

export

export(key: str, val: str | None = None) -> tuple[str, str]

Export/Set an environment variable

Parameters:

• key

  (str) –

  environment variable name/key

• val

  (str, default: None ) –

  environment variable value

Raises:

• ValueError –

  if unable to parse key/val

setenv

setenv(key: str, val: str | None = None) -> tuple[str, str]

Export/Set an environment variable

Parameters:

• key

  (str) –

  environment variable name/key

• val

  (str, default: None ) –

  environment variable value

Returns:

• tuple[str, str] –

  tuple[str, str]: environment variable key/value pair

shplit

shplit(
    string: str,
    *,
    comments: bool = False,
    posix: bool = True,
) -> list[str]

Typed alias for shlex.split

quote

quote(string: str) -> str

Typed alias for shlex.quote

Parameters:

• string

  (str) –

  string to quote

Returns:

• str ( str ) –

  quoted string

Examples:

>>> quote("hello world")
"'hello world'"
>>> quote("hello 'world'")
'\'hello \'"\'"\'world\'"\'"\'\''

q

q(string: str) -> str

Typed alias for shlex.quote

Parameters:

• string

  (str) –

  string to quote

Returns:

• str ( str ) –

  quoted string

Examples:

>>> q("hello world")
"'hello world'"
>>> q("hello 'world'")
'\'hello \'"\'"\'world\'"\'"\'\''

which

which(cmd: str, path: str | None = None) -> str | None

Return the result of shutil.which

Parameters:

• cmd

  (str) –

  Command/exe to find path of

• path

  (str, default: None ) –

  System path to use

Returns:

• str | None –

  Path to the command/exe, or None if not found

where

where(cmd: str, path: str | None = None) -> str | None

Return the result of shutil.which; alias of shellfish.sh.which

Parameters:

• cmd

  (str) –

  Command/exe to find path of

• path

  (str, default: None ) –

  System path to use

Returns:

• str | None –

  Path to the command/exe, or None if not found

which_lru cached

which_lru(cmd: str, path: str | None = None) -> str | None

Return the result of shutil.which and cache the results

Parameters:

• cmd

  (str) –

  Command/exe to find path of

• path

  (str, default: None ) –

  System path to use

Returns:

• str | None –

  Path to the command/exe, or None if not found

tree

tree(
    dirpath: FsPath,
    filterfn: Callable[[str], bool] | None = None,
) -> str

Create a directory tree string given a directory path

Parameters:

• dirpath

  (FsPath) –

  Directory string to make tree for

• filterfn

  (Callable[[str], bool] | None, default: None ) –

  Function to filter sub-directories and sub-files with

Returns:

• str ( str ) –

  Directory-tree string

Examples:

>>> tmpdir = 'tree.doctest'
>>> from os import makedirs; makedirs(tmpdir, exist_ok=True)
>>> from pathlib import Path
>>> filepath_parts = [
...     ("dir", "file1.txt"),
...     ("dir", "file2.txt"),
...     ("dir", "file3.txt"),
...     ("dir", "dir2", "file1.txt"),
...     ("dir", "dir2", "file2.txt"),
...     ("dir", "dir2", "file3.txt"),
...     ("dir", "dir2a", "file1.txt"),
...     ("dir", "dir2a", "file2.txt"),
...     ("dir", "dir2a", "file3.txt"),
... ]
>>> expected_files = []
>>> for f in filepath_parts:
...     fspath = path.join(*f)
...     fspath = path.join(tmpdir, fspath)
...     dirpath = path.dirname(fspath)
...     expected_files.append(fspath)
...     makedirs(dirpath, exist_ok=True)
...     Path(fspath).touch()
>>> print(tree(tmpdir))
tree.doctest/
└── dir/
    ├── dir2/
    │   ├── file1.txt
    │   ├── file2.txt
    │   └── file3.txt
    ├── dir2a/
    │   ├── file1.txt
    │   ├── file2.txt
    │   └── file3.txt
    ├── file1.txt
    ├── file2.txt
    └── file3.txt
>>> print(tree(tmpdir, lambda s: _DirTree._default_filter(s) and not "file2" in s))
tree.doctest/
└── dir/
    ├── dir2/
    │   ├── file1.txt
    │   └── file3.txt
    ├── dir2a/
    │   ├── file1.txt
    │   └── file3.txt
    ├── file1.txt
    └── file3.txt
>>> from shutil import rmtree
>>> rmtree(tmpdir)

ls

ls(
    dirpath: FsPath = ".", *, abspath: bool = False
) -> list[str]

List files and dirs given a dirpath (defaults to pwd)

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  path-string to directory to list

• abspath

  (bool, default: False ) –

  Give absolute paths

Returns:

• list[str] –

  List of the directory items

ls_files

ls_files(
    dirpath: FsPath = ".", *, abspath: bool = False
) -> list[str]

List the files in a given directory path

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  Directory path for which one might want to list files

• abspath

  (bool, default: False ) –

  Return absolute filepaths

Returns:

• list[str] –

  List of files as strings

ls_dirs

ls_dirs(
    dirpath: FsPath = ".", *, abspath: bool = False
) -> list[str]

List the directories in a given directory path

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  Directory path for which one might want list directories

• abspath

  (bool, default: False ) –

  Return absolute directory paths

Returns:

• list[str] –

  List of directories as strings

ls_files_dirs

ls_files_dirs(
    dirpath: FsPath = ".", *, abspath: bool = False
) -> tuple[list[str], list[str]]

List the files and directories given directory path

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  Directory path to execute on

• abspath

  (bool, default: False ) –

  Return absolute file/directory paths

Returns:

• tuple[list[str], list[str]] –

  Two lists of strings; the first is a list of the files and the second is a list of the directories

ls_async async

ls_async(
    dirpath: FsPath = ".", *, abspath: bool = False
) -> list[str]

List files and dirs given a dirpath (defaults to pwd)

Parameters:

• dirpath

  (FsPath, default: '.' ) –

  path-string to directory to list

• abspath

  (bool, default: False ) –

  Give absolute paths

Returns:

• list[str] –

  List of the directory items

rm

rm(
    fspath: FsPath,
    *,
    force: bool = False,
    recursive: bool = False,
    verbose: bool = False,
    f: bool = False,
    r: bool = False,
    v: bool = False,
    dryrun: bool = False,
) -> None

Remove files & directories in the style of the shell

Parameters:

• fspath

  (FsPath) –

  Path to file or directory to remove

• force

  (bool, default: False ) –

  Flag to force removal; ignore missing

• recursive

  (bool, default: False ) –

  Flag to remove recursively (like the -r in rm -r dir)

• verbose

  (bool, default: False ) –

  Flag to be verbose

• f

  (bool, default: False ) –

  alias for force kwarg

• v

  (bool, default: False ) –

  alias for verbose

• r

  (bool, default: False ) –

  alias for recursive kwarg

• dryrun

  (bool, default: False ) –

  Flag to not actually remove anything

Raises:

• ValueError –

  If recursive and r are False and fspath is a directory

mv

mv(src: FsPath, dest: FsPath) -> None

Move file(s) like on the command line

Parameters:

• src

  (FsPath) –

  source file(s)

• dest

  (FsPath) –

  destination path

source

source(filepath: FsPath, *, _globals: bool = True) -> None

Execute/run a python file given a fspath and put globals in globasl

Parameters:

• filepath

  (FsPath) –

  Path to python file

• _globals

  (bool, default: True ) –

  Exec using globals