Tasks API

Each task must implement a TaskInterface (directly, or through a Base Task)

Attention

Methods marked as abstract must be implemented by your task that extends directly from TaskInterface.

Tip

During configuration and execution stage every task is having it’s own ExecutionContext instance. ExecutionContext (called ctx) gives access to parameters, environment variables, user input (e.g. stdin) Do not try to manually read from stdin, or os.environment - read more about this topic in Best practices chapter.

class rkd.core.api.contract. TaskInterface

abstract configure_argparse ( parser : argparse.ArgumentParser )

Allows a task to configure ArgumentParser (argparse)

def configure_argparse(self, parser: ArgumentParser):
    parser.add_argument('--php', help='PHP version ("php" docker image tag)', default='8.0-alpine')
    parser.add_argument('--image', help='Docker image name', default='php')

copy_internal_dependencies ( task )

Allows to execute a task-in-task, by copying dependent services from one task to other task

exec ( cmd : str , capture : bool = False , background : bool = False ) → Optional [ str ]

Starts a process in shell. Throws exception on error. To capture output set capture=True

NOTICE: Use instead of subprocess. Raw subprocess is less supported and output from raw subprocess

may be not catch properly into the logs

abstract execute ( context : rkd.core.api.contract.ExecutionContext ) → bool

Executes a task. True/False should be returned as return

extends_task ( )

Provides information if this Task has a Parent Task

Returns

format_task_name ( name : str ) → str

Allows to add a fancy formatting to the task name, when the task is displayed eg. on the :tasks list

get_become_as ( ) → str

User name in UNIX/Linux system, optional. When defined, then current task will be executed as this user (WARNING: a forked process would be started)

classmethod get_declared_envs ( ) → Dict [ str , Union [ str , rkd.core.api.contract.ArgumentEnv ] ]

Dictionary of allowed envs to override: KEY -> DEFAULT VALUE

All environment variables fetched from the ExecutionContext needs to be defined there. Declared values there are automatically documented in –help

@classmethod
def get_declared_envs(cls) -> Dict[str, Union[str, ArgumentEnv]]:
    return {
        'PHP': ArgumentEnv('PHP', '--php', '8.0-alpine'),
        'IMAGE': ArgumentEnv('IMAGE', '--image', 'php')
    }

get_full_name ( )

Returns task full name, including group name

abstract get_group_name ( ) → str

Group name where the task belongs eg. “:publishing”, can be empty.

abstract get_name ( ) → str

Task name eg. “:sh”

io ( ) → rkd.core.api.inputoutput.IO

Gives access to Input/Output object

py ( code : str = '' , become : Optional [ str ] = None , capture : bool = False , script_path : Optional [ str ] = None , arguments : str = '' ) → Optional [ str ]

Executes a Python code in a separate process

NOTICE: Use instead of subprocess. Raw subprocess is less supported and output from raw subprocess

may be not catch properly into the logs

rkd ( args : list , verbose : bool = False , capture : bool = False ) → str

Spawns an RKD subprocess

NOTICE: Use instead of subprocess. Raw subprocess is less supported and output from raw subprocess

may be not catch properly into the logs

sh ( cmd : str , capture : bool = False , verbose : bool = False , strict : bool = True , env : Optional [ dict ] = None , use_subprocess : bool = False ) → Optional [ str ]

Executes a shell script in bash. Throws exception on error. To capture output set capture=True

NOTICE: Use instead of subprocess. Raw subprocess is less supported and output from raw subprocess

may be not catch properly into the logs

should_fork ( ) → bool

Decides if task should be ran in a separate Python process (be careful with it)

silent_sh ( cmd : str , verbose : bool = False , strict : bool = True , env : Optional [ dict ] = None ) → bool

sh() shortcut that catches errors and displays using IO().error_msg()

NOTICE: Use instead of subprocess. Raw subprocess is less supported and output from raw subprocess

may be not catch properly into the logs

To include a task, wrap it in a declaration

Note

Task declaration declares a Task (TaskInterface implementation) to be a runnable task imported into a given Makefile.

Tip

With TaskDeclaration there is a possibility to customize things like task name, environment, working directory and other attributes.

class rkd.core.api.syntax. TaskDeclaration ( task : rkd.core.api.contract.TaskInterface , env : Optional [ Dict [ str , str ] ] = None , args : Optional [ List [ str ] ] = None , workdir : Optional [ str ] = None , internal : Optional [ bool ] = None , name : Optional [ str ] = None )

Task Declaration is a DECLARED USAGE of a Task (instance of TaskInterface)

Examples of usage:

TaskDeclaration(MyNiceTask(), env={'SOME': 'thing'}, workdir='/tmp', name=':custom:task:name')

To create an alias for task or multiple tasks

Note

TaskAlias is a simplified pipeline form, it is a chain of tasks written in a string form.

class rkd.core.api.syntax. TaskAliasDeclaration ( name : str , to_execute : List [ Union [ str , rkd.core.api.contract.PipelinePartInterface ] ] , env : Optional [ Dict [ str , str ] ] = None , description : str = '' )

Deprecated: Name will be removed in RKD 6.0

Execution context provides parsed shell arguments and environment variables

class rkd.core.api.contract. ExecutionContext ( declaration : rkd.core.api.contract.TaskDeclarationInterface , parent : Optional [ rkd.core.api.contract.GroupDeclarationInterface ] = None , args : Dict [ str , str ] = {} , env : Dict [ str , str ] = {} , defined_args : Dict [ str , dict ] = {} )

Defines which objects could be accessed by Task. It’s a scope of a single task execution.

can_mutate_globals ( ) → bool

Is task having a special permissions to mutate globals such as OS environment :return:

get_arg ( name : str ) → Optional [ str ]

Get argument or option

Usage:

ctx.get_arg(‘–name’) # for options ctx.get_arg(‘name’) # for arguments

Raises:

KeyError when argument/option was not defined

Returns:

Actual value or default value

get_arg_or_env ( name : str ) → Optional [ str ]

Provides value of user input

Usage:

get_arg_or_env(‘–file-path’) resolves into FILE_PATH env variable, and –file-path switch (file_path in argparse)

Behavior:

When user provided explicitly switch eg. –history-id, then it’s value will be taken in priority. If switch –history-id was not used, but user provided HISTORY_ID environment variable, then it will be considered.

If no switch provided and no environment variable provided, but a switch has default value - it would be returned.

If no switch provided and no environment variable provided, the switch does not have default, but environment variable has a default value defined, it would be returned.

When the –switch has default value (user does not use it, or user sets it explicitly to default value), and environment variable SWITCH is defined, then environment variable would be taken.

From RKD 2.1 the environment variable names can be mapped to any ArgParse switch.

Below example maps “COMMAND” environment variable to “–cmd” switch.

def get_declared_envs(self) -> Dict[str, Union[str, ArgumentEnv]]:
    return {
        'COMMAND': ArgumentEnv(name='COMMAND', switch='--cmd', default='')
    }

Raises:

MissingInputException: When no switch and no environment variable was provided, then an exception is thrown.

get_env ( name : str , switch : str = '' , error_on_not_used : bool = False )

Get environment variable value

Interaction with input and output

Tip

From inside a Task the IO can be accessed with self.io()

Caution

Every task has it’s own instance of IO, with customized per-task log level.

class rkd.core.api.inputoutput. IO

Interacting with input and output - stdout/stderr/stdin, logging

add_output_processor ( callback : Callable [ [ Union [ str , bytes ] , str ] , Union [ str , bytes ] ] )

Registers a output processing callback Each byte outputted by this IO instance will go through a set of registered processors

Example use cases:

• Hide sensitive information (secrets)

• Reformat output

• Strip long stdouts from commands

• Change colors

• Add/remove formatting

Parameters

callback –

Returns

capture_descriptors ( target_files : List [ str ] = None , stream = None , enable_standard_out : bool = True )

Capture stdout and stderr from a block of code - use with ‘with’

critical ( text )

Logger: critical

debug ( text )

Logger: debug

err ( text )

Standard error

errln ( text )

Standard error + newline

error ( text )

Logger: error

error_msg ( text )

Error message

static format_table ( header : list , body : list , tablefmt : str = 'simple' , floatfmt : str = 'g' , numalign : str = 'decimal' , stralign : str = 'left' , missingval : str = '' , showindex : str = 'default' , disable_numparse : bool = False , colalign : Optional [ str ] = None )

Renders a table

Parameters:

header: body: tablefmt: floatfmt: numalign: stralign: missingval: showindex: disable_numparse: colalign:

Returns:

Formatted table as string

h1 ( text )

Heading #1 (optional output)

h2 ( text )

Heading #2 (optional output)

h3 ( text )

Heading #3 (optional output)

h4 ( text )

Heading #3 (optional output)

info ( text )

Logger: info

info_msg ( text )

Informational message (optional output)

internal ( text )

Logger: internal Should be used only by RKD core for more intensive logging

internal_lifecycle ( text )

Should be used only by RKD core for more intensive logging :param text: :return:

is_silent ( ) → bool

Is output silent? In silent mode OPTIONAL MESSAGES are not shown

opt_errln ( text )

Optional errln()

opt_out ( text )

Optional output - fancy output skipped in –silent mode

opt_outln ( text )

Optional output - fancy output skipped in –silent mode + newline

out ( text )

Standard output

outln ( text )

Standard output + newline

print_group ( text )

Prints a colored text inside brackets [text] (optional output)

print_line ( )

Prints a newline

print_opt_line ( )

Prints a newline (optional output)

print_separator ( status : Optional [ bool ] = None )

Prints a text separator (optional output)

success_msg ( text )

Success message (optional output)

warn ( text )

Logger: warn

warn_msg ( text ) → None

Warning message (optional output)

Storing temporary files

Tip

From inside a Task the TempManager can be accessed with self.temp

class rkd.core.api.temp. TempManager ( chdir : str = './.rkd/' )

Manages temporary files inside .rkd directory Using this class you make sure your code is more safe to use on Continuous Integration systems (CI)

Usage:

path = self.temp.assign_temporary_file(mode=0o755)

assign_temporary_file ( mode : int = 493 ) → str

Assign a path for writing temporary files in RKD workspace

Note: The RKD is executing the finally_clean_up() at the end of each task

Usage:

try:

path = RKDTemp.assign_temporary_file_path() # (…) some action there

finally:

RKDTemp.finally_clean_up()

finally_clean_up ( )

Used to clean up all temporary files at the end of the code execution

TaskExecutor is running this method after each finished task

Parsing RKD syntax

class rkd.core.api.parsing. SyntaxParsing

static parse_import_as_type ( import_str : str ) → Type [ rkd.core.api.contract.TaskInterface ]

Import a Python class as a type

Example: rkd.core.standardlib.jinja.FileRendererTask

Parameters

import_str –

Returns

classmethod parse_imports_by_list_of_classes ( classes_or_modules : List [ str ] ) → List [ rkd.core.api.syntax.TaskDeclaration ]

Parses a List[str] of imports, like in YAML syntax. Produces a List[TaskDeclaration] with imported list of tasks.

Could be used to import & validate RKD tasks.

Examples:

• rkd.core.standardlib

• rkd.core.standardlib.jinja.FileRendererTask

:raises ParsingException :return:

Testing

Tip

BasicTestingCase is best for unit testing

class rkd.core.api.testing. BasicTestingCase ( methodName = 'runTest' )

Provides minimum of:

• Doing backup of environment and cwd

• Methods for mocking task dependencies (RKD-specific like ExecutionContext)

environment ( environ : dict )

Mocks environment

Example usage:

with self.environment({'RKD_PATH': SCRIPT_DIR_PATH + '/../docs/examples/env-in-yaml/.rkd'}):
    # code there

Parameters

environ –

Returns

static list_to_str ( in_list : list ) → List [ str ]

Execute __str__ on each list element, and replace element with the result

static mock_execution_context ( task : rkd.core.api.contract.TaskInterface , args : Optional [ Dict [ str , Union [ str , bool ] ] ] = None , env : Optional [ Dict [ str , str ] ] = None , defined_args : Optional [ Dict [ str , dict ] ] = None ) → rkd.core.api.contract.ExecutionContext

Prepares a simplified rkd.core.api.contract.ExecutionContext instance

Parameters

• task –

• args –

• env –

• defined_args –

Returns

static satisfy_task_dependencies ( task : rkd.core.api.contract.TaskInterface , io : Optional [ rkd.core.api.inputoutput.IO ] = None ) → rkd.core.api.contract.TaskInterface

Inserts required dependencies to your task that implements rkd.core.api.contract.TaskInterface

Parameters

• task –

• io –

Returns

setUp ( ) → None

Hook method for setting up the test fixture before exercising it.

tearDown ( ) → None

Hook method for deconstructing the test fixture after testing it.

Tip

FunctionalTestingCase should be using for tests that are running single task and asserting output contents.

class rkd.core.api.testing. FunctionalTestingCase ( methodName = 'runTest' )

Provides methods for running RKD task or multiple tasks with output and exit code capturing. Inherits OutputCapturingSafeTestCase.

execute_mocked_task_and_get_output ( task : rkd.core.api.contract.TaskInterface , args = None , env = None ) → str

Run a single task, capturing it’s output in a simplified way. There is no whole RKD bootstrapped in this operation.

Parameters

• task ( TaskInterface ) –

• args ( dict ) –

• env ( dict ) –

Returns

classmethod filter_out_task_events_from_log ( out : str )

Produces an array of events unformatted

[
    "Executing :sh -c echo 'Rocker' [part of :example]",
    "Executing :sh -c echo 'Kropotkin' [part of :example]",
    'Executing :sh -c echo "The Conquest of Bread"; exit 1 [part of :example]',
    'Retrying :sh -c echo "The Conquest of Bread"; exit 1 [part of :example]',
    'Executing :sh -c exit 0 ',
    'Executing :sh -c echo "Modern Science and Anarchism"; [part of :example]'
]

Parameters

out –

Returns

run_and_capture_output ( argv : list , verbose : bool = False ) → Tuple [ str , int ]

Run task(s) and capture output + exit code. Whole RKD from scratch will be bootstrapped there.

Example usage:

full_output, exit_code = self.run_and_capture_output([‘:tasks’])

Parameters

• argv ( list ) – List of tasks, arguments, commandline switches

• verbose ( bool ) – Print all output also to stdout

Returns

with_temporary_workspace_containing ( files : Dict [ str , str ] )

Creates a temporary directory as a workspace and fills up with files specified in “file” parameter

Parameters

files – Dict of [filename: contents to write to a file]

Returns

class rkd.core.api.testing. OutputCapturingSafeTestCase ( methodName = 'runTest' )

Provides hooks for keeping stdout/stderr immutable between tests.

setUp ( ) → None

Hook method for setting up the test fixture before exercising it.

tearDown ( ) → None

Hook method for deconstructing the test fixture after testing it.