Skip to content

promises ⚓︎

shellfish.fs.promises

Functions:

  • dir_exists

    Return True if the directory exists; False otherwise

  • file_exists

    Return True if the file exists; False otherwise

  • is_dir

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

  • is_file

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

  • is_link

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

  • isdir

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

  • isfile

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

  • islink

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

  • lstat

    Async version of os.lstat

  • read_bytes

    (ASYNC) Load/Read bytes from a fspath

  • read_bytes_gen

    Yield (asynchronously) bytes from a given fspath

  • read_str

    (ASYNC) Load/Read a string given a fspath

  • stat

    Async version of os.stat

  • write_bytes

    (ASYNC) Write/Save bytes to a fspath

  • write_bytes_gen

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

  • write_str

    (ASYNC) Save/Write a string to fspath

dir_exists async ⚓︎

dir_exists(fspath: FsPath) -> bool

Return True if the directory exists; False otherwise

file_exists async ⚓︎

file_exists(fspath: FsPath) -> bool

Return True if the file exists; False otherwise

is_dir async ⚓︎

is_dir(fspath: FsPath) -> bool

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

is_file async ⚓︎

is_file(fspath: FsPath) -> bool

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

is_link(fspath: FsPath) -> bool

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

isdir async ⚓︎

isdir(fspath: FsPath) -> bool

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

isfile async ⚓︎

isfile(fspath: FsPath) -> bool

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

islink(fspath: FsPath) -> bool

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

lstat async ⚓︎

lstat(fspath: FsPath) -> stat_result

Async version of os.lstat

read_bytes async ⚓︎

read_bytes(filepath: FsPath) -> bytes

(ASYNC) Load/Read bytes from a fspath

Parameters:

  • filepath ⚓︎

    (FsPath) –

    fspath read as bytes

Returns:

  • bytes

    bytes from the fspath

Examples:

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

read_bytes_gen async ⚓︎

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

Yield (asynchronously) bytes from a given fspath

Parameters:

  • filepath ⚓︎

    (FsPath) –

    fspath to read from

  • blocksize ⚓︎

    (int, default: 65536 ) –

    size of the block to read

Yields:

Examples:

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

read_str async ⚓︎

read_str(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

stat async ⚓︎

stat(fspath: FsPath) -> stat_result

Async version of os.stat

write_bytes async ⚓︎

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

(ASYNC) Write/Save bytes to a fspath

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

Parameters:

  • append ⚓︎

    (bool, default: False ) –

    Append to the fspath if True; otherwise overwrite

  • filepath ⚓︎

    (FsPath) –

    fspath to write to

  • bites ⚓︎

    (bytes) –

    Bytes to be written

  • chmod ⚓︎

    (int | None, default: None ) –

    chmod the fspath to this mode after writing

Returns:

Examples:

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

write_bytes_gen async ⚓︎

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

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

Parameters:

  • filepath ⚓︎

    (FsPath) –

    fspath to write to

  • bytes_gen ⚓︎

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

    AsyncIterable/Iterator of bytes to write

  • append ⚓︎

    (bool, default: False ) –

    Append to the fspath if True; otherwise overwrite

  • chmod ⚓︎

    (int | None, default: None ) –

    chmod the fspath if not None

Returns:

  • int ( int ) –

    number of bytes written

Examples:

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

write_str async ⚓︎

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

(ASYNC) Save/Write a string to fspath

Parameters:

  • filepath ⚓︎

    (FsPath) –

    fspath to write to

  • string ⚓︎

    (str) –

    string to be written

  • encoding ⚓︎

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

    File encoding (Default='utf-8')

  • append ⚓︎

    (bool, default: False ) –

    Append to the fspath if True; default is False

  • chmod ⚓︎

    (Optional[int], default: None ) –

    chmod the fspath if not None

Returns:

  • int ( int ) –

    number of bytes written