Tasks API¶
Each task must implement a TaskInterface¶
-
class
rkd.api.contract.
TaskInterface
¶ -
configure_argparse
(parser: argparse.ArgumentParser)¶ Allows a task to configure ArgumentParser (argparse)
-
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
-
execute
(context: rkd.api.contract.ExecutionContext) → bool¶ Executes a task. True/False should be returned as return
-
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)
-
get_declared_envs
() → Dict[str, Union[str, rkd.api.contract.ArgumentEnv]]¶ Dictionary of allowed envs to override: KEY -> DEFAULT VALUE
-
get_full_name
()¶ Returns task full name, including group name
-
get_group_name
() → str¶ Group name where the task belongs eg. “:publishing”, can be empty.
-
get_name
() → str¶ Task name eg. “:sh”
-
io
() → rkd.api.inputoutput.IO¶ Gives access to Input/Output object
-
py
(code: str = '', become: str = None, capture: bool = False, script_path: 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: dict = None) → 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: 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
-
static
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: str = None)¶ Renders a table
Parameters: - header –
- body –
- tablefmt –
- floatfmt –
- numalign –
- stralign –
- missingval –
- showindex –
- disable_numparse –
- colalign –
Returns: Formatted table as string
-
To include a task, wrap it in a declaration¶
-
class
rkd.api.syntax.
TaskDeclaration
(task: rkd.api.contract.TaskInterface, env: Dict[str, str] = {}, args: List[str] = [])¶
To create an alias for task or multiple tasks¶
-
class
rkd.api.syntax.
TaskAliasDeclaration
(name: str, to_execute: List[str], env: Dict[str, str] = {}, description: str = '')¶ Allows to define a custom task name that triggers other tasks in proper order
Execution context provides parsed shell arguments and environment variables¶
-
class
rkd.api.contract.
ExecutionContext
(declaration: rkd.api.contract.TaskDeclarationInterface, parent: Optional[rkd.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.
-
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.
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¶
-
class
rkd.api.inputoutput.
IO
¶ Interacting with input and output - stdout/stderr/stdin, logging
-
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 (optional output)
-
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)
-
is_silent
() → bool¶ Is output silent? In silent mode OPTIONAL MESSAGES are not shown
-
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
()¶ Prints a text separator (optional output)
-
success_msg
(text)¶ Success message (optional output)
-
warn
(text)¶ Logger: warn
-
Storing temporary files¶
-
class
rkd.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
Testing¶
-
class
rkd.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’}):
- …
Parameters: environ – Returns:
-
static
mock_execution_context
(task: rkd.api.contract.TaskInterface, args: Dict[str, str] = None, env: Dict[str, str] = None, defined_args: Dict[str, dict] = None) → rkd.api.contract.ExecutionContext¶ Prepares a simplified rkd.api.contract.ExecutionContext instance
Parameters: - task –
- args –
- env –
- defined_args –
Returns:
-
static
satisfy_task_dependencies
(task: rkd.api.contract.TaskInterface, io: rkd.api.inputoutput.IO = None) → rkd.api.contract.TaskInterface¶ Inserts required dependencies to your task that implements rkd.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.
-
class
rkd.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.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:
-
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:
-
-
class
rkd.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.
-