cl_utils package

Submodules

cl_utils.core module

class cl_utils.core.EnvType

Bases: type

async cl_utils.core.async_stream_command(*command: str, stdout_streamer: ~typing.Callable[[~asyncio.streams.StreamReader], ~typing.Awaitable[None]] = <function subprocess_streamer.<locals>._output_stream>, stderr_streamer: ~typing.Callable[[~asyncio.streams.StreamReader], ~typing.Awaitable[None]] = <function subprocess_streamer.<locals>._output_stream>, **kwargs: ~typing.Any) → int

Execute a command in the background and stream its output.

Parameters:

  • stdout_streamer (SubprocessStreamerType, optional) – , defaults to subprocess_streamer( cast(io.IOBase, sys.stdout.buffer) )

  • stderr_streamer (SubprocessStreamerType, optional) – _description_, defaults to subprocess_streamer( cast(io.IOBase, sys.stderr.buffer) )

Returns:

_description_

Return type:

int

cl_utils.core.chdir(path: str | pathlib.Path) → None
class cl_utils.core.env

Bases: object

cl_utils.core.flush_print(*args: Any, **kwargs: Any) → None
cl_utils.core.isdir(f: str | pathlib.Path) → bool
cl_utils.core.isfile(f: str | pathlib.Path) → bool
cl_utils.core.load_project_dotenv(dotenv_path=None)
cl_utils.core.mkdir(path: str | pathlib.Path, *args, **kwargs) → None
class cl_utils.core.pushd(path: str | pathlib.Path, verbose: bool = False)

Bases: object

This is a context manager for pushd. It will remember where you were and take you back there when you exit the context manager.

Parameters:

path (str | Path) – The path to change to upon entering the context manager.

cl_utils.core.stream_command(*args: str, **kwargs: Any) → int
cl_utils.core.subprocess_streamer(*output_stream: ~io.IOBase, filter: ~typing.Callable[[bytes], ~typing.Union[bytes, ~typing.Sequence[bytes]]] = <function <lambda>>) → Callable[[StreamReader], Awaitable[None]]

This is for use with stream_command: provide it with output(s) and (optionally) a filter, and it will stream data from from the subprocess’s stdout/stderr to the `output_stream`(s) given below. :param filter: _description_, defaults to lambdaline:line :type filter: _type_, optional :return: _description_ :rtype: SubprocessStreamerType

cl_utils.core.xqt(*cmds: list[str], check: bool = True, shell: bool = True, **kwargs: Any) → subprocess.CompletedProcess | list[subprocess.CompletedProcess]

Module contents