cmdi 1.1.2

No project description provided

cmdi - Command Interface

Description

A decorator @command that applies a special interface called the Command Interface to its decorated function. Initially written for the buildlib.

The Command Interface allows you to control the exectuion of a function via the Command Interface:

• It allows you to save/redirect/mute output streams (stdout/stderr) for its decorated function. This works on file descriptor level, thus it's possible to redirect output of subproesses and C code.
• It allows you to catch exceptions for its decorated function and return them with the CmdResult(), including return codes, error messages and colored status messages.
• It allows you to print status messages and summaries for a command at runtime.
• And more...

A function that is decorated with @command can receive a set of sepcial keyword arguments (_verbose=..., _stdout=..., _stderr=..., catch_err=..., etc.) and it returns a CmdResult() object.

Requirements

Python >= 3.7

Install

pip install cmdi

Usage

The @command decorator

Use the @command decorator to apply the command interface to a function.

from cmdi import command

@command
def foo_cmd(x, **cmdargs) -> CmdResult:
    print(x)
    return x * 2

Now you can use foo_cmd as a command:

result = foo_cmd(10)

Which will print the following output (in color):

Cmd: foo_cmd
------------
10
foo_cmd: Ok

and return a CmdResult object:

CmdResult(
    val=20,
    code=0,
    name='foo_cmd',
    status='Ok',
    color=0,
    stdout=None,
    stderr=None
)

Command Function Arguments

You can define the behaviour of a command function using a set of special keyword argumnets that are applied to the decorated function.

In this example we redirect the output of foo_cmd to a custom writer and catch exceptions, the output and information of the exception are then returned with the CmdResult() object:

from cmdi import CmdResult, Pipe


result = foo_cmd(10, _stdout=Pipe(), _catch_err=True)

isinstance(result, CmdResult) # True

print(result.stdout) # prints caught output.

More about special keyword arguments can be found in the API documentation below.

Customizing the Result of a command function

A command always returns a CmdResult object, for which the @command wrapper function automatically guesses the values, which is good enough in many situations. But sometimes you need fine grained control over the output, e.g. to create function specific return codes:

@command
def foo_cmd(x: str, **cmdargs) -> CmdResult:

    print(x)
    somestr = "foo" + x

    if x == "bar":
        code = 0
    else:
        code = 42

    # Return a customized Command Result:

    return CmdResult(
        val=somestr,
        code=code,
    )

Note: In the example above, we return a customized CmdResult for which we only customize the fields val and code. You can customize every field of the CmdResult object (optionally). The fields you leave out are set automatically.

Command Interface Wrappers

Sometimes you want to use the Command Interface for an existing function, without touching the function definition. You can do so by creating a Command Interface Wrapper:

from cmdi import command, strip_cmdargs, CmdResult

# This function wraps the Command Interface around an existing function:

@command
def foo_cmd(x, **cmdargs) -> CmdResult:
    return foo(**strip_cmdargs(loclas()))


# The original function that is being wrapped:

def foo(x) -> int:
    print(x)
    return x * 2

Command Interface Wrappers and subprocess return codes.

If you need to create a Command Interface Wrapper for an existing function that runs a subprocess and your command depends on the returncode of that, you can use the subprocess.CalledProcessError exception to compose something. E.g.:

import subprocess as sp
from cmdi import command, CmdResult, Status

@command
def foo_cmd(x, **cmdargs) -> CmdResult:

    try:
        return foo(**strip_cmdargs(locals()))

    except sp.CalledProcessError as e:

        if e.returncode == 13:
            return CmdResult(
                code=e.returncode,
                status=Status.ok,
            )
        elif e.returncode == 42:
            return CmdResult(
                code=e.returncode,
                status=Status.skip,
            )
        else:
            raise sp.CalledProcessError(e.returncode, e.args)


def foo(x) -> int:
    return sp.run(["my_arg"], check=True, ...)

API

decorator @command

This decorator allows you to apply the command interface to a function.

A function decorated with @command can take the following keyword arguments:

_verbose: bool = True

Enable/Disable printing of header/status message during runtime.

Example:

result = my_command_func("some_arg", _verbose=False)

_color: bool = True

Enable/Disable color for header/status message.

Example:

result = my_command_func("some_arg", _color=False)

_stdout: Optional[Pipe] = None

Redirect stdout of the child function.

Example:

from cmdi import Pipe

pipe = Pipe(text=False, tty=True, ...) # See Pipe doc for arguments...

result = my_command_func('foo', _stdout=pipe)

print(result.stdout) # Prints the caught ouput.

_stderr: Union[Optional[Pipe], STDOUT] = None

Redirect stderr of the child function.

Example:

from cmdi import Pipe

pipe = Pipe(text=False, tty=True, ...) # See Pipe doc for arguments...

result = my_command_func('foo', _stderr=pipe))

print(result.stderr) # Prints the caught ouput.

If you want to redirect stderr to stdout, you can use this:

from cmdi import STDOUT

result = my_command_func('foo', _stderr=STDOUT))

_catch_err: bool = True

Catch errors from child function.

This will let the runtime continue, even if a child function throws an exception. If an exception occurs the CmdResult object will provide information about the error at result.stderr, result.code and result.status. The status message will appear in red.

Example: from cmdi import Pipe

r = my_command_func("some_arg", _catch_err=True, _stderr=Pipe())

r.status # Error
r.code # 1
r.stdout # The stderr output from the function call.

dataclass CmdResult()

The command result object.

A function decorated with @command returns a CmdResult object:

@dataclass
class CmdResult:
    val: Optional[Any]
    code: Optional[int]
    name: Optional[str]
    status: Optional[str]
    color: Optional[int]
    out: Optional[TextIO]
    err: Optional[TextIO]

dataclass Pipe()

Use this type to configure stdout/stderr for a command call.

Parameters

• save: bool = True - Save the function ouput if True.
• text: bool = True - Save function output as text if True else save as bytes.
• dup: bool = False - Redirect ouput at file discriptor level if True. This allows you to redirect output of subprocesses and C code.
• tty: bool = False - Keep ANSI sequences for saved output if True, else strip ANSI sequences.
• mute: bool = False - Mute output of function call in terminal if True. NOTE: You can still save and return the ouput if this is enabled.

Example:

from cmdi import CmdResult, Pipe

out_pipe = Pipe(text=False, dup=True, mute=True)
err_pipe = Pipe(text=False, dup=True, mute=False)

result = foo_cmd(10, _stdout=out_pipe, _stderr=err_pipe, _catch_err=True)

print(result.stdout) # prints caught output.
print(result.stderr) # prints caught output.

Redirect ouput of functions that run subprocesses or C code.

If you want to redirect the ouput of a function that runs a subprocess or calls C code, you have to use a Pipe with the argument dup=True. This will catch the output of stdout/stderr at a lower level (by duping file descriptors):

import subprocess
from cmdi import command, Pipe

@command
def foo(x, **cmdargs) -> CmdResult:
    subprocess.run("my_script")

# Catch stdout of the function via low level redirect:
foo(_stdout=Pipe(dup=True))

function strip_cmdargs(locals_)

Parameters

• locals_: Dict[str, Any]

Returns

• Dict[str, Any]

Remove cmdargs from dictionary. This function is useful for Command Interface Wrappers.

Example usage:

def foo(x):
    # Do a lot of stuff
    return x * 2

@command
def foo_cmd(x, **cmdargs):
    return foo(strip_cmdargs(locals()))

function print_title(result, color, file)

Parameter

• result: CmdResult
• color: bool = True
• file: Optional[IO[str]] = None

Returns

• None

Print the title for a command result

Example usage:

result = my_cmd('foo')

print_title(result)

Output:

Cmd: my_cmd
-----------

function print_status(result, color, file)

Parameter

• result: CmdResult
• color: bool = True
• file: Optional[IO[str]] = None

Returns

• None

Print the status of a command result.

Example usage:

result = my_cmd('foo')

print_status(result)

Output:

my_cmd: Ok

function print_result(result, color, file)

Parameter

• result: CmdResult
• color: bool = True
• file: Optional[IO[str]] = None

Returns

• None

Print out the CmdResult object.

Example usage:

result = my_cmd('foo')

print_result(result)

Output:

Cmd: my_cmd
-----------
Stdout:
Runtime output of my_cmd...
Stderr:
Some err
foo_cmd3: Ok

function print_summary(results, color, headline, file)

Parameter

• results: CmdResult
• color: bool = True
• headline: bool = True
• file: Optional[IO[str]] = None

Returns

• None

from cmdi import print_summary

results = []

results.append(my_foo_cmd())
results.append(my_bar_cmd())
results.append(my_baz_cmd())

print_summary(results)

Output:

Cmd: my_foo_cmd
---------------
stdout of foo function...
my_foo_cmd: Ok

Cmd: my_bar_cmd
---------------
stdout of bar function...
my_bar_cmd: Ok

Cmd: my_baz_cmd
---------------
stdout of baz function...
my_baz_cmd: Ok

function read_popen_pipes(p, interval)

Parameter

• p: subprocess.Popen
• interval: int = 10 - The interval which the output streams are read and written with.

Returns

• Iterator[Tuple[str, str]]

This returns an iterator which returns Popen pipes line by line for both stdout and stderr in realtime.

Example usage:

from cmdi import POPEN_DEFAULTS, read_popen_pipes

p = subprocess.Popen(mycmd, **POPEN_DEFAULTS)

for out_line, err_line in read_popen_pipes:
    print(out_line, end='')
    print(err_line, end='')

code = p.poll()

function resolve_popen()

Parameter

• p: subprocess.Popen
• save_stdout: bool = False - If set to True, the function output is returned.
• save_stderr: bool = False - If set to True, the function error output is returned.
• mute_stdout: bool = False - If set to True, the function output won't be written to sys.stdout.
• mute_stderr: bool = False - If set to True, the function error output won't be written to sys.stderr.
• catch: List[int] = [] - Do not raise error for returncodes defined here. You can use ["*"] to prevent exceptions for all returncodes.
• interval: int = 10 - The interval which the output streams are read and written with.

Returns

• subprocess.CompletedProcess

Handle running Popen process.

Example usage:

from cmdi import POPEN_DEFAULTS, resolve_popen

p = subprocess.Popen(mycmd, **POPEN_DEFAULTS)

# Do stuff with p.

# Get CompletedProcess object.
cp = resolve_popen(p, save_stdout=True, mute_stdout=True)

function run_subprocess()

Parameter

• p: subprocess.Popen
• save_stdout: bool = False - If set to True, the function output is returned.
• save_stderr: bool = False - If set to True, the function error output is returned.
• mute_stdout: bool = False - If set to True, the function output won't be written to sys.stdout.
• mute_stderr: bool = False - If set to True, the function error output won't be written to sys.stderr.
• catch: List[int] = [] - Do not raise error for returncodes defined here. You can use ["*"] to prevent exceptions for all returncodes.
• interval: int = 10 - The interval which the output streams are read and written with.
• cwd: Optional[str] = None - See subprocess Popen.
• shell: bool = False - See subprocess Popen.

Returns

• subprocess.CompletedProcess

Run Popen process with convenient options.

Example usage:

from cmdi import POPEN_DEFAULTS, run_subprocess

cp = run_subprocess(mycmd, **POPEN_DEFAULTS)