Python API

Python API#

This page describes some of the Python utility modules for writing your own custom autopilot app or a Custom Python Scripts with PAPSR.

yaku.autopilot_utils.cli_base#

A utility module for writing Python apps.

Python App Template#

This module provides some utility functions for easily creating Python apps without having to write a lot of boilerplate code for command line argument parsing, logging setup, result handling, etc.

There are example applications available in the tests/app_* folders.

Simple App#

A very simple (fetcher) app looks like this:

# Module yaku.app_single_command.cli
import click
from yaku.autopilot_utils.cli_base import make_autopilot_app, read_version_from_package
from loguru import logger


class CLI:
    click_name = "app_single_command"
    click_help_text = "Simple demo program for test purposes with just a simple command."

    click_setup = [
        click.option("--fail", is_flag=True),
    ]

    @staticmethod
    def click_command(fail: bool):
        logger.info("Inside click_command")
        if fail:
            raise Exception("Failing as requested...")
        logger.info("Should be doing something useful here!")


cli = make_autopilot_app(
    provider=CLI,
    version_callback=read_version_from_package(__package__)
)

if __name__ == "__main__":
    cli()

The BUILD file looks like this:

pex_binary(
    name="app_single_command",
    entry_point="yaku.app_single_command.cli",  # the name of the module above
)

All the setup of the app happens in the make_autopilot_app() function. It requires two arguments:

a provider which can be a class or a module.
a package_with_version_file, which must be a package name inside which a _version.txt file is located which contains the app version number.

The provider class (could also be a module) usually has three attributes:

A click_name string which contains the app name (should be the same as the name of the pex binary in the BUILD file).
A click_setup (which can be omitted or an empty list) which contains a list of click.option or click.argument decorators (but omit the @). Use those decorators to add arguments or options to your app.
A click_command function (which can be missing if there are subcommands) which must accept the arguments according to the click_setup list and which contains the app logic.

For more complex use-cases, other attributes are also possible:

click_evaluator_callback needs to be defined if results are collected during the execution of the click_command function.
click_subcommands can be used to define subcommands for a main command, e.g. when having an Excel evaluator which can be called like excel-evaluate cell --location=A1 --equals="yes" or like excel-evaluate column --column-index=F --values="yes|no".

Click argument validation#

When using the validation callback=... argument for click.option, you need to adhere to the way how click is handling exceptions.

Instead of raising AutopilotConfigurationError or similar exceptions from yaku.autopilot_utils.errors, use click.BadParameter or click.UsageError instead.

For details, see ClickUsageErrorHandlerDecorator below.

Complex apps/evaluators#

There are multiple app configurations possible:

Simple app with a single command, not evaluation (see app_single_command/).
Simple evaluator, called from a single command (see app_single_evaluator/).
Complex app with multiple subcommands, but no evaluation (see app_multi_command/).
Complex evaluator with multiple independent evaluators which can not be chained (see app_multi_evaluator/). This means that you can only call one of the subcommands at a time. This also means that each evaluator needs to compute its own evaluation result (that’s why the subcommand providers in this example have all a custom click_evaluator_callback function)
Complex evaluator with chainable subcommand evaluators. (see app_chained_multi_evaluator/). This means that you can call all the subcommands on the command line in a row. But this also means that the results generated by these subcommands need to be evaluated by the main command provider, and not by each of the subcommand providers. So only the main provider needs a click_evaluator_callback function.

Module contents#

yaku.autopilot_utils.cli_base.make_autopilot_app(provider, *, version_callback, allow_chaining=True)#

Create a click application from a special type of class/module.

Parameters:

provider (a class or module which provides necessary attributes, e.g.,) – click_setup, click_command, click_help_text, …
version_callback (a function which returns the version number of) – the app. See also: read_version_from_package().
allow_chaining (Flag to enable or disable the possibility to run) – multiple evaluators as chained subcommands, so that the main CLI provider does an overall evaluation of the aggregated results of all subcommands. If this is disabled, only one subcommand at a time can be used.

yaku.autopilot_utils.cli_base.read_version_from_package(package_with_version_file, version_file='_version.txt')#

Return a function which reads a version number from a file in a Python package.

To be used as version_callback in the make_autopilot_app() function.

Example:

cli = make_autopilot_app(
    provider=MyCliClass,
    version_callback=read_version_from_package("mycompany.mypackage", "version.txt"),
)

yaku.autopilot_utils.cli_base.ClickUsageErrorHandlerDecorator(f)#

Special decorator which wraps around the outermost click.command decorator.

This wrapping is done automatically by the make_autopilot_app() function.

This is a necessary workaround, because input parameter validation happens inside click.command. Usually, click handles parameter validation errors on its own, e.g. when a click.UsageError or click.BadParameter is raised inside a validation callback function, it automatically prints out the command usage instructions, together with the exception’s message and exits with an exit code of 2.

However we want to deal with usage errors differently in our autopilot interface:

we want to exit with code 0 in case of expected errors
we want to print out a JSON line with status=FAILED and a proper reason.

This is why this decorator exists: it wraps the outermost click.command (or click.group) decorator and simply forwards all unknown accesses to our object to the underlying f object (inside the __getattr__ function).

There are two possible invocation paths:

In case of testing, the main attribute is called.
In case of normal CLI invocation, the object/function is simply called.

That’s why accesses to main are rerouted as well as __call__ (in case the class is called).

Both accesses call the _wrapper_error_handler method, which has the correct logic for filtering out SystemExit exceptions which have a __context__ coming from a click.exceptions.UsageError.

Only in this case, the exception is modified to become a SystemExit(0) and a proper JSON line is printed.

All other cases are simply re-raised at the end of the _wrapper_error_handler method.

yaku.autopilot_utils.subprocess#

Support for running other autopilot apps as subprocesses.

When developing Custom Python Scripts, it might be necessary to call other autopilot apps. As they are standalone applications, they can only be called through normal system subprocesses.

To support this use-case better, this module provides a special run() function which mimics the builtin subprocess.run function, but extends the result of this function by some extra fields, e.g. for getting the list of outputs from an autopilot call. See run() for details.

As an example, we want to run two autopilot apps in a row and provide a special result message if both succeed:

# first we call the other app
step1 = run(["sharepoint-fetcher"])
# if the other app returned a non-zero exit code, the next line
# would immediately exit the program and print the other app's
# stdout and stderr log
step1.exit_for_returncode()
# if the other app returned a non-GREEN, non-RED, or non-YELLOW status,
# we can use this line to raise an `AutopilotSubprocessFailure` and stop
# any further processing in our current function.
# This would still print the other app's stdout and stderr log as well
# as the outputs and results of the other app.
step1.raise_for_status()
# now, we can (but don't have to) add the results from step 1
# to the main log so that they appear later in the HTML result
for r in step1.results:
    RESULTS.append(r)

# then we do the same for the second app
step2 = run(["artifactory-fetcher"])
step2.exit_for_returncode()
step2.raise_for_status()
for r in step2.results:
    RESULTS.append(r)

# and finally, when both previous steps have succeeded (GREEN or YELLOW)
# we can prepare a final message:
RESULTS.append(
    Result(
        criterion="Both files are fetched correctly.",
        fulfilled=True,
        justification="Both fetchers finished successfully.",
    )
)

Of course, this final step could also be done in the click_evaluator_callback, but the logic above allows full flexibility on the flow control between the different subprocesses. Also, _outputs_ could be retrieved from step1 and used for step2.

yaku.autopilot_utils.subprocess.run(command, /, shell=False, extra_env=None, **kwargs)#

Run another autopilot app in a subprocess.

This function is a wrapper around subprocess.run, so it executes the given command and passes any kwargs to the subprocess.run function.

Motivation for this wrapper was that there should be an easy way to call an autopilot app as a subfunction and access its status and results:

result = run(["sharepoint-fetcher"])
if result.status == 'GREEN': ...
if result.results[0].fulfilled: ...
# and so on

After the command has finished, it parses the resulting stdout and extracts status, reason, clean_stdout, results data and attaches it to the returned object.

Additionally, it provides two functions in the returned object: :rtype: ProcessResult

ProcessResult.exit_for_returncode() can be called if you want to do a system exit in case of a non-zero return code.
ProcessResult.raise_for_status() which raises a AutopilotSubprocessFailure if the subprocess app status is not GREEN, YELLOW, or RED (None is ignored as well).

class yaku.autopilot_utils.subprocess.ProcessResult(*args, **kwargs)#

Result of a subprocess run command.

Contains all attributes and methods from the built-in subprocess.run function.

Has extra attributes:

status: The result status of the autopilot app, e.g. GREEN or RED.
reason: The reason text for the status.
clean_stdout: A cleaned version of the stdout output, where all JSON lines have been removed.
results: A list of Result objects with the collected results of the autopilot app.
outputs: An OutputMap containing all autopilot outputs.

There are also two useful extra methods:

exit_for_returncode(): for returncodes unequal to zero, causes a program termination when called.
raise_for_status(): will raise an exception if there is no valid status.

Utilities#

exception yaku.autopilot_utils.subprocess.AutopilotSubprocessFailure(process_result)#

Indicate a failing subprocess.

If a subprocess fails, it is often required to skip any following checks. This exception can be used to indicate that some steps have failed and that evaluation should be done immediately.

class yaku.autopilot_utils.subprocess.OutputMap(mapping=None)#

Dictionary of a subprocess’ outputs, used by the run() function.

Has an extra to_json() method to convert the output data back into a JSON string.

yaku.autopilot_utils.results#

Handle evaluator results.

Usage#

Just import the RESULTS singleton and append results to it:

from yaku.autopilot_utils.results import RESULTS, Result

RESULTS.append(Result(criterion="...", fulfilled=False, justification="..."))

When using the click app template in [cli_base.py](./cli_base.py), make sure to define a click_evaluator_callback which receives the collected results and must return a status and a reason:

def click_evaluator_callback(results: ResultsCollector) -> Tuple[str, str]:
    for result in results:
        # ... examine results
        # ... and define reason, status
    return status, reason

Testing#

During tests, when modifying the RESULTS singleton, it must be reset after the test. This can be done with the protect_results decorator which you simply put around your test function:

from yaku.autopilot_utils.results import RESULTS, protect_results

@protect_results
def test_result_handling():
    # do something with RESULTS
    RESULTS.append(...)

# after the test has finished, RESULTS will be reset to previous state

Module contents#

yaku.autopilot_utils.results.RESULTS = []#

Singleton for storing and accessing results.

You can use it in your autopilot to store results for later evaluation:

RESULTS.append(Result(criterion="...", fulfilled=False, justification="..."))

yaku.autopilot_utils.results.DEFAULT_EVALUATOR(results)#

Evaluate results and return RED status if any criterion is not fulfilled.

This is the default implementation of the evaluator and can be used in papsr or autopilot apps. Simply use it as:

class CLI:
    click_evaluator_callback = DEFAULT_EVALUATOR

class yaku.autopilot_utils.results.Output(key, value)#

Simple data class to store key and value of an output.

Has an extra method to_json to convert the output data back to a JSON line.

class yaku.autopilot_utils.results.Result(criterion, fulfilled, justification, metadata=<factory>)#

Simple data class to store a result of an autopilot app.

Has fields criterion, fulfilled, and justification.

class yaku.autopilot_utils.results.ResultsCollector(iterable=(), /)#

List of Result objects.

Is used for the RESULTS singleton to collect all results of an autopilot run before the final evaluation.

Has an append(result: Result) method and a to_json() method.

Test helpers#

yaku.autopilot_utils.results.assert_no_result_status(output)#: Assert that there is no JSON line with a status in the output.

yaku.autopilot_utils.results.assert_result_status(output, expected_status, reason=None)#

Parse JSON lines in output and look for status and reason properties.

This function is a testing utility when you want to verify autopilot output. It parses the output and looks for JSON lines and then verifies that there is a status property with the expected_status and optionally it matches the reason property against a regular expression given in reason.

yaku.autopilot_utils.errors#

The following classes are used to differentiate between different error cases.

Collection of various exception classes for reporting different types of autopilot errors.

There are two main types of exception classes:

AutopilotException (used for wrapping internal failures into more helpful error messages)
- AutopilotError
AutopilotFailure (for all errors which the user might be able to resolve, e.g. configuration mistakes)

exception yaku.autopilot_utils.errors.AutopilotException#

Base class for all autopilot errors.

Those errors are not treated differently than normal Python exceptions. This base class is just there so that developers can use it to wrap other (built-in) exceptions into a custom exception with a custom error message.

exception yaku.autopilot_utils.errors.AutopilotError#

Exception for (more or less) unexpected autopilot errors.

For example can be used to annotate builtin exceptions with extra message, e.g.:

try:
    ...
except Exception as e:
    raise AutopilotError("During ..., an error occurred!") from e

exception yaku.autopilot_utils.errors.AutopilotFailure(reason)#

Base class for all autopilot failures.

Will immediately exit the command line invocation and print out a ‘FAILED’ status with the given reason. (this is implemented in yaku.autopilot_utils.cli_base)

Will not print out a stack trace as normal exceptions. (this is implemented in yaku.autopilot_utils.cli_base)

exception yaku.autopilot_utils.errors.AutopilotConfigurationError(reason)#: Errors related to autopilot configuration.

exception yaku.autopilot_utils.errors.EnvironmentVariableError(reason)#: Errors related to environment variables.

exception yaku.autopilot_utils.errors.AutopilotFileNotFoundError(reason)#: A required file is not found.

exception yaku.autopilot_utils.errors.FileNotFoundError(reason)#

A required file is not found.

Deprecated: use AutopilotFileNotFoundError instead.

yaku.autopilot_utils.environment#

yaku.autopilot_utils.environment.require_environment_variable(variable_name)#

Ensure that an environment variable is there and return its value.

Return type:: str

Python API

Contents

Python API#

yaku.autopilot_utils.cli_base#

Python App Template#

Simple App#

Click argument validation#

Complex apps/evaluators#

Module contents#

yaku.autopilot_utils.subprocess#

Utilities#

yaku.autopilot_utils.results#

Usage#

Testing#

Module contents#

Test helpers#

yaku.autopilot_utils.errors#

yaku.autopilot_utils.environment#