is_dir

is_dir(fspath: FsPath) -> bool

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

shellfish

shellfish ~ shell and file-system utils

Modules:

• aios –

  aios = asyncio + os

• aioshutil –

  aios = asyncio + shutil

• batman –

  batman = bat/cmd windows utils

• const –

  Constants

• dev –

  UNDER CONSTRUCTION

• dotenv –

  dot.env utils

• echo –

  Echo/Print

• exe –

  Exes/commands

• fs –

  file-system utils

• libhash –
• libsh –

  shellfish internals

• osfs –

  Os specific filesystem utils/operations

• process –

  Current running process info

• psu –

  psutils-utils

• sh –

  shell utils

• sp –
• stdio –

  stdio utils

• testing –
• tests –

  shellfish dist tests

Classes:

• Done –

  PRun => 'ProcessRun' for finished processes

• DoneError –

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

• DoneObj –

  Todo

  deprecate this in favor of DoneDict

• Flag –

  Flag obj

• FlagMeta –

  Meta class

• HrTime –

  High resolution time

• LIN –

  Linux (and Mac) shell commands/methods container

• Stdio –

  Standard-io enum object

• WIN –

  Windows shell commands/methods container

Functions:

• basename –

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

• cd –

  Change directory to given dirpath; alias for os.chdir

• 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

• decode_stdio_bytes –

  Return Stdio bytes from stdout/stderr as a string

• 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

• dirname –

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

• dirpath_gen –

  Yield all dirpaths as pathlib.Path objects beneath a dirpath

• dirs_gen –

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

• do –

  Run a subprocess synchronously

• do_async –

  Run a subprocess and await its completion

• doa –

  Run a subprocess and await its completion

• exists –

  Return True if the given path exists; False otherwise

• export –

  Export/Set an environment variable

• 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

• flatten_args –

  Flatten possibly nested iterables of sequences to a list of strings

• fspath –

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

• 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

• ls –

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

• ls_dirs –

  List the directories in a given directory path

• ls_files –

  List the files in a given directory path

• ls_files_dirs –

  List the files and directories given directory path

• 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

• mv –

  Move file(s) like on the command line

• path_gen –

  Yield all filepaths as pathlib.Path objects beneath a dirpath

• pwd –

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

• q –

  Typed alias for shlex.quote

• quote –

  Typed alias for shlex.quote

• rbytes –

  Load/Read bytes from a fspath

• rbytes_async –

  (ASYNC) Load/Read bytes from a fspath

• rbytes_gen –

  Yield bytes from a given fspath

• rbytes_gen_async –

  Yield (asynchronously) bytes from a given fspath

• rjson –

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

• rjson_async –

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

• rm –

  Remove files & directories in the style of the shell

• rm_gen –

  Remove files & directories in the style of the shell

• rmdir –

  Remove directory at given fspath

• rmfile –

  Remove a file at given fspath

• rstring –

  Load/Read a string given a fspath

• rstring_async –

  (ASYNC) Load/Read a string given a 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

• seconds2hrtime –

  Return hr-time Tuple[int, int] (seconds, nanoseconds)

• 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

• setenv –

  Export/Set an environment variable

• shebang –

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

• shell –

  Run a subprocess synchronously in current shell

• shplit –

  Typed alias for shlex.split

• source –

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

• stat –

  Return the os.stat_result object for a given fspath

• stat_async –

  Async version of os.lstat

• touch –

  Create an empty file given a fspath

• tree –

  Create a directory tree string given a directory path

• walk_gen –

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

• wbytes –

  Write/Save bytes to a fspath

• wbytes_async –

  (ASYNC) Write/Save bytes to a fspath

• wbytes_gen –

  Write/Save bytes to a fspath

• wbytes_gen_async –

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

• where –

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

• which –

  Return the result of shutil.which

• which_lru –

  Return the result of shutil.which and cache the results

• wjson –

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

• wjson_async –

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

• wstring –

  Save/Write a string to fspath

• wstring_async –

  (ASYNC) Save/Write a string to fspath

Done

Done(*args: Dict[_KT, _VT])

Done(*args: Dict[_KT, _VT], **kwargs: _VT)

Done(*args: Mapping[_KT, _VT])

Done(*args: Mapping[_KT, _VT], **kwargs: _VT)

Done(*args: Any, **kwargs: _VT)

Bases: JsonBaseModel

PRun => 'ProcessRun' for finished processes

Methods:

• asdict –

  Return the JsonObj object (and children) as a python dictionary

• check –

  Check returncode and stderr

• completed_process –

  Return subprocess.CompletedProcess object

• defaults_dict –

  Return a dictionary of non-required keys -> default value(s)

• dict –

  Alias for model_dump

• done_dict –

  Return Done object as typed-dict

• done_obj –

  Return Done object typed dict

• dot_items –

  Yield tuples of the form (dot-key, value)

• dot_items_list –

  Return list of tuples of the form (dot-key, value)

• dot_keys –

  Yield the JsonObj's dot-notation keys

• dot_keys_list –

  Return a list of the JsonObj's dot-notation friendly keys

• dot_keys_set –

  Return a set of the JsonObj's dot-notation friendly keys

• dot_lookup –

  Look up JsonObj keys using dot notation as a string

• eject –

  Eject to python-builtin dictionary object

• entries –

  Alias for items

• filter_false –

  Filter key-values where the value is false-y

• filter_none –

  Filter key-values where the value is None but not false-y

• from_dict –

  Return a JsonObj object from a dictionary of data

• from_dict_filtered –

  Create class from dict filtering keys not in (sub)class' fields

• from_json –

  Return a JsonObj object from a json string

• grep –

  Return lines in stdout that have

• has_required_fields –

  Return True/False if the (sub)class has any fields that are required

• is_default –

  Check if the object is equal to the default value for its fields

• items –

  Return an items view of the JsonObj object

• json –

  Alias for model_dumps

• json_parse –

  Return json parsed stdout

• json_parse_stderr –

  Return json parsed stderr

• json_parse_stdout –

  Return json parsed stdout

• keys –

  Return the keys view of the JsonObj object

• parse_json –

  Return json parsed stdout (alias bc I keep flip-flopping the fn name)

• recurse –

  Recursively convert all sub dictionaries to JsonObj objects

• stringify –

  Return JSON string of the JsonObj object (and children)

• sys_print –

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

• to_dict –

  Eject and return object as plain jane dictionary

• to_dict_filter_defaults –

  Eject object and filter key-values equal to (sub)class' default

• to_dict_filter_none –

  Eject object and filter key-values equal to (sub)class' default

• to_json –

  Return JSON string of the JsonObj object (and children)

• to_json_dict –

  Eject object and sub-objects to jsonbourne.JsonObj

• to_json_obj –

  Eject object and sub-objects to jsonbourne.JsonObj

• to_json_obj_filter_defaults –

  Eject to JsonObj and filter key-values equal to (sub)class' default

• to_json_obj_filter_none –

  Eject to JsonObj and filter key-values where the value is None

• validate_type –

  Validate and convert a value to a JsonObj object

• write_stderr –

  Write stderr as a string to a fspath

• write_stdout –

  Write stdout as a string to a fspath

asdict

asdict() -> Dict[_KT, Any]

Return the JsonObj object (and children) as a python dictionary

check

check(
    ok_code: Union[
        int, List[int], Tuple[int, ...], Set[int]
    ] = 0,
) -> None

Check returncode and stderr

Raises:

• DoneError –

  If return code is non-zero and stderr is not None

completed_process

completed_process() -> CompletedProcess[str]

Return subprocess.CompletedProcess object

defaults_dict classmethod

defaults_dict() -> Dict[str, Any]

Return a dictionary of non-required keys -> default value(s)

Returns:

• Dict[str, Any] –

  Dict[str, Any]: Dictionary of non-required keys -> default value

Examples:

>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm')
>>> t.to_dict_filter_defaults()
{}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{})
>>> t = Thing(a=123)
>>> t
Thing(a=123, b='herm')
>>> t.to_dict_filter_defaults()
{'a': 123}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{'a': 123})
>>> t.defaults_dict()
{'a': 1, 'b': 'herm'}

dict

dict(*args: Any, **kwargs: Any) -> Dict[str, Any]

Alias for model_dump

done_dict

done_dict() -> DoneDict

Return Done object as typed-dict

done_obj

done_obj() -> DoneObj

Return Done object typed dict

dot_items

dot_items() -> Iterator[Tuple[Tuple[str, ...], _VT]]

Yield tuples of the form (dot-key, value)

OG-version

def dot_items(self) -> Iterator[Tuple[str, Any]]: return ((dk, self.dot_lookup(dk)) for dk in self.dot_keys())

Readable-version

for k, value in self.items(): value = jsonify(value) if isinstance(value, JsonObj) or hasattr(value, 'dot_items'): yield from ((f"{k}.{dk}", dv) for dk, dv in value.dot_items()) else: yield k, value

dot_items_list

dot_items_list() -> List[Tuple[Tuple[str, ...], Any]]

Return list of tuples of the form (dot-key, value)

dot_keys

dot_keys() -> Iterable[Tuple[str, ...]]

Yield the JsonObj's dot-notation keys

Returns:

• Iterable[Tuple[str, ...]] –

  Iterable[str]: List of the dot-notation friendly keys

The Non-chain version (shown below) is very slightly slower than the itertools.chain version.

NON-CHAIN VERSION:

for k, value in self.items(): value = jsonify(value) if isinstance(value, JsonObj): yield from (f"{k}.{dk}" for dk in value.dot_keys()) else: yield k

dot_keys_list

dot_keys_list(
    sort_keys: bool = False,
) -> List[Tuple[str, ...]]

Return a list of the JsonObj's dot-notation friendly keys

Parameters:

• sort_keys

  (bool, default: False ) –

  Flag to have the dot-keys be returned sorted

Returns:

• List[Tuple[str, ...]] –

  List[str]: List of the dot-notation friendly keys

dot_keys_set

dot_keys_set() -> Set[Tuple[str, ...]]

Return a set of the JsonObj's dot-notation friendly keys

Returns:

• Set[Tuple[str, ...]] –

  Set[str]: List of the dot-notation friendly keys

dot_lookup

dot_lookup(
    key: Union[str, Tuple[str, ...], List[str]],
) -> Any

Look up JsonObj keys using dot notation as a string

Parameters:

• key

  (str) –

  dot-notation key to look up ('key1.key2.third_key')

Returns:

• Any –

  The result of the dot-notation key look up

Raises:

• KeyError –

  Raised if the dot-key is not in in the object

• ValueError –

  Raised if key is not a str/Tuple[str, ...]/List[str]

eject

eject() -> Dict[_KT, _VT]

Eject to python-builtin dictionary object

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

entries

entries() -> ItemsView[_KT, _VT]

Alias for items

filter_false

filter_false(recursive: bool = False) -> JsonObj[_VT]

Filter key-values where the value is false-y

Parameters:

• recursive

  (bool, default: False ) –

  Recurse into sub JsonObjs and dictionaries

Returns:

• JsonObj[_VT] –

  JsonObj that has been filtered

Examples:

>>> d = {
...     'falsey_dict': {},
...     'falsey_list': [],
...     'falsey_string': '',
...     'is_false': False,
...     'a': None,
...     'b': 2,
...     'c': {
...         'd': 'herm',
...         'e': None,
...         'falsey_dict': {},
...         'falsey_list': [],
...         'falsey_string': '',
...         'is_false': False,
...     },
...     }
...
>>> d = JsonObj(d)
>>> print(d)
JsonObj(**{
    'a': None,
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> print(d.filter_false())
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False}
})
>>> print(d.filter_false(recursive=True))
JsonObj(**{
    'b': 2, 'c': {'d': 'herm'}
})

filter_none

filter_none(recursive: bool = False) -> JsonObj[_VT]

Filter key-values where the value is None but not false-y

Parameters:

• recursive

  (bool, default: False ) –

  Recursively filter out None values

Returns:

• JsonObj[_VT] –

  JsonObj that has been filtered of None values

Examples:

>>> d = {
...     'falsey_dict': {},
...     'falsey_list': [],
...     'falsey_string': '',
...     'is_false': False,
...     'a': None,
...     'b': 2,
...     'c': {
...         'd': 'herm',
...         'e': None,
...         'falsey_dict': {},
...         'falsey_list': [],
...         'falsey_string': '',
...         'is_false': False,
...     },
...     }
...
>>> d = JsonObj(d)
>>> print(d)
JsonObj(**{
    'a': None,
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> print(d.filter_none())
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> from pprint import pprint
>>> print(d.filter_none(recursive=True))
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})

from_dict classmethod

from_dict(data: Dict[_KT, _VT]) -> JsonObj[_VT]

Return a JsonObj object from a dictionary of data

from_dict_filtered classmethod

from_dict_filtered(
    dictionary: Dict[str, Any],
) -> JsonBaseModelT

Create class from dict filtering keys not in (sub)class' fields

from_json classmethod

from_json(json_string: Union[bytes, str]) -> JsonObj[_VT]

Return a JsonObj object from a json string

Parameters:

• json_string

  (str) –

  JSON string to convert to a JsonObj

Returns:

• JsonObjT ( JsonObj[_VT] ) –

  JsonObj object for the given JSON string

grep

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

Return lines in stdout that have

Parameters:

• string

  (str) –

  String to search for

Returns:

• List[str] –

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

has_required_fields classmethod

has_required_fields() -> bool

Return True/False if the (sub)class has any fields that are required

Returns:

• bool ( bool ) –

  True if any fields for a (sub)class are required

is_default

is_default() -> bool

Check if the object is equal to the default value for its fields

Returns:

• bool –

  True if object is equal to the default value for all fields; False otherwise

Examples:

>>> class Thing(JsonBaseModel):
...    a: int = 1
...    b: str = 'b'
...
>>> t = Thing()
>>> t.is_default()
True
>>> t = Thing(a=2)
>>> t.is_default()
False

items

items() -> ItemsView[_KT, _VT]

Return an items view of the JsonObj object

json

json(*args: Any, **kwargs: Any) -> str

Alias for model_dumps

json_parse

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

Return json parsed stdout

json_parse_stderr

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

Return json parsed stderr

json_parse_stdout

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

Return json parsed stdout

keys

keys() -> KeysView[_KT]

Return the keys view of the JsonObj object

parse_json

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

Return json parsed stdout (alias bc I keep flip-flopping the fn name)

recurse

recurse() -> None

Recursively convert all sub dictionaries to JsonObj objects

stringify

stringify(
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    **kwargs: Any,
) -> str

Return JSON string of the JsonObj object (and children)

    Args:
        fmt (bool): If True, return a JSON string with newlines and indentation
        pretty (bool): If True, return a JSON string with newlines and indentation
        sort_keys (bool): Sort dictionary keys if True
        append_newline (bool): Append a newline '

' to JSON string if True default: default function hook for JSON serialization **kwargs (Any): additional kwargs to be passed down to jsonlib.dumps

    Returns:
        str: JSON string of the JsonObj object

sys_print

sys_print() -> None

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

to_dict

to_dict() -> Dict[str, Any]

Eject and return object as plain jane dictionary

to_dict_filter_defaults

to_dict_filter_defaults() -> Dict[str, Any]

Eject object and filter key-values equal to (sub)class' default

Examples:

>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm')
>>> t.to_dict_filter_defaults()
{}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{})
>>> t = Thing(a=123)
>>> t
Thing(a=123, b='herm')
>>> t.to_dict_filter_defaults()
{'a': 123}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{'a': 123})

to_dict_filter_none

to_dict_filter_none() -> Dict[str, Any]

Eject object and filter key-values equal to (sub)class' default

Examples:

>>> from typing import Optional
>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...     c: Optional[str] = None
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm', c=None)
>>> t.to_dict_filter_none()
{'a': 1, 'b': 'herm'}
>>> t.to_json_obj_filter_none()
JsonObj(**{'a': 1, 'b': 'herm'})

to_json

to_json(
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    **kwargs: Any,
) -> str

Return JSON string of the JsonObj object (and children)

    Args:
        fmt (bool): If True, return a JSON string with newlines and indentation
        pretty (bool): If True, return a JSON string with newlines and indentation
        sort_keys (bool): Sort dictionary keys if True
        append_newline (bool): Append a newline '

' to JSON string if True default: default function hook for JSON serialization **kwargs (Any): additional kwargs to be passed down to jsonlib.dumps

    Returns:
        str: JSON string of the JsonObj object

to_json_dict

to_json_dict() -> JsonObj[Any]

Eject object and sub-objects to jsonbourne.JsonObj

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

to_json_obj

to_json_obj() -> JsonObj[Any]

Eject object and sub-objects to jsonbourne.JsonObj

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

to_json_obj_filter_defaults

to_json_obj_filter_defaults() -> JsonObj[Any]

Eject to JsonObj and filter key-values equal to (sub)class' default

to_json_obj_filter_none

to_json_obj_filter_none() -> JsonObj[Any]

Eject to JsonObj and filter key-values where the value is None

validate_type classmethod

validate_type(val: Any) -> JsonObj[_VT]

Validate and convert a value to a JsonObj 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 ) –

  Flag to append to file or plain write to file

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 ) –

  Flag to append to file or plain write to file

DoneError

DoneError(done: Done)

Bases: SubprocessError

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

Attributes:

• cmd (str) –

  command that was run

• returncode (int) –

  exit status of the process

• stdout (str) –

  standard output (stdout) of the process

• stderr (str) –

  standard error (stderr) of the process

DoneObj

Bases: TypedDict

Todo

deprecate this in favor of DoneDict

Flag

Flag obj

Examples:

>>> Flag.__help
'--help'
>>> Flag._v
'-v'

FlagMeta

Bases: type

Meta class

Methods:

• attr2flag –

  Convert and return attr to string

attr2flag cached staticmethod

attr2flag(string: str) -> str

Convert and return attr to string

HrTime

HrTime(*args: Dict[_KT, _VT])

HrTime(*args: Dict[_KT, _VT], **kwargs: _VT)

HrTime(*args: Mapping[_KT, _VT])

HrTime(*args: Mapping[_KT, _VT], **kwargs: _VT)

HrTime(*args: Any, **kwargs: _VT)

Bases: JsonBaseModel

High resolution time

Methods:

• asdict –

  Return the JsonObj object (and children) as a python dictionary

• defaults_dict –

  Return a dictionary of non-required keys -> default value(s)

• dict –

  Alias for model_dump

• dot_items –

  Yield tuples of the form (dot-key, value)

• dot_items_list –

  Return list of tuples of the form (dot-key, value)

• dot_keys –

  Yield the JsonObj's dot-notation keys

• dot_keys_list –

  Return a list of the JsonObj's dot-notation friendly keys

• dot_keys_set –

  Return a set of the JsonObj's dot-notation friendly keys

• dot_lookup –

  Look up JsonObj keys using dot notation as a string

• eject –

  Eject to python-builtin dictionary object

• entries –

  Alias for items

• filter_false –

  Filter key-values where the value is false-y

• filter_none –

  Filter key-values where the value is None but not false-y

• from_dict –

  Return a JsonObj object from a dictionary of data

• from_dict_filtered –

  Create class from dict filtering keys not in (sub)class' fields

• from_json –

  Return a JsonObj object from a json string

• from_seconds –

  Return HrTime object from seconds

• has_required_fields –

  Return True/False if the (sub)class has any fields that are required

• is_default –

  Check if the object is equal to the default value for its fields

• items –

  Return an items view of the JsonObj object

• json –

  Alias for model_dumps

• keys –

  Return the keys view of the JsonObj object

• recurse –

  Recursively convert all sub dictionaries to JsonObj objects

• stringify –

  Return JSON string of the JsonObj object (and children)

• to_dict –

  Eject and return object as plain jane dictionary

• to_dict_filter_defaults –

  Eject object and filter key-values equal to (sub)class' default

• to_dict_filter_none –

  Eject object and filter key-values equal to (sub)class' default

• to_json –

  Return JSON string of the JsonObj object (and children)

• to_json_dict –

  Eject object and sub-objects to jsonbourne.JsonObj

• to_json_obj –

  Eject object and sub-objects to jsonbourne.JsonObj

• to_json_obj_filter_defaults –

  Eject to JsonObj and filter key-values equal to (sub)class' default

• to_json_obj_filter_none –

  Eject to JsonObj and filter key-values where the value is None

• validate_type –

  Validate and convert a value to a JsonObj object

asdict

asdict() -> Dict[_KT, Any]

Return the JsonObj object (and children) as a python dictionary

defaults_dict classmethod

defaults_dict() -> Dict[str, Any]

Return a dictionary of non-required keys -> default value(s)

Returns:

• Dict[str, Any] –

  Dict[str, Any]: Dictionary of non-required keys -> default value

Examples:

>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm')
>>> t.to_dict_filter_defaults()
{}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{})
>>> t = Thing(a=123)
>>> t
Thing(a=123, b='herm')
>>> t.to_dict_filter_defaults()
{'a': 123}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{'a': 123})
>>> t.defaults_dict()
{'a': 1, 'b': 'herm'}

dict

dict(*args: Any, **kwargs: Any) -> Dict[str, Any]

Alias for model_dump

dot_items

dot_items() -> Iterator[Tuple[Tuple[str, ...], _VT]]

Yield tuples of the form (dot-key, value)

OG-version

def dot_items(self) -> Iterator[Tuple[str, Any]]: return ((dk, self.dot_lookup(dk)) for dk in self.dot_keys())

Readable-version

for k, value in self.items(): value = jsonify(value) if isinstance(value, JsonObj) or hasattr(value, 'dot_items'): yield from ((f"{k}.{dk}", dv) for dk, dv in value.dot_items()) else: yield k, value

dot_items_list

dot_items_list() -> List[Tuple[Tuple[str, ...], Any]]

Return list of tuples of the form (dot-key, value)

dot_keys

dot_keys() -> Iterable[Tuple[str, ...]]

Yield the JsonObj's dot-notation keys

Returns:

• Iterable[Tuple[str, ...]] –

  Iterable[str]: List of the dot-notation friendly keys

The Non-chain version (shown below) is very slightly slower than the itertools.chain version.

NON-CHAIN VERSION:

for k, value in self.items(): value = jsonify(value) if isinstance(value, JsonObj): yield from (f"{k}.{dk}" for dk in value.dot_keys()) else: yield k

dot_keys_list

dot_keys_list(
    sort_keys: bool = False,
) -> List[Tuple[str, ...]]

Return a list of the JsonObj's dot-notation friendly keys

Parameters:

• sort_keys

  (bool, default: False ) –

  Flag to have the dot-keys be returned sorted

Returns:

• List[Tuple[str, ...]] –

  List[str]: List of the dot-notation friendly keys

dot_keys_set

dot_keys_set() -> Set[Tuple[str, ...]]

Return a set of the JsonObj's dot-notation friendly keys

Returns:

• Set[Tuple[str, ...]] –

  Set[str]: List of the dot-notation friendly keys

dot_lookup

dot_lookup(
    key: Union[str, Tuple[str, ...], List[str]],
) -> Any

Look up JsonObj keys using dot notation as a string

Parameters:

• key

  (str) –

  dot-notation key to look up ('key1.key2.third_key')

Returns:

• Any –

  The result of the dot-notation key look up

Raises:

• KeyError –

  Raised if the dot-key is not in in the object

• ValueError –

  Raised if key is not a str/Tuple[str, ...]/List[str]

eject

eject() -> Dict[_KT, _VT]

Eject to python-builtin dictionary object

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

entries

entries() -> ItemsView[_KT, _VT]

Alias for items

filter_false

filter_false(recursive: bool = False) -> JsonObj[_VT]

Filter key-values where the value is false-y

Parameters:

• recursive

  (bool, default: False ) –

  Recurse into sub JsonObjs and dictionaries

Returns:

• JsonObj[_VT] –

  JsonObj that has been filtered

Examples:

>>> d = {
...     'falsey_dict': {},
...     'falsey_list': [],
...     'falsey_string': '',
...     'is_false': False,
...     'a': None,
...     'b': 2,
...     'c': {
...         'd': 'herm',
...         'e': None,
...         'falsey_dict': {},
...         'falsey_list': [],
...         'falsey_string': '',
...         'is_false': False,
...     },
...     }
...
>>> d = JsonObj(d)
>>> print(d)
JsonObj(**{
    'a': None,
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> print(d.filter_false())
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False}
})
>>> print(d.filter_false(recursive=True))
JsonObj(**{
    'b': 2, 'c': {'d': 'herm'}
})

filter_none

filter_none(recursive: bool = False) -> JsonObj[_VT]

Filter key-values where the value is None but not false-y

Parameters:

• recursive

  (bool, default: False ) –

  Recursively filter out None values

Returns:

• JsonObj[_VT] –

  JsonObj that has been filtered of None values

Examples:

>>> d = {
...     'falsey_dict': {},
...     'falsey_list': [],
...     'falsey_string': '',
...     'is_false': False,
...     'a': None,
...     'b': 2,
...     'c': {
...         'd': 'herm',
...         'e': None,
...         'falsey_dict': {},
...         'falsey_list': [],
...         'falsey_string': '',
...         'is_false': False,
...     },
...     }
...
>>> d = JsonObj(d)
>>> print(d)
JsonObj(**{
    'a': None,
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> print(d.filter_none())
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> from pprint import pprint
>>> print(d.filter_none(recursive=True))
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})

from_dict classmethod

from_dict(data: Dict[_KT, _VT]) -> JsonObj[_VT]

Return a JsonObj object from a dictionary of data

from_dict_filtered classmethod

from_dict_filtered(
    dictionary: Dict[str, Any],
) -> JsonBaseModelT

Create class from dict filtering keys not in (sub)class' fields

from_json classmethod

from_json(json_string: Union[bytes, str]) -> JsonObj[_VT]

Return a JsonObj object from a json string

Parameters:

• json_string

  (str) –

  JSON string to convert to a JsonObj

Returns:

• JsonObjT ( JsonObj[_VT] ) –

  JsonObj object for the given JSON string

from_seconds classmethod

from_seconds(seconds: float) -> 'HrTime'

Return HrTime object from seconds

Parameters:

• seconds

  (float) –

  number of seconds

Returns:

• 'HrTime' –

  HrTime object

has_required_fields classmethod

has_required_fields() -> bool

Return True/False if the (sub)class has any fields that are required

Returns:

• bool ( bool ) –

  True if any fields for a (sub)class are required

is_default

is_default() -> bool

Check if the object is equal to the default value for its fields

Returns:

• bool –

  True if object is equal to the default value for all fields; False otherwise

Examples:

>>> class Thing(JsonBaseModel):
...    a: int = 1
...    b: str = 'b'
...
>>> t = Thing()
>>> t.is_default()
True
>>> t = Thing(a=2)
>>> t.is_default()
False

items

items() -> ItemsView[_KT, _VT]

Return an items view of the JsonObj object

json

json(*args: Any, **kwargs: Any) -> str

Alias for model_dumps

keys

keys() -> KeysView[_KT]

Return the keys view of the JsonObj object

recurse

recurse() -> None

Recursively convert all sub dictionaries to JsonObj objects

stringify

stringify(
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    **kwargs: Any,
) -> str

Return JSON string of the JsonObj object (and children)

    Args:
        fmt (bool): If True, return a JSON string with newlines and indentation
        pretty (bool): If True, return a JSON string with newlines and indentation
        sort_keys (bool): Sort dictionary keys if True
        append_newline (bool): Append a newline '

' to JSON string if True default: default function hook for JSON serialization **kwargs (Any): additional kwargs to be passed down to jsonlib.dumps

    Returns:
        str: JSON string of the JsonObj object

to_dict

to_dict() -> Dict[str, Any]

Eject and return object as plain jane dictionary

to_dict_filter_defaults

to_dict_filter_defaults() -> Dict[str, Any]

Eject object and filter key-values equal to (sub)class' default

Examples:

>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm')
>>> t.to_dict_filter_defaults()
{}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{})
>>> t = Thing(a=123)
>>> t
Thing(a=123, b='herm')
>>> t.to_dict_filter_defaults()
{'a': 123}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{'a': 123})

to_dict_filter_none

to_dict_filter_none() -> Dict[str, Any]

Eject object and filter key-values equal to (sub)class' default

Examples:

>>> from typing import Optional
>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...     c: Optional[str] = None
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm', c=None)
>>> t.to_dict_filter_none()
{'a': 1, 'b': 'herm'}
>>> t.to_json_obj_filter_none()
JsonObj(**{'a': 1, 'b': 'herm'})

to_json

to_json(
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    **kwargs: Any,
) -> str

Return JSON string of the JsonObj object (and children)

    Args:
        fmt (bool): If True, return a JSON string with newlines and indentation
        pretty (bool): If True, return a JSON string with newlines and indentation
        sort_keys (bool): Sort dictionary keys if True
        append_newline (bool): Append a newline '

' to JSON string if True default: default function hook for JSON serialization **kwargs (Any): additional kwargs to be passed down to jsonlib.dumps

    Returns:
        str: JSON string of the JsonObj object

to_json_dict

to_json_dict() -> JsonObj[Any]

Eject object and sub-objects to jsonbourne.JsonObj

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

to_json_obj

to_json_obj() -> JsonObj[Any]

Eject object and sub-objects to jsonbourne.JsonObj

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

to_json_obj_filter_defaults

to_json_obj_filter_defaults() -> JsonObj[Any]

Eject to JsonObj and filter key-values equal to (sub)class' default

to_json_obj_filter_none

to_json_obj_filter_none() -> JsonObj[Any]

Eject to JsonObj and filter key-values where the value is None

validate_type classmethod

validate_type(val: Any) -> JsonObj[_VT]

Validate and convert a value to a JsonObj object

LIN

Bases: LIN

Linux (and Mac) shell commands/methods container

Methods:

• link_dir –

  Make a directory symlink

• link_dirs –

  Make multiple directory symlinks

• link_file –

  Make a file symlink

• link_files –

  Make multiple file symlinks

• rsync –

  Run an rsync subprocess

• rsync_args –

  Return args for rsync command on linux/mac

• 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

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

  (str, 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.

rsync staticmethod

rsync(
    src: str,
    dest: str,
    delete: bool = False,
    mkdirs: bool = False,
    dry_run: bool = False,
    exclude: Optional[IterableStr] = None,
    include: Optional[IterableStr] = 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

  (Optional[IterableStr], default: None ) –

  Strings/patterns to exclude

• include

  (Optional[IterableStr], 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

rsync_args staticmethod

rsync_args(
    src: str,
    dest: str,
    delete: bool = False,
    dry_run: bool = False,
    exclude: Optional[IterableStr] = None,
    include: Optional[IterableStr] = 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

  (Optional[IterableStr], default: None ) –

  Strings/patterns to exclude

• include

  (Optional[IterableStr], 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

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: IterableStr) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (IterableStr) –

  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: IterableStr) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (IterableStr) –

  Iterable of paths to links

Stdio

Bases: IntEnum

Standard-io enum object

WIN

Bases: WIN

Windows shell commands/methods container

Methods:

• link_dir –

  Make a directory symlink

• link_dirs –

  Make multiple directory symlinks

• link_file –

  Make a file symlink

• link_files –

  Make multiple file symlinks

• robocopy –

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

• robocopy_args –

  Return list of robocopy command args

• 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

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

robocopy staticmethod

robocopy(
    src: str,
    dest: str,
    *,
    mkdirs: bool = True,
    delete: bool = False,
    exclude_files: Optional[Iterable[str]] = None,
    exclude_dirs: Optional[Iterable[str]] = 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

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

  Strings/patterns with which to exclude files

• exclude_dirs

  (Optional[Iterable[str]], 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.

robocopy_args staticmethod

robocopy_args(
    src: str,
    dest: str,
    *,
    delete: bool = False,
    exclude_files: Optional[List[str]] = None,
    exclude_dirs: Optional[List[str]] = 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

  (Optional[List[str]], default: None ) –

  Strings/patterns with which to exclude files

• exclude_dirs

  (Optional[List[str]], 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.

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: IterableStr) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (IterableStr) –

  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: IterableStr) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (IterableStr) –

  Iterable of paths to links

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

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

decode_stdio_bytes

decode_stdio_bytes(
    stdio_bytes: Union[str, bytes], lf: bool = True
) -> str

Return Stdio bytes from stdout/stderr as a string

Args:
    stdio_bytes (bytes): STDOUT/STDERR bytes
    lf (bool): Replace `

line endings with `

Returns:
    str: decoded stdio bytes

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

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

dirpath_gen

dirpath_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = False,
    topdown: bool = True,
    onerror: Optional[Callable[[OSError], Any]] = 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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)

do

do(
    *popenargs: PopenArgs,
    args: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    extenv: bool = True,
    cwd: Optional[FsPath] = None,
    shell: bool = False,
    check: bool = False,
    tee: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    timeout: Optional[Union[float, int]] = None,
    ok_code: Union[
        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

  (Optional[PopenArgs], default: None ) –

  Args as strings for the subprocess

• env

  (Optional[Dict[str, str]], 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

  (Optional[FsPath], 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

  (Optional[int], default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (Union[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

do_async async

do_async(
    *popenargs: PopenArgs,
    args: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    extenv: bool = True,
    cwd: Optional[str] = None,
    shell: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    check: bool = False,
    timeout: Optional[float] = None,
    ok_code: Union[
        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

  (Optional[PopenArgs], default: None ) –

  Args as strings for the subprocess

• check

  (bool, default: False ) –

  Check the result returncode

• env

  (Optional[Dict[str, str]], default: None ) –

  Environment variables as a dictionary (Default value = None)

• extenv

  (bool, default: True ) –

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

• cwd

  (Optional[str], 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

  (Optional[int], default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (Union[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: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    extenv: bool = True,
    cwd: Optional[str] = None,
    shell: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    check: bool = False,
    timeout: Optional[float] = None,
    ok_code: Union[
        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

  (Optional[PopenArgs], default: None ) –

  Args as strings for the subprocess

• check

  (bool, default: False ) –

  Check the result returncode

• env

  (Optional[Dict[str, str]], default: None ) –

  Environment variables as a dictionary (Default value = None)

• cwd

  (Optional[str], 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

  (Optional[int], default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (Union[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

exists

exists(fspath: FsPath) -> bool

Return True if the given path exists; False otherwise

export

export(
    key: str, val: Optional[str] = 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

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 wstring
>>> wstring(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: Optional[Callable[[OSError], Any]] = 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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

flatten_args

flatten_args(*args: Union[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']

fspath

fspath(fspath: FsPath) -> str

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

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

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_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

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_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

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

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

path_gen

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

Yield all filepaths as pathlib.Path objects beneath a dirpath

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

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\'"\'"\'\''

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\'"\'"\'\''

rbytes

rbytes(filepath: FsPath) -> bytes

Load/Read bytes from a fspath

Parameters:

• filepath

  (FsPath) –

  fspath read as bytes

Returns:

• bytes –

  bytes from the fspath

Examples:

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

rbytes_async async

rbytes_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 rbytes_async, wbytes_async
>>> from asyncio import run as aiorun
>>> fspath = "rbytes_async.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> aiorun(wbytes_async(fspath, bites_to_save))
20
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> aiorun(rbytes_async(fspath))
b'These are some bytes'
>>> import os; os.remove(fspath)

rbytes_gen

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

Yield bytes from a given fspath

rbytes_gen_async async

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

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[Union[bytes, str]] –

  bytes from AsyncIterable[bytes] of the file bytes

Examples:

>>> from os import remove
>>> from asyncio import run
>>> from shellfish.fs._async import wbytes_gen_async, rbytes_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(wbytes_gen_async(fspath, bites_to_save))
35
>>> async def read():
...     async for b in rbytes_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(wbytes_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(wbytes_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)

rjson

rjson(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 rjson, wjson

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_dict.doctest.json"
>>> wjson(fspath, data)
19
>>> rjson(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"
>>> wjson(fspath, data)
25
>>> rjson(fspath)
[['a', 1], ['b', 2], ['c', 3]]
>>> os.remove(fspath)

rjson_async async

rjson_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 rjson_async, wjson_async

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_async_dict.doctest.json"
>>> run(wjson_async(fspath, data))
19
>>> run(rjson_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(wjson_async(fspath, data))
25
>>> run(rjson_async(fspath))
[['a', 1], ['b', 2], ['c', 3]]

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

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

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

rstring

rstring(
    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 rstring, wstring
>>> fspath = "lstring.doctest.txt"
>>> sstring(fspath, r'Check out this string')
21
>>> lstring(fspath)
'Check out this string'
>>> import os; os.remove(fspath)

rstring_async async

rstring_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

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

seconds2hrtime

seconds2hrtime(
    seconds: Union[float, int],
) -> Tuple[int, int]

Return hr-time Tuple[int, int] (seconds, nanoseconds)

Parameters:

• seconds

  (float) –

  number of seconds

Returns:

• Tuple[int, int] –

  Tuple[int, int]: (seconds, nanoseconds)

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

setenv

setenv(
    key: str, val: Optional[str] = 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

shebang

shebang(fspath: FsPath) -> Union[None, str]

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

Parameters:

• fspath

  (FsPath) –

  Path to file that might have a shebang

Returns:

• Union[None, str] –

  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)

shell

shell(
    *popenargs: PopenArgs,
    args: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    shell: bool = True,
    extenv: bool = True,
    cwd: Optional[FsPath] = None,
    check: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    timeout: Optional[Union[float, int]] = None,
    ok_code: Union[
        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

  (Optional[PopenArgs], default: None ) –

  Args as strings for the subprocess

• env

  (Optional[Dict[str, str]], default: None ) –

  Environment variables as a dictionary (Default value = None)

• shell

  (bool, default: True ) –

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

• extenv

  (bool, default: True ) –

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

• cwd

  (Optional[FsPath], 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

  (Optional[int], default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (Union[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

shplit

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

Typed alias for shlex.split

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

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.lstat

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

tree

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

Create a directory tree string given a directory path

Parameters:

• dirpath

  (FsPath) –

  Directory string to make tree for

• filterfn

  (Optional[Callable[[str], bool]], 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)

walk_gen

walk_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = True,
    topdown: bool = True,
    onerror: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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)

wbytes

wbytes(
    filepath: FsPath,
    bites: bytes,
    *,
    append: bool = False,
    chmod: Optional[int] = 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 rbytes, wbytes
>>> fspath = "wbytes.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> wbytes(fspath, bites_to_save)
20
>>> rbytes(fspath)
b'These are some bytes'
>>> import os; os.remove(fspath)

wbytes_async async

wbytes_async(
    filepath: FsPath,
    bites: bytes,
    *,
    append: bool = False,
    chmod: Optional[int] = 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

  (Optional[int], default: None ) –

  chmod the fspath to this mode after writing

Returns:

• int –

  None

Examples:

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

wbytes_gen

wbytes_gen(
    filepath: FsPath,
    bytes_gen: Iterable[bytes],
    append: bool = False,
    chmod: Optional[int] = 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 rbytes, wbytes
>>> 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!')
>>> wbytes_gen(fspath, (b for b in bites_to_save))
35
>>> rbytes(fspath)
b'These are some bytes... more bytes!'
>>> import os; os.remove(fspath)

wbytes_gen_async async

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

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

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• bytes_gen

  (Union[Iterable[bytes], AsyncIterable[bytes]]) –

  AsyncIterable/Iterator of bytes to write

• append

  (bool, default: False ) –

  Append to the fspath if True; otherwise overwrite

• chmod

  (Optional[int], 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 wbytes_gen_async, rbytes_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(wbytes_gen_async(fspath, bites_to_save))
35
>>> async def read():
...     async for b in rbytes_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(wbytes_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(wbytes_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)

where

where(
    cmd: str, path: Optional[str] = None
) -> Optional[str]

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:

• Optional[str] –

  Optional[str]: path to command/exe

which

which(
    cmd: str, path: Optional[str] = None
) -> Optional[str]

Return the result of shutil.which

Parameters:

• cmd

  (str) –

  Command/exe to find path of

• path

  (str, default: None ) –

  System path to use

Returns:

• Optional[str] –

  Optional[str]: path to command/exe

which_lru cached

which_lru(
    cmd: str, path: Optional[str] = None
) -> Optional[str]

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:

• Optional[str] –

  Optional[str]: path to command/exe

wjson

wjson(
    filepath: FsPath,
    data: Any,
    *,
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    chmod: Optional[int] = 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

  (Optional[Callable[[Any], Any]], 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 rjson, wjson

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_dict.doctest.json"
>>> wjson(fspath, data)
19
>>> rjson(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"
>>> wjson(fspath, data)
25
>>> rjson(fspath)
[['a', 1], ['b', 2], ['c', 3]]
>>> os.remove(fspath)

wjson_async async

wjson_async(
    filepath: FsPath,
    data: Any,
    *,
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    append: bool = False,
    chmod: Optional[int] = 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

  (Optional[Callable[[Any], Any]], 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 rjson_async, wjson_async

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "wjson_async_dict.doctest.json"
>>> run(wjson_async(fspath, data))
19
>>> run(rjson_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(wjson_async(fspath, data))
25
>>> run(rjson_async(fspath))
[['a', 1], ['b', 2], ['c', 3]]

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

wstring

wstring(
    filepath: FsPath,
    string: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    chmod: Optional[int] = 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 rstring, wstring
>>> fspath = "sstring.doctest.txt"
>>> wstring(fspath, r'Check out this string')
21
>>> rstring(fspath)
'Check out this string'
>>> import os; os.remove(fspath)

wstring_async async

wstring_async(
    filepath: FsPath,
    string: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    chmod: Optional[int] = 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

---

fs

file-system utils

Modules:

• promises –

  shellfish.fs.promises

Classes:

• Stdio –

  Standard-io enum object

Functions:

• 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

• rbytes –

  Load/Read bytes from a fspath

• rbytes_async –

  (ASYNC) Load/Read bytes from a fspath

• rbytes_gen –

  Yield bytes from a given fspath

• rbytes_gen_async –

  Yield (asynchronously) bytes from a given fspath

• rjson –

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

• rjson_async –

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

• rm –

  Remove files & directories in the style of the shell

• rm_gen –

  Remove files & directories in the style of the shell

• rmdir –

  Remove directory at given fspath

• rmfile –

  Remove a file at given fspath

• rstring –

  Load/Read a string given a fspath

• rstring_async –

  (ASYNC) Load/Read a string given a 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.lstat

• touch –

  Create an empty file given a fspath

• walk_gen –

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

• wbytes –

  Write/Save bytes to a fspath

• wbytes_async –

  (ASYNC) Write/Save bytes to a fspath

• wbytes_gen –

  Write/Save bytes to a fspath

• wbytes_gen_async –

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

• wjson –

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

• wjson_async –

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

• wstring –

  Save/Write a string to fspath

• wstring_async –

  (ASYNC) Save/Write a string to fspath

Stdio

Bases: IntEnum

Standard-io enum object

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: Optional[Callable[[OSError], Any]] = 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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 wstring
>>> wstring(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: Optional[Callable[[OSError], Any]] = 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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: Optional[Callable[[OSError], Any]] = None,
    followlinks: bool = False,
    check: bool = True,
) -> Iterator[Path]

Yield all filepaths as pathlib.Path objects beneath a dirpath

rbytes

rbytes(filepath: FsPath) -> bytes

Load/Read bytes from a fspath

Parameters:

• filepath

  (FsPath) –

  fspath read as bytes

Returns:

• bytes –

  bytes from the fspath

Examples:

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

rbytes_async async

rbytes_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 rbytes_async, wbytes_async
>>> from asyncio import run as aiorun
>>> fspath = "rbytes_async.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> aiorun(wbytes_async(fspath, bites_to_save))
20
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> aiorun(rbytes_async(fspath))
b'These are some bytes'
>>> import os; os.remove(fspath)

rbytes_gen

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

Yield bytes from a given fspath

rbytes_gen_async async

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

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[Union[bytes, str]] –

  bytes from AsyncIterable[bytes] of the file bytes

Examples:

>>> from os import remove
>>> from asyncio import run
>>> from shellfish.fs._async import wbytes_gen_async, rbytes_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(wbytes_gen_async(fspath, bites_to_save))
35
>>> async def read():
...     async for b in rbytes_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(wbytes_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(wbytes_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)

rjson

rjson(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 rjson, wjson

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_dict.doctest.json"
>>> wjson(fspath, data)
19
>>> rjson(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"
>>> wjson(fspath, data)
25
>>> rjson(fspath)
[['a', 1], ['b', 2], ['c', 3]]
>>> os.remove(fspath)

rjson_async async

rjson_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 rjson_async, wjson_async

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_async_dict.doctest.json"
>>> run(wjson_async(fspath, data))
19
>>> run(rjson_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(wjson_async(fspath, data))
25
>>> run(rjson_async(fspath))
[['a', 1], ['b', 2], ['c', 3]]

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

rm

rm(
    fspath: FsPath,
    *,
    force: bool = False,
    recursive: bool = False,
    dryrun: bool = False,
    verbose: bool = False,
) -> Union[List[str], None]

Remove files & directories in the style of the shell

Parameters:

• fspath

  (FsPath) –

  Path to file or directory to remove

• force

  (bool, default: False ) –

  ignore errors and missing files/dirs; default is False

• 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

• verbose

  (bool, default: False ) –

  Print the files being removed

Raises:

• ValueError –

  If recursive and r are False and fspath is a directory

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

rstring

rstring(
    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 rstring, wstring
>>> fspath = "lstring.doctest.txt"
>>> sstring(fspath, r'Check out this string')
21
>>> lstring(fspath)
'Check out this string'
>>> import os; os.remove(fspath)

rstring_async async

rstring_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

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) -> Union[None, str]

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

Parameters:

• fspath

  (FsPath) –

  Path to file that might have a shebang

Returns:

• Union[None, str] –

  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.lstat

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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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)

wbytes

wbytes(
    filepath: FsPath,
    bites: bytes,
    *,
    append: bool = False,
    chmod: Optional[int] = 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 rbytes, wbytes
>>> fspath = "wbytes.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> wbytes(fspath, bites_to_save)
20
>>> rbytes(fspath)
b'These are some bytes'
>>> import os; os.remove(fspath)

wbytes_async async

wbytes_async(
    filepath: FsPath,
    bites: bytes,
    *,
    append: bool = False,
    chmod: Optional[int] = 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

  (Optional[int], default: None ) –

  chmod the fspath to this mode after writing

Returns:

• int –

  None

Examples:

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

wbytes_gen

wbytes_gen(
    filepath: FsPath,
    bytes_gen: Iterable[bytes],
    append: bool = False,
    chmod: Optional[int] = 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 rbytes, wbytes
>>> 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!')
>>> wbytes_gen(fspath, (b for b in bites_to_save))
35
>>> rbytes(fspath)
b'These are some bytes... more bytes!'
>>> import os; os.remove(fspath)

wbytes_gen_async async

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

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

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• bytes_gen

  (Union[Iterable[bytes], AsyncIterable[bytes]]) –

  AsyncIterable/Iterator of bytes to write

• append

  (bool, default: False ) –

  Append to the fspath if True; otherwise overwrite

• chmod

  (Optional[int], 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 wbytes_gen_async, rbytes_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(wbytes_gen_async(fspath, bites_to_save))
35
>>> async def read():
...     async for b in rbytes_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(wbytes_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(wbytes_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)

wjson

wjson(
    filepath: FsPath,
    data: Any,
    *,
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    chmod: Optional[int] = 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

  (Optional[Callable[[Any], Any]], 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 rjson, wjson

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_dict.doctest.json"
>>> wjson(fspath, data)
19
>>> rjson(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"
>>> wjson(fspath, data)
25
>>> rjson(fspath)
[['a', 1], ['b', 2], ['c', 3]]
>>> os.remove(fspath)

wjson_async async

wjson_async(
    filepath: FsPath,
    data: Any,
    *,
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    append: bool = False,
    chmod: Optional[int] = 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

  (Optional[Callable[[Any], Any]], 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 rjson_async, wjson_async

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "wjson_async_dict.doctest.json"
>>> run(wjson_async(fspath, data))
19
>>> run(rjson_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(wjson_async(fspath, data))
25
>>> run(rjson_async(fspath))
[['a', 1], ['b', 2], ['c', 3]]

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

wstring

wstring(
    filepath: FsPath,
    string: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    chmod: Optional[int] = 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 rstring, wstring
>>> fspath = "sstring.doctest.txt"
>>> wstring(fspath, r'Check out this string')
21
>>> rstring(fspath)
'Check out this string'
>>> import os; os.remove(fspath)

wstring_async async

wstring_async(
    filepath: FsPath,
    string: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    chmod: Optional[int] = 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

sh

shell utils

Classes:

• Done –

  PRun => 'ProcessRun' for finished processes

• DoneError –

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

• DoneObj –

  Todo

  deprecate this in favor of DoneDict

• Flag –

  Flag obj

• FlagMeta –

  Meta class

• HrTime –

  High resolution time

• HrTimeDict –

  High resolution time

• HrTimeObj –

  Todo

  deprecate this in favor of HrTimeDict

• LIN –

  Linux (and Mac) shell commands/methods container

• Stdio –

  Standard-io enum object

• WIN –

  Windows shell commands/methods container

Functions:

• basename –

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

• cd –

  Change directory to given dirpath; alias for os.chdir

• 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

• decode_stdio_bytes –

  Return Stdio bytes from stdout/stderr as a string

• 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

• dirname –

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

• dirpath_gen –

  Yield all dirpaths as pathlib.Path objects beneath a dirpath

• dirs_gen –

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

• do –

  Run a subprocess synchronously

• do_async –

  Run a subprocess and await its completion

• do_asyncify –

  Run a subprocess asynchronously using asyncified version of do

• doa –

  Run a subprocess and await its completion

• echo –

  Print/echo function

• exists –

  Return True if the given path exists; False otherwise

• export –

  Export/Set an environment variable

• 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

• flatten_args –

  Flatten possibly nested iterables of sequences to a list of strings

• 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

• ls –

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

• ls_async –

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

• ls_dirs –

  List the directories in a given directory path

• ls_files –

  List the files in a given directory path

• ls_files_dirs –

  List the files and directories given directory path

• 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

• mv –

  Move file(s) like on the command line

• path_gen –

  Yield all filepaths as pathlib.Path objects beneath a dirpath

• pstderr –

  Get the STDERR as a string from a subprocess

• pstdout –

  Get the STDOUT as a string from a subprocess

• pstdout_pstderr –

  Get the STDOUT and STDERR as strings from a subprocess

• pwd –

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

• q –

  Typed alias for shlex.quote

• quote –

  Typed alias for shlex.quote

• rbytes –

  Load/Read bytes from a fspath

• rbytes_async –

  (ASYNC) Load/Read bytes from a fspath

• rbytes_gen –

  Yield bytes from a given fspath

• rbytes_gen_async –

  Yield (asynchronously) bytes from a given fspath

• rjson –

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

• rjson_async –

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

• rm –

  Remove files & directories in the style of the shell

• rm_gen –

  Remove files & directories in the style of the shell

• rmdir –

  Remove directory at given fspath

• rmfile –

  Remove a file at given fspath

• rstring –

  Load/Read a string given a fspath

• rstring_async –

  (ASYNC) Load/Read a string given a 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

• seconds2hrtime –

  Return hr-time Tuple[int, int] (seconds, nanoseconds)

• 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

• setenv –

  Export/Set an environment variable

• shebang –

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

• shell –

  Run a subprocess synchronously in current shell

• shplit –

  Typed alias for shlex.split

• source –

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

• stat –

  Return the os.stat_result object for a given fspath

• stat_async –

  Async version of os.lstat

• touch –

  Create an empty file given a fspath

• tree –

  Create a directory tree string given a directory path

• walk_gen –

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

• wbytes –

  Write/Save bytes to a fspath

• wbytes_async –

  (ASYNC) Write/Save bytes to a fspath

• wbytes_gen –

  Write/Save bytes to a fspath

• wbytes_gen_async –

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

• where –

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

• which –

  Return the result of shutil.which

• which_lru –

  Return the result of shutil.which and cache the results

• wjson –

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

• wjson_async –

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

• wstring –

  Save/Write a string to fspath

• wstring_async –

  (ASYNC) Save/Write a string to fspath

Done

Done(*args: Dict[_KT, _VT])

Done(*args: Dict[_KT, _VT], **kwargs: _VT)

Done(*args: Mapping[_KT, _VT])

Done(*args: Mapping[_KT, _VT], **kwargs: _VT)

Done(*args: Any, **kwargs: _VT)

Bases: JsonBaseModel

PRun => 'ProcessRun' for finished processes

Methods:

• asdict –

  Return the JsonObj object (and children) as a python dictionary

• check –

  Check returncode and stderr

• completed_process –

  Return subprocess.CompletedProcess object

• defaults_dict –

  Return a dictionary of non-required keys -> default value(s)

• dict –

  Alias for model_dump

• done_dict –

  Return Done object as typed-dict

• done_obj –

  Return Done object typed dict

• dot_items –

  Yield tuples of the form (dot-key, value)

• dot_items_list –

  Return list of tuples of the form (dot-key, value)

• dot_keys –

  Yield the JsonObj's dot-notation keys

• dot_keys_list –

  Return a list of the JsonObj's dot-notation friendly keys

• dot_keys_set –

  Return a set of the JsonObj's dot-notation friendly keys

• dot_lookup –

  Look up JsonObj keys using dot notation as a string

• eject –

  Eject to python-builtin dictionary object

• entries –

  Alias for items

• filter_false –

  Filter key-values where the value is false-y

• filter_none –

  Filter key-values where the value is None but not false-y

• from_dict –

  Return a JsonObj object from a dictionary of data

• from_dict_filtered –

  Create class from dict filtering keys not in (sub)class' fields

• from_json –

  Return a JsonObj object from a json string

• grep –

  Return lines in stdout that have

• has_required_fields –

  Return True/False if the (sub)class has any fields that are required

• is_default –

  Check if the object is equal to the default value for its fields

• items –

  Return an items view of the JsonObj object

• json –

  Alias for model_dumps

• json_parse –

  Return json parsed stdout

• json_parse_stderr –

  Return json parsed stderr

• json_parse_stdout –

  Return json parsed stdout

• keys –

  Return the keys view of the JsonObj object

• parse_json –

  Return json parsed stdout (alias bc I keep flip-flopping the fn name)

• recurse –

  Recursively convert all sub dictionaries to JsonObj objects

• stringify –

  Return JSON string of the JsonObj object (and children)

• sys_print –

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

• to_dict –

  Eject and return object as plain jane dictionary

• to_dict_filter_defaults –

  Eject object and filter key-values equal to (sub)class' default

• to_dict_filter_none –

  Eject object and filter key-values equal to (sub)class' default

• to_json –

  Return JSON string of the JsonObj object (and children)

• to_json_dict –

  Eject object and sub-objects to jsonbourne.JsonObj

• to_json_obj –

  Eject object and sub-objects to jsonbourne.JsonObj

• to_json_obj_filter_defaults –

  Eject to JsonObj and filter key-values equal to (sub)class' default

• to_json_obj_filter_none –

  Eject to JsonObj and filter key-values where the value is None

• validate_type –

  Validate and convert a value to a JsonObj object

• write_stderr –

  Write stderr as a string to a fspath

• write_stdout –

  Write stdout as a string to a fspath

asdict

asdict() -> Dict[_KT, Any]

Return the JsonObj object (and children) as a python dictionary

check

check(
    ok_code: Union[
        int, List[int], Tuple[int, ...], Set[int]
    ] = 0,
) -> None

Check returncode and stderr

Raises:

• DoneError –

  If return code is non-zero and stderr is not None

completed_process

completed_process() -> CompletedProcess[str]

Return subprocess.CompletedProcess object

defaults_dict classmethod

defaults_dict() -> Dict[str, Any]

Return a dictionary of non-required keys -> default value(s)

Returns:

• Dict[str, Any] –

  Dict[str, Any]: Dictionary of non-required keys -> default value

Examples:

>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm')
>>> t.to_dict_filter_defaults()
{}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{})
>>> t = Thing(a=123)
>>> t
Thing(a=123, b='herm')
>>> t.to_dict_filter_defaults()
{'a': 123}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{'a': 123})
>>> t.defaults_dict()
{'a': 1, 'b': 'herm'}

dict

dict(*args: Any, **kwargs: Any) -> Dict[str, Any]

Alias for model_dump

done_dict

done_dict() -> DoneDict

Return Done object as typed-dict

done_obj

done_obj() -> DoneObj

Return Done object typed dict

dot_items

dot_items() -> Iterator[Tuple[Tuple[str, ...], _VT]]

Yield tuples of the form (dot-key, value)

OG-version

def dot_items(self) -> Iterator[Tuple[str, Any]]: return ((dk, self.dot_lookup(dk)) for dk in self.dot_keys())

Readable-version

for k, value in self.items(): value = jsonify(value) if isinstance(value, JsonObj) or hasattr(value, 'dot_items'): yield from ((f"{k}.{dk}", dv) for dk, dv in value.dot_items()) else: yield k, value

dot_items_list

dot_items_list() -> List[Tuple[Tuple[str, ...], Any]]

Return list of tuples of the form (dot-key, value)

dot_keys

dot_keys() -> Iterable[Tuple[str, ...]]

Yield the JsonObj's dot-notation keys

Returns:

• Iterable[Tuple[str, ...]] –

  Iterable[str]: List of the dot-notation friendly keys

The Non-chain version (shown below) is very slightly slower than the itertools.chain version.

NON-CHAIN VERSION:

for k, value in self.items(): value = jsonify(value) if isinstance(value, JsonObj): yield from (f"{k}.{dk}" for dk in value.dot_keys()) else: yield k

dot_keys_list

dot_keys_list(
    sort_keys: bool = False,
) -> List[Tuple[str, ...]]

Return a list of the JsonObj's dot-notation friendly keys

Parameters:

• sort_keys

  (bool, default: False ) –

  Flag to have the dot-keys be returned sorted

Returns:

• List[Tuple[str, ...]] –

  List[str]: List of the dot-notation friendly keys

dot_keys_set

dot_keys_set() -> Set[Tuple[str, ...]]

Return a set of the JsonObj's dot-notation friendly keys

Returns:

• Set[Tuple[str, ...]] –

  Set[str]: List of the dot-notation friendly keys

dot_lookup

dot_lookup(
    key: Union[str, Tuple[str, ...], List[str]],
) -> Any

Look up JsonObj keys using dot notation as a string

Parameters:

• key

  (str) –

  dot-notation key to look up ('key1.key2.third_key')

Returns:

• Any –

  The result of the dot-notation key look up

Raises:

• KeyError –

  Raised if the dot-key is not in in the object

• ValueError –

  Raised if key is not a str/Tuple[str, ...]/List[str]

eject

eject() -> Dict[_KT, _VT]

Eject to python-builtin dictionary object

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

entries

entries() -> ItemsView[_KT, _VT]

Alias for items

filter_false

filter_false(recursive: bool = False) -> JsonObj[_VT]

Filter key-values where the value is false-y

Parameters:

• recursive

  (bool, default: False ) –

  Recurse into sub JsonObjs and dictionaries

Returns:

• JsonObj[_VT] –

  JsonObj that has been filtered

Examples:

>>> d = {
...     'falsey_dict': {},
...     'falsey_list': [],
...     'falsey_string': '',
...     'is_false': False,
...     'a': None,
...     'b': 2,
...     'c': {
...         'd': 'herm',
...         'e': None,
...         'falsey_dict': {},
...         'falsey_list': [],
...         'falsey_string': '',
...         'is_false': False,
...     },
...     }
...
>>> d = JsonObj(d)
>>> print(d)
JsonObj(**{
    'a': None,
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> print(d.filter_false())
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False}
})
>>> print(d.filter_false(recursive=True))
JsonObj(**{
    'b': 2, 'c': {'d': 'herm'}
})

filter_none

filter_none(recursive: bool = False) -> JsonObj[_VT]

Filter key-values where the value is None but not false-y

Parameters:

• recursive

  (bool, default: False ) –

  Recursively filter out None values

Returns:

• JsonObj[_VT] –

  JsonObj that has been filtered of None values

Examples:

>>> d = {
...     'falsey_dict': {},
...     'falsey_list': [],
...     'falsey_string': '',
...     'is_false': False,
...     'a': None,
...     'b': 2,
...     'c': {
...         'd': 'herm',
...         'e': None,
...         'falsey_dict': {},
...         'falsey_list': [],
...         'falsey_string': '',
...         'is_false': False,
...     },
...     }
...
>>> d = JsonObj(d)
>>> print(d)
JsonObj(**{
    'a': None,
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> print(d.filter_none())
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> from pprint import pprint
>>> print(d.filter_none(recursive=True))
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})

from_dict classmethod

from_dict(data: Dict[_KT, _VT]) -> JsonObj[_VT]

Return a JsonObj object from a dictionary of data

from_dict_filtered classmethod

from_dict_filtered(
    dictionary: Dict[str, Any],
) -> JsonBaseModelT

Create class from dict filtering keys not in (sub)class' fields

from_json classmethod

from_json(json_string: Union[bytes, str]) -> JsonObj[_VT]

Return a JsonObj object from a json string

Parameters:

• json_string

  (str) –

  JSON string to convert to a JsonObj

Returns:

• JsonObjT ( JsonObj[_VT] ) –

  JsonObj object for the given JSON string

grep

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

Return lines in stdout that have

Parameters:

• string

  (str) –

  String to search for

Returns:

• List[str] –

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

has_required_fields classmethod

has_required_fields() -> bool

Return True/False if the (sub)class has any fields that are required

Returns:

• bool ( bool ) –

  True if any fields for a (sub)class are required

is_default

is_default() -> bool

Check if the object is equal to the default value for its fields

Returns:

• bool –

  True if object is equal to the default value for all fields; False otherwise

Examples:

>>> class Thing(JsonBaseModel):
...    a: int = 1
...    b: str = 'b'
...
>>> t = Thing()
>>> t.is_default()
True
>>> t = Thing(a=2)
>>> t.is_default()
False

items

items() -> ItemsView[_KT, _VT]

Return an items view of the JsonObj object

json

json(*args: Any, **kwargs: Any) -> str

Alias for model_dumps

json_parse

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

Return json parsed stdout

json_parse_stderr

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

Return json parsed stderr

json_parse_stdout

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

Return json parsed stdout

keys

keys() -> KeysView[_KT]

Return the keys view of the JsonObj object

parse_json

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

Return json parsed stdout (alias bc I keep flip-flopping the fn name)

recurse

recurse() -> None

Recursively convert all sub dictionaries to JsonObj objects

stringify

stringify(
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    **kwargs: Any,
) -> str

Return JSON string of the JsonObj object (and children)

    Args:
        fmt (bool): If True, return a JSON string with newlines and indentation
        pretty (bool): If True, return a JSON string with newlines and indentation
        sort_keys (bool): Sort dictionary keys if True
        append_newline (bool): Append a newline '

' to JSON string if True default: default function hook for JSON serialization **kwargs (Any): additional kwargs to be passed down to jsonlib.dumps

    Returns:
        str: JSON string of the JsonObj object

sys_print

sys_print() -> None

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

to_dict

to_dict() -> Dict[str, Any]

Eject and return object as plain jane dictionary

to_dict_filter_defaults

to_dict_filter_defaults() -> Dict[str, Any]

Eject object and filter key-values equal to (sub)class' default

Examples:

>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm')
>>> t.to_dict_filter_defaults()
{}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{})
>>> t = Thing(a=123)
>>> t
Thing(a=123, b='herm')
>>> t.to_dict_filter_defaults()
{'a': 123}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{'a': 123})

to_dict_filter_none

to_dict_filter_none() -> Dict[str, Any]

Eject object and filter key-values equal to (sub)class' default

Examples:

>>> from typing import Optional
>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...     c: Optional[str] = None
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm', c=None)
>>> t.to_dict_filter_none()
{'a': 1, 'b': 'herm'}
>>> t.to_json_obj_filter_none()
JsonObj(**{'a': 1, 'b': 'herm'})

to_json

to_json(
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    **kwargs: Any,
) -> str

Return JSON string of the JsonObj object (and children)

    Args:
        fmt (bool): If True, return a JSON string with newlines and indentation
        pretty (bool): If True, return a JSON string with newlines and indentation
        sort_keys (bool): Sort dictionary keys if True
        append_newline (bool): Append a newline '

' to JSON string if True default: default function hook for JSON serialization **kwargs (Any): additional kwargs to be passed down to jsonlib.dumps

    Returns:
        str: JSON string of the JsonObj object

to_json_dict

to_json_dict() -> JsonObj[Any]

Eject object and sub-objects to jsonbourne.JsonObj

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

to_json_obj

to_json_obj() -> JsonObj[Any]

Eject object and sub-objects to jsonbourne.JsonObj

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

to_json_obj_filter_defaults

to_json_obj_filter_defaults() -> JsonObj[Any]

Eject to JsonObj and filter key-values equal to (sub)class' default

to_json_obj_filter_none

to_json_obj_filter_none() -> JsonObj[Any]

Eject to JsonObj and filter key-values where the value is None

validate_type classmethod

validate_type(val: Any) -> JsonObj[_VT]

Validate and convert a value to a JsonObj 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 ) –

  Flag to append to file or plain write to file

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 ) –

  Flag to append to file or plain write to file

DoneError

DoneError(done: Done)

Bases: SubprocessError

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

Attributes:

• cmd (str) –

  command that was run

• returncode (int) –

  exit status of the process

• stdout (str) –

  standard output (stdout) of the process

• stderr (str) –

  standard error (stderr) of the process

DoneObj

Bases: TypedDict

Todo

deprecate this in favor of DoneDict

Flag

Flag obj

Examples:

>>> Flag.__help
'--help'
>>> Flag._v
'-v'

FlagMeta

Bases: type

Meta class

Methods:

• attr2flag –

  Convert and return attr to string

attr2flag cached staticmethod

attr2flag(string: str) -> str

Convert and return attr to string

HrTime

HrTime(*args: Dict[_KT, _VT])

HrTime(*args: Dict[_KT, _VT], **kwargs: _VT)

HrTime(*args: Mapping[_KT, _VT])

HrTime(*args: Mapping[_KT, _VT], **kwargs: _VT)

HrTime(*args: Any, **kwargs: _VT)

Bases: JsonBaseModel

High resolution time

Methods:

• asdict –

  Return the JsonObj object (and children) as a python dictionary

• defaults_dict –

  Return a dictionary of non-required keys -> default value(s)

• dict –

  Alias for model_dump

• dot_items –

  Yield tuples of the form (dot-key, value)

• dot_items_list –

  Return list of tuples of the form (dot-key, value)

• dot_keys –

  Yield the JsonObj's dot-notation keys

• dot_keys_list –

  Return a list of the JsonObj's dot-notation friendly keys

• dot_keys_set –

  Return a set of the JsonObj's dot-notation friendly keys

• dot_lookup –

  Look up JsonObj keys using dot notation as a string

• eject –

  Eject to python-builtin dictionary object

• entries –

  Alias for items

• filter_false –

  Filter key-values where the value is false-y

• filter_none –

  Filter key-values where the value is None but not false-y

• from_dict –

  Return a JsonObj object from a dictionary of data

• from_dict_filtered –

  Create class from dict filtering keys not in (sub)class' fields

• from_json –

  Return a JsonObj object from a json string

• from_seconds –

  Return HrTime object from seconds

• has_required_fields –

  Return True/False if the (sub)class has any fields that are required

• is_default –

  Check if the object is equal to the default value for its fields

• items –

  Return an items view of the JsonObj object

• json –

  Alias for model_dumps

• keys –

  Return the keys view of the JsonObj object

• recurse –

  Recursively convert all sub dictionaries to JsonObj objects

• stringify –

  Return JSON string of the JsonObj object (and children)

• to_dict –

  Eject and return object as plain jane dictionary

• to_dict_filter_defaults –

  Eject object and filter key-values equal to (sub)class' default

• to_dict_filter_none –

  Eject object and filter key-values equal to (sub)class' default

• to_json –

  Return JSON string of the JsonObj object (and children)

• to_json_dict –

  Eject object and sub-objects to jsonbourne.JsonObj

• to_json_obj –

  Eject object and sub-objects to jsonbourne.JsonObj

• to_json_obj_filter_defaults –

  Eject to JsonObj and filter key-values equal to (sub)class' default

• to_json_obj_filter_none –

  Eject to JsonObj and filter key-values where the value is None

• validate_type –

  Validate and convert a value to a JsonObj object

asdict

asdict() -> Dict[_KT, Any]

Return the JsonObj object (and children) as a python dictionary

defaults_dict classmethod

defaults_dict() -> Dict[str, Any]

Return a dictionary of non-required keys -> default value(s)

Returns:

• Dict[str, Any] –

  Dict[str, Any]: Dictionary of non-required keys -> default value

Examples:

>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm')
>>> t.to_dict_filter_defaults()
{}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{})
>>> t = Thing(a=123)
>>> t
Thing(a=123, b='herm')
>>> t.to_dict_filter_defaults()
{'a': 123}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{'a': 123})
>>> t.defaults_dict()
{'a': 1, 'b': 'herm'}

dict

dict(*args: Any, **kwargs: Any) -> Dict[str, Any]

Alias for model_dump

dot_items

dot_items() -> Iterator[Tuple[Tuple[str, ...], _VT]]

Yield tuples of the form (dot-key, value)

OG-version

def dot_items(self) -> Iterator[Tuple[str, Any]]: return ((dk, self.dot_lookup(dk)) for dk in self.dot_keys())

Readable-version

for k, value in self.items(): value = jsonify(value) if isinstance(value, JsonObj) or hasattr(value, 'dot_items'): yield from ((f"{k}.{dk}", dv) for dk, dv in value.dot_items()) else: yield k, value

dot_items_list

dot_items_list() -> List[Tuple[Tuple[str, ...], Any]]

Return list of tuples of the form (dot-key, value)

dot_keys

dot_keys() -> Iterable[Tuple[str, ...]]

Yield the JsonObj's dot-notation keys

Returns:

• Iterable[Tuple[str, ...]] –

  Iterable[str]: List of the dot-notation friendly keys

The Non-chain version (shown below) is very slightly slower than the itertools.chain version.

NON-CHAIN VERSION:

for k, value in self.items(): value = jsonify(value) if isinstance(value, JsonObj): yield from (f"{k}.{dk}" for dk in value.dot_keys()) else: yield k

dot_keys_list

dot_keys_list(
    sort_keys: bool = False,
) -> List[Tuple[str, ...]]

Return a list of the JsonObj's dot-notation friendly keys

Parameters:

• sort_keys

  (bool, default: False ) –

  Flag to have the dot-keys be returned sorted

Returns:

• List[Tuple[str, ...]] –

  List[str]: List of the dot-notation friendly keys

dot_keys_set

dot_keys_set() -> Set[Tuple[str, ...]]

Return a set of the JsonObj's dot-notation friendly keys

Returns:

• Set[Tuple[str, ...]] –

  Set[str]: List of the dot-notation friendly keys

dot_lookup

dot_lookup(
    key: Union[str, Tuple[str, ...], List[str]],
) -> Any

Look up JsonObj keys using dot notation as a string

Parameters:

• key

  (str) –

  dot-notation key to look up ('key1.key2.third_key')

Returns:

• Any –

  The result of the dot-notation key look up

Raises:

• KeyError –

  Raised if the dot-key is not in in the object

• ValueError –

  Raised if key is not a str/Tuple[str, ...]/List[str]

eject

eject() -> Dict[_KT, _VT]

Eject to python-builtin dictionary object

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

entries

entries() -> ItemsView[_KT, _VT]

Alias for items

filter_false

filter_false(recursive: bool = False) -> JsonObj[_VT]

Filter key-values where the value is false-y

Parameters:

• recursive

  (bool, default: False ) –

  Recurse into sub JsonObjs and dictionaries

Returns:

• JsonObj[_VT] –

  JsonObj that has been filtered

Examples:

>>> d = {
...     'falsey_dict': {},
...     'falsey_list': [],
...     'falsey_string': '',
...     'is_false': False,
...     'a': None,
...     'b': 2,
...     'c': {
...         'd': 'herm',
...         'e': None,
...         'falsey_dict': {},
...         'falsey_list': [],
...         'falsey_string': '',
...         'is_false': False,
...     },
...     }
...
>>> d = JsonObj(d)
>>> print(d)
JsonObj(**{
    'a': None,
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> print(d.filter_false())
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False}
})
>>> print(d.filter_false(recursive=True))
JsonObj(**{
    'b': 2, 'c': {'d': 'herm'}
})

filter_none

filter_none(recursive: bool = False) -> JsonObj[_VT]

Filter key-values where the value is None but not false-y

Parameters:

• recursive

  (bool, default: False ) –

  Recursively filter out None values

Returns:

• JsonObj[_VT] –

  JsonObj that has been filtered of None values

Examples:

>>> d = {
...     'falsey_dict': {},
...     'falsey_list': [],
...     'falsey_string': '',
...     'is_false': False,
...     'a': None,
...     'b': 2,
...     'c': {
...         'd': 'herm',
...         'e': None,
...         'falsey_dict': {},
...         'falsey_list': [],
...         'falsey_string': '',
...         'is_false': False,
...     },
...     }
...
>>> d = JsonObj(d)
>>> print(d)
JsonObj(**{
    'a': None,
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> print(d.filter_none())
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'e': None,
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})
>>> from pprint import pprint
>>> print(d.filter_none(recursive=True))
JsonObj(**{
    'b': 2,
    'c': {'d': 'herm',
          'falsey_dict': {},
          'falsey_list': [],
          'falsey_string': '',
          'is_false': False},
    'falsey_dict': {},
    'falsey_list': [],
    'falsey_string': '',
    'is_false': False
})

from_dict classmethod

from_dict(data: Dict[_KT, _VT]) -> JsonObj[_VT]

Return a JsonObj object from a dictionary of data

from_dict_filtered classmethod

from_dict_filtered(
    dictionary: Dict[str, Any],
) -> JsonBaseModelT

Create class from dict filtering keys not in (sub)class' fields

from_json classmethod

from_json(json_string: Union[bytes, str]) -> JsonObj[_VT]

Return a JsonObj object from a json string

Parameters:

• json_string

  (str) –

  JSON string to convert to a JsonObj

Returns:

• JsonObjT ( JsonObj[_VT] ) –

  JsonObj object for the given JSON string

from_seconds classmethod

from_seconds(seconds: float) -> 'HrTime'

Return HrTime object from seconds

Parameters:

• seconds

  (float) –

  number of seconds

Returns:

• 'HrTime' –

  HrTime object

has_required_fields classmethod

has_required_fields() -> bool

Return True/False if the (sub)class has any fields that are required

Returns:

• bool ( bool ) –

  True if any fields for a (sub)class are required

is_default

is_default() -> bool

Check if the object is equal to the default value for its fields

Returns:

• bool –

  True if object is equal to the default value for all fields; False otherwise

Examples:

>>> class Thing(JsonBaseModel):
...    a: int = 1
...    b: str = 'b'
...
>>> t = Thing()
>>> t.is_default()
True
>>> t = Thing(a=2)
>>> t.is_default()
False

items

items() -> ItemsView[_KT, _VT]

Return an items view of the JsonObj object

json

json(*args: Any, **kwargs: Any) -> str

Alias for model_dumps

keys

keys() -> KeysView[_KT]

Return the keys view of the JsonObj object

recurse

recurse() -> None

Recursively convert all sub dictionaries to JsonObj objects

stringify

stringify(
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    **kwargs: Any,
) -> str

Return JSON string of the JsonObj object (and children)

    Args:
        fmt (bool): If True, return a JSON string with newlines and indentation
        pretty (bool): If True, return a JSON string with newlines and indentation
        sort_keys (bool): Sort dictionary keys if True
        append_newline (bool): Append a newline '

' to JSON string if True default: default function hook for JSON serialization **kwargs (Any): additional kwargs to be passed down to jsonlib.dumps

    Returns:
        str: JSON string of the JsonObj object

to_dict

to_dict() -> Dict[str, Any]

Eject and return object as plain jane dictionary

to_dict_filter_defaults

to_dict_filter_defaults() -> Dict[str, Any]

Eject object and filter key-values equal to (sub)class' default

Examples:

>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm')
>>> t.to_dict_filter_defaults()
{}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{})
>>> t = Thing(a=123)
>>> t
Thing(a=123, b='herm')
>>> t.to_dict_filter_defaults()
{'a': 123}
>>> t.to_json_obj_filter_defaults()
JsonObj(**{'a': 123})

to_dict_filter_none

to_dict_filter_none() -> Dict[str, Any]

Eject object and filter key-values equal to (sub)class' default

Examples:

>>> from typing import Optional
>>> class Thing(JsonBaseModel):
...     a: int = 1
...     b: str = "herm"
...     c: Optional[str] = None
...
>>> t = Thing()
>>> t
Thing(a=1, b='herm', c=None)
>>> t.to_dict_filter_none()
{'a': 1, 'b': 'herm'}
>>> t.to_json_obj_filter_none()
JsonObj(**{'a': 1, 'b': 'herm'})

to_json

to_json(
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    **kwargs: Any,
) -> str

Return JSON string of the JsonObj object (and children)

    Args:
        fmt (bool): If True, return a JSON string with newlines and indentation
        pretty (bool): If True, return a JSON string with newlines and indentation
        sort_keys (bool): Sort dictionary keys if True
        append_newline (bool): Append a newline '

' to JSON string if True default: default function hook for JSON serialization **kwargs (Any): additional kwargs to be passed down to jsonlib.dumps

    Returns:
        str: JSON string of the JsonObj object

to_json_dict

to_json_dict() -> JsonObj[Any]

Eject object and sub-objects to jsonbourne.JsonObj

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

to_json_obj

to_json_obj() -> JsonObj[Any]

Eject object and sub-objects to jsonbourne.JsonObj

Examples:

>>> d = JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> d
JsonObj(**{'uno': 'ONE', 'tres': 3, 'dos': 2})
>>> plain_ol_dict = d.eject()
>>> plain_ol_dict
{'uno': 'ONE', 'tres': 3, 'dos': 2}
>>> type(plain_ol_dict)
<class 'dict'>

to_json_obj_filter_defaults

to_json_obj_filter_defaults() -> JsonObj[Any]

Eject to JsonObj and filter key-values equal to (sub)class' default

to_json_obj_filter_none

to_json_obj_filter_none() -> JsonObj[Any]

Eject to JsonObj and filter key-values where the value is None

validate_type classmethod

validate_type(val: Any) -> JsonObj[_VT]

Validate and convert a value to a JsonObj object

HrTimeDict

Bases: TypedDict

High resolution time

HrTimeObj

Bases: TypedDict

Todo

deprecate this in favor of HrTimeDict

LIN

Bases: LIN

Linux (and Mac) shell commands/methods container

Methods:

• link_dir –

  Make a directory symlink

• link_dirs –

  Make multiple directory symlinks

• link_file –

  Make a file symlink

• link_files –

  Make multiple file symlinks

• rsync –

  Run an rsync subprocess

• rsync_args –

  Return args for rsync command on linux/mac

• 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

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

  (str, 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.

rsync staticmethod

rsync(
    src: str,
    dest: str,
    delete: bool = False,
    mkdirs: bool = False,
    dry_run: bool = False,
    exclude: Optional[IterableStr] = None,
    include: Optional[IterableStr] = 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

  (Optional[IterableStr], default: None ) –

  Strings/patterns to exclude

• include

  (Optional[IterableStr], 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

rsync_args staticmethod

rsync_args(
    src: str,
    dest: str,
    delete: bool = False,
    dry_run: bool = False,
    exclude: Optional[IterableStr] = None,
    include: Optional[IterableStr] = 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

  (Optional[IterableStr], default: None ) –

  Strings/patterns to exclude

• include

  (Optional[IterableStr], 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

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: IterableStr) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (IterableStr) –

  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: IterableStr) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (IterableStr) –

  Iterable of paths to links

Stdio

Bases: IntEnum

Standard-io enum object

WIN

Bases: WIN

Windows shell commands/methods container

Methods:

• link_dir –

  Make a directory symlink

• link_dirs –

  Make multiple directory symlinks

• link_file –

  Make a file symlink

• link_files –

  Make multiple file symlinks

• robocopy –

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

• robocopy_args –

  Return list of robocopy command args

• 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

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

robocopy staticmethod

robocopy(
    src: str,
    dest: str,
    *,
    mkdirs: bool = True,
    delete: bool = False,
    exclude_files: Optional[Iterable[str]] = None,
    exclude_dirs: Optional[Iterable[str]] = 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

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

  Strings/patterns with which to exclude files

• exclude_dirs

  (Optional[Iterable[str]], 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.

robocopy_args staticmethod

robocopy_args(
    src: str,
    dest: str,
    *,
    delete: bool = False,
    exclude_files: Optional[List[str]] = None,
    exclude_dirs: Optional[List[str]] = 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

  (Optional[List[str]], default: None ) –

  Strings/patterns with which to exclude files

• exclude_dirs

  (Optional[List[str]], 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.

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: IterableStr) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (IterableStr) –

  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: IterableStr) -> None

Unlink directory symlinks given the paths the links

Parameters:

• links

  (IterableStr) –

  Iterable of paths to links

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

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

decode_stdio_bytes

decode_stdio_bytes(
    stdio_bytes: Union[str, bytes], lf: bool = True
) -> str

Return Stdio bytes from stdout/stderr as a string

Args:
    stdio_bytes (bytes): STDOUT/STDERR bytes
    lf (bool): Replace `

line endings with `

Returns:
    str: decoded stdio bytes

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

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

dirpath_gen

dirpath_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = False,
    topdown: bool = True,
    onerror: Optional[Callable[[OSError], Any]] = 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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)

do

do(
    *popenargs: PopenArgs,
    args: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    extenv: bool = True,
    cwd: Optional[FsPath] = None,
    shell: bool = False,
    check: bool = False,
    tee: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    timeout: Optional[Union[float, int]] = None,
    ok_code: Union[
        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

  (Optional[PopenArgs], default: None ) –

  Args as strings for the subprocess

• env

  (Optional[Dict[str, str]], 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

  (Optional[FsPath], 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

  (Optional[int], default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (Union[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

do_async async

do_async(
    *popenargs: PopenArgs,
    args: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    extenv: bool = True,
    cwd: Optional[str] = None,
    shell: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    check: bool = False,
    timeout: Optional[float] = None,
    ok_code: Union[
        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

  (Optional[PopenArgs], default: None ) –

  Args as strings for the subprocess

• check

  (bool, default: False ) –

  Check the result returncode

• env

  (Optional[Dict[str, str]], default: None ) –

  Environment variables as a dictionary (Default value = None)

• extenv

  (bool, default: True ) –

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

• cwd

  (Optional[str], 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

  (Optional[int], default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (Union[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

do_asyncify async

do_asyncify(
    *popenargs: PopenArgs,
    args: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    extenv: bool = True,
    cwd: Optional[str] = None,
    shell: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    check: bool = False,
    timeout: Optional[Union[float, int]] = None,
    ok_code: Union[
        int, List[int], Tuple[int, ...], Set[int]
    ] = 0,
    dryrun: bool = False,
) -> Done

Run a subprocess asynchronously using asyncified version of do

doa async

doa(
    *popenargs: PopenArgs,
    args: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    extenv: bool = True,
    cwd: Optional[str] = None,
    shell: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    check: bool = False,
    timeout: Optional[float] = None,
    ok_code: Union[
        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

  (Optional[PopenArgs], default: None ) –

  Args as strings for the subprocess

• check

  (bool, default: False ) –

  Check the result returncode

• env

  (Optional[Dict[str, str]], default: None ) –

  Environment variables as a dictionary (Default value = None)

• cwd

  (Optional[str], 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

  (Optional[int], default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (Union[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

echo

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

Print/echo function

Args:
    *objects: Item(s) to print/echo
    sep: Separator to print with
    end: End of print suffix; defaults to `

` file: File like object to write to if not stdout flush: Flush the file after writing

Examples:
    >>> echo("shellfish")
    shellfish

exists

exists(fspath: FsPath) -> bool

Return True if the given path exists; False otherwise

export

export(
    key: str, val: Optional[str] = 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

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 wstring
>>> wstring(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: Optional[Callable[[OSError], Any]] = 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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

flatten_args

flatten_args(*args: Union[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']

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

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_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

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

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_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

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

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

path_gen

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

Yield all filepaths as pathlib.Path objects beneath a dirpath

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

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

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)

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

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\'"\'"\'\''

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\'"\'"\'\''

rbytes

rbytes(filepath: FsPath) -> bytes

Load/Read bytes from a fspath

Parameters:

• filepath

  (FsPath) –

  fspath read as bytes

Returns:

• bytes –

  bytes from the fspath

Examples:

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

rbytes_async async

rbytes_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 rbytes_async, wbytes_async
>>> from asyncio import run as aiorun
>>> fspath = "rbytes_async.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> aiorun(wbytes_async(fspath, bites_to_save))
20
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> aiorun(rbytes_async(fspath))
b'These are some bytes'
>>> import os; os.remove(fspath)

rbytes_gen

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

Yield bytes from a given fspath

rbytes_gen_async async

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

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[Union[bytes, str]] –

  bytes from AsyncIterable[bytes] of the file bytes

Examples:

>>> from os import remove
>>> from asyncio import run
>>> from shellfish.fs._async import wbytes_gen_async, rbytes_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(wbytes_gen_async(fspath, bites_to_save))
35
>>> async def read():
...     async for b in rbytes_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(wbytes_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(wbytes_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)

rjson

rjson(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 rjson, wjson

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_dict.doctest.json"
>>> wjson(fspath, data)
19
>>> rjson(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"
>>> wjson(fspath, data)
25
>>> rjson(fspath)
[['a', 1], ['b', 2], ['c', 3]]
>>> os.remove(fspath)

rjson_async async

rjson_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 rjson_async, wjson_async

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_async_dict.doctest.json"
>>> run(wjson_async(fspath, data))
19
>>> run(rjson_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(wjson_async(fspath, data))
25
>>> run(rjson_async(fspath))
[['a', 1], ['b', 2], ['c', 3]]

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

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

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

rstring

rstring(
    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 rstring, wstring
>>> fspath = "lstring.doctest.txt"
>>> sstring(fspath, r'Check out this string')
21
>>> lstring(fspath)
'Check out this string'
>>> import os; os.remove(fspath)

rstring_async async

rstring_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

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

seconds2hrtime

seconds2hrtime(
    seconds: Union[float, int],
) -> Tuple[int, int]

Return hr-time Tuple[int, int] (seconds, nanoseconds)

Parameters:

• seconds

  (float) –

  number of seconds

Returns:

• Tuple[int, int] –

  Tuple[int, int]: (seconds, nanoseconds)

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

setenv

setenv(
    key: str, val: Optional[str] = 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

shebang

shebang(fspath: FsPath) -> Union[None, str]

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

Parameters:

• fspath

  (FsPath) –

  Path to file that might have a shebang

Returns:

• Union[None, str] –

  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)

shell

shell(
    *popenargs: PopenArgs,
    args: Optional[PopenArgs] = None,
    env: Optional[Dict[str, str]] = None,
    shell: bool = True,
    extenv: bool = True,
    cwd: Optional[FsPath] = None,
    check: bool = False,
    verbose: bool = False,
    input: STDIN = None,
    timeout: Optional[Union[float, int]] = None,
    ok_code: Union[
        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

  (Optional[PopenArgs], default: None ) –

  Args as strings for the subprocess

• env

  (Optional[Dict[str, str]], default: None ) –

  Environment variables as a dictionary (Default value = None)

• shell

  (bool, default: True ) –

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

• extenv

  (bool, default: True ) –

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

• cwd

  (Optional[FsPath], 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

  (Optional[int], default: None ) –

  Timeout in seconds for the process if not None

• ok_code

  (Union[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

shplit

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

Typed alias for shlex.split

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

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.lstat

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

tree

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

Create a directory tree string given a directory path

Parameters:

• dirpath

  (FsPath) –

  Directory string to make tree for

• filterfn

  (Optional[Callable[[str], bool]], 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)

walk_gen

walk_gen(
    dirpath: FsPath = ".",
    *,
    abspath: bool = True,
    topdown: bool = True,
    onerror: Optional[Callable[[OSError], Any]] = 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

  (Optional[Callable[[OSError], Any]], 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)

wbytes

wbytes(
    filepath: FsPath,
    bites: bytes,
    *,
    append: bool = False,
    chmod: Optional[int] = 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 rbytes, wbytes
>>> fspath = "wbytes.doctest.txt"
>>> bites_to_save = b"These are some bytes"
>>> bites_to_save  # they are bytes!
b'These are some bytes'
>>> wbytes(fspath, bites_to_save)
20
>>> rbytes(fspath)
b'These are some bytes'
>>> import os; os.remove(fspath)

wbytes_async async

wbytes_async(
    filepath: FsPath,
    bites: bytes,
    *,
    append: bool = False,
    chmod: Optional[int] = 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

  (Optional[int], default: None ) –

  chmod the fspath to this mode after writing

Returns:

• int –

  None

Examples:

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

wbytes_gen

wbytes_gen(
    filepath: FsPath,
    bytes_gen: Iterable[bytes],
    append: bool = False,
    chmod: Optional[int] = 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 rbytes, wbytes
>>> 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!')
>>> wbytes_gen(fspath, (b for b in bites_to_save))
35
>>> rbytes(fspath)
b'These are some bytes... more bytes!'
>>> import os; os.remove(fspath)

wbytes_gen_async async

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

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

Parameters:

• filepath

  (FsPath) –

  fspath to write to

• bytes_gen

  (Union[Iterable[bytes], AsyncIterable[bytes]]) –

  AsyncIterable/Iterator of bytes to write

• append

  (bool, default: False ) –

  Append to the fspath if True; otherwise overwrite

• chmod

  (Optional[int], 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 wbytes_gen_async, rbytes_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(wbytes_gen_async(fspath, bites_to_save))
35
>>> async def read():
...     async for b in rbytes_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(wbytes_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(wbytes_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)

where

where(
    cmd: str, path: Optional[str] = None
) -> Optional[str]

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:

• Optional[str] –

  Optional[str]: path to command/exe

which

which(
    cmd: str, path: Optional[str] = None
) -> Optional[str]

Return the result of shutil.which

Parameters:

• cmd

  (str) –

  Command/exe to find path of

• path

  (str, default: None ) –

  System path to use

Returns:

• Optional[str] –

  Optional[str]: path to command/exe

which_lru cached

which_lru(
    cmd: str, path: Optional[str] = None
) -> Optional[str]

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:

• Optional[str] –

  Optional[str]: path to command/exe

wjson

wjson(
    filepath: FsPath,
    data: Any,
    *,
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    chmod: Optional[int] = 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

  (Optional[Callable[[Any], Any]], 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 rjson, wjson

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "rjson_dict.doctest.json"
>>> wjson(fspath, data)
19
>>> rjson(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"
>>> wjson(fspath, data)
25
>>> rjson(fspath)
[['a', 1], ['b', 2], ['c', 3]]
>>> os.remove(fspath)

wjson_async async

wjson_async(
    filepath: FsPath,
    data: Any,
    *,
    fmt: bool = False,
    pretty: bool = False,
    sort_keys: bool = False,
    append_newline: bool = False,
    default: Optional[Callable[[Any], Any]] = None,
    append: bool = False,
    chmod: Optional[int] = 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

  (Optional[Callable[[Any], Any]], 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 rjson_async, wjson_async

Dictionaries:

>>> data = {'a': 1, 'b': 2, 'c': 3}
>>> fspath = "wjson_async_dict.doctest.json"
>>> run(wjson_async(fspath, data))
19
>>> run(rjson_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(wjson_async(fspath, data))
25
>>> run(rjson_async(fspath))
[['a', 1], ['b', 2], ['c', 3]]

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

wstring

wstring(
    filepath: FsPath,
    string: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    chmod: Optional[int] = 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 rstring, wstring
>>> fspath = "sstring.doctest.txt"
>>> wstring(fspath, r'Check out this string')
21
>>> rstring(fspath)
'Check out this string'
>>> import os; os.remove(fspath)

wstring_async async

wstring_async(
    filepath: FsPath,
    string: str,
    *,
    encoding: str = "utf-8",
    append: bool = False,
    chmod: Optional[int] = 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