API Reference

class firexapp.application.ExitSignalHandler(app: FireXBaseApp)

Bases: object

first_warning = '\nExiting due to signal %s'

last_warning = "\nFINE! We'll stop. But you might have leaked a celery instance or a broker instance."

second_warning = '\nWe know! Have a little patience for crying out loud!'

class firexapp.application.FireXBaseApp(submit_app=None, info_app=None)

Bases: object

create_arg_parser(description=None) → ArgumentParser

main_error_exit_handler(reason=None, run_revoked=False)

run(sys_argv=None)

exception firexapp.application.JsonContentNotList

Bases: Exception

firexapp.application.get_app_task(task_short_name: str, all_tasks=None)

firexapp.application.get_app_tasks(tasks, all_tasks=None)

firexapp.application.get_args_from_json(json_file: str) → List[str]

firexapp.application.get_args_from_json_from_all_args(all_args: List[str]) → List[str]

firexapp.application.import_microservices(plugins_files=None, imports: tuple | None = None) → []

firexapp.application.main()

class firexapp.celery_manager.CeleryManager(plugins=None, logs_dir=None, worker_log_level='debug', cap_concurrency=None, app='firexapp.engine', env=None, broker=None)

Bases: object

static cap_cpu_count(count, cap_concurrency)

property celery_bin

celery_bin_name = 'celery'

property celery_logs_dir

property celery_pids_dir

extract_errors_from_celery_logs(celery_log_file, max_errors=20)

find_all_procs()

static find_procs(pid_file)

static get_celery_logs_dir(logs_dir)

static get_celery_pids_dir(logs_dir)

classmethod get_pid(logs_dir, workername, hostname='runner--azerasq-project-33192579-concurrent-0')

classmethod get_pid_file(logs_dir, workername, hostname='runner--azerasq-project-33192579-concurrent-0')

classmethod get_pid_from_file(pid_file)

static get_plugins_env(plugins)

static get_worker_and_host(workername, hostname)

classmethod get_worker_log_file(logs_dir, worker_and_host)

static get_worker_logs_dir(logs_dir)

classmethod get_worker_pids(logs_dir, hostname, workernames)

kill_all_forked(pid_file)

classmethod log(msg, header=None, level=10)

shutdown(timeout=60)

start(workername, queues=None, wait=True, timeout=900, concurrency=None, worker_log_level=None, app=None, cap_concurrency=None, cwd=None, soft_time_limit=None, autoscale: tuple | None = None, detach: bool = True)

classmethod terminate(pid, timeout=60)

update_env(env)

wait_for_shutdown(timeout=15)

wait_until_active(pid_file, stdout_file, workername, timeout=900)

property workers_logs_dir

exception firexapp.celery_manager.CeleryWorkerStartFailed

Bases: Exception

firexapp.common.create_link(src, target, delete_link=None, relative=False)

firexapp.common.delimit2list(str_to_split, delimiters=(',', ';', '|', ' ')) → []

firexapp.common.dict2str(mydict, sort=False, sep='    ', usevrepr=True, line_prefix='')

firexapp.common.find(keys, input_dict)

firexapp.common.find_procs(name, cmdline_regex=None, cmdline_contains=None)

firexapp.common.get_available_port(so_reuseport=True)

firexapp.common.poll_until_dir_empty(dir_path, timeout=15)

firexapp.common.poll_until_existing_file_not_empty(file_path, timeout=10)

firexapp.common.poll_until_file_exist(file_path, timeout=10)

firexapp.common.poll_until_file_not_empty(file_path, timeout=10)

firexapp.common.poll_until_path_exist(path, timeout=10)

firexapp.common.proc_matches(proc_info, pname, cmdline_regex, cmdline_contains)

firexapp.common.qualify_firex_bin(bin_name)

firexapp.common.render_template(template_str, template_args)

firexapp.common.select_env_vars(env_names)

firexapp.common.silent_mkdir(path, exist_ok=True, **kwargs)

firexapp.common.wait_until(predicate, timeout, sleep_for, *args, **kwargs)

firexapp.common.wait_until_pid_not_exist(pid, timeout=7, sleep_for=1)

class firexapp.discovery.PkgVersionInfo(pkg=None, version=None, commit=None)

Bases: PkgVersionInfo

firexapp.discovery.discover_package_modules(current_path, root_path=None) → [<class 'str'>]

firexapp.discovery.find_firex_task_bundles() → [<class 'str'>]

firexapp.discovery.get_all_pkg_versions() → [<class 'firexapp.discovery.PkgVersionInfo'>]

firexapp.discovery.get_all_pkg_versions_as_dict() → dict

firexapp.discovery.get_all_pkg_versions_str() → str

firexapp.discovery.get_firex_dependant_package_versions() → [<class 'firexapp.discovery.PkgVersionInfo'>]

firexapp.discovery.get_firex_tracking_services_entry_points() → [<class 'entrypoints.EntryPoint'>]

firexapp.discovery.loaded_firex_bundles_entry_points(path=None) → Dict[EntryPoint, object]

firexapp.discovery.loaded_firex_core_entry_points(path=None) → Dict[EntryPoint, object]

firexapp.discovery.loaded_firex_entry_points(path=None)

firexapp.discovery.prune_duplicate_module_entry_points(entry_points) → [<class 'entrypoints.EntryPoint'>]

class firexapp.fileregistry.FileRegistry(*args, **kwargs)

Bases: object

classmethod destroy()

dump_to_file(path)

get_file(key, uid_or_logsdir)

get_relative_path(key)

static read_from_file(path)

register_file(key, relative_path)

static resolve_path(uid_or_logsdir, relative_path)

exception firexapp.fileregistry.KeyAlreadyRegistered

Bases: Exception

exception firexapp.fileregistry.KeyNotRegistered

Bases: Exception

class firexapp.fileregistry.Singleton

Bases: type

class firexapp.info.InfoBaseApp

Bases: object

create_info_sub_parser(sub_parser)

create_list_sub_parser(sub_parser)

create_version_sub_parser(sub_parser)

classmethod parse_task_docstring(task)

static print_argument_used(plugins: str)

static print_available_microservices(plugins: str)

print_details(entity, plugins, all_tasks=None)

print_partial_task_matches(entity, all_tasks)

classmethod print_task_details(task)

run_info(args)

run_list(args)

static version(_args)

firexapp.info.get_argument_use(all_tasks) → dict

class firexapp.plugins.CommaDelimitedListAction(option_strings, dest, nargs=None, **kwargs)

Bases: Action

exception firexapp.plugins.PluginLoadError

Bases: Exception

firexapp.plugins.cdl2list(plugin_files)

firexapp.plugins.create_replacement_task(original, name_postfix, sigs)

firexapp.plugins.find_plugin_file(file_path)

firexapp.plugins.get_active_plugins()

firexapp.plugins.get_plugin_module_name(plugin_file)

firexapp.plugins.get_plugin_module_names(plugin_files)

firexapp.plugins.get_plugin_module_names_from_env()

firexapp.plugins.get_short_name(long_name: str) → str

firexapp.plugins.identify_duplicate_tasks(all_tasks, priority_modules: list) → [[]]

Returns a list of substitution. Each substitution is a list of microservices. The last will be the ‘dominant’ one. It will be the one used.

firexapp.plugins.import_plugin_file(plugin_file)

firexapp.plugins.import_plugin_files(plugin_files) → set[str]

firexapp.plugins.load_plugin_modules(plugin_files)

firexapp.plugins.load_plugin_modules_from_env()

firexapp.plugins.merge_plugins(*plugin_lists) → []

Merge comma delimited lists of plugins into a single list. Right-handed most significant plugin

firexapp.plugins.set_plugins_env(plugin_files)

firexapp.engine.celery.add_items_to_conf(conf=None, **_kwargs)

exception firexapp.submit.arguments.ChainArgException

Bases: Exception

class firexapp.submit.arguments.InputConverter

Bases: object

This class uses a singleton object design to store converters which parse the cli arguments. Converter functions are stored into the singleton InputConverter object by adding the @register decorator to the top of each desired function.

classmethod convert(pre_load=None, **kwargs) → dict

Activates conversion. kwargs provided are passed to any registered converter. This function should be called twice, and only twice. Once with initially loaded converters, and then with the secondary ones.

Parameters:

pre_load – Used for testing. preload is defaulted to None and will auto populate

classmethod instance() → ConverterRegister

Used for unit testing only

pre_load_was_run = False

classmethod register(*args)

Registers a callable object to be run during conversion. The callable should take in kwargs, and return a dict with any changes to the input arguments, or None if no changes are necessary.

Example single argument converter:

@InputConverter.register @SingleArgDecorator(‘something’) def convert_something(arg_value):

arg_value = arg_value.upper() return arg_value

Optionally, dependencies can defined at registration:

@InputConverter.register(‘other_converter’, ‘and_another_converter’) @SingleArgDecorator(‘something’) def convert_something(arg_value):

arg_value = arg_value.upper() return arg_value

Conversion occurs on two occasions, before microservices are loaded, or after. You can explicitly mark a converter to run pre-loading or post-loading of the ALL microservices by passing True (pre) or False (post) during registration. This design is used in the spirit of failing fast, providing early failure of runs before the bulk of microservices are imported. If bool is not provided, it will register to run pre unless loading has already occurred.

@InputConverter.register(‘other_converter’, False) @SingleArgDecorator(‘something’) def convert_something(arg_value):

… return arg_value

When a conversion fails the given function can simply call raise to instruct the user how to correct their inputs.

firexapp.submit.arguments.auto_load_pydev_debugging_plugin(kwargs)

firexapp.submit.arguments.convert_booleans(kwargs)

Converts standard true/false/none values to bools and None

firexapp.submit.arguments.find_unused_arguments(chain_args: {}, ignore_list: [], all_tasks: [])

Function to detect any arguments that are not explicitly consumed by any microservice.

Note:

This should be run AFTER all microservices have been loaded.

Parameters:

• chain_args (dict) – The dictionary of chain args to check

• ignore_list (list) – A list of exception arguments that are acceptable. This usually includes application args.

• all_tasks – A list of all microservices. Usually app.tasks

Returns:

A dictionary of un-applicable arguments

firexapp.submit.arguments.get_chain_args(other_args: [])

This function converts a flat list of –key value pairs into a dictionary

firexapp.submit.arguments.whitelist_arguments(argument_list: str | list)

Function for adding argument keys to the global argument whitelist. Used during validation of input arguments

:param argument_list:List of argument keys to whitelist. :type argument_list: list

class firexapp.submit.console.ChainInterruptedExceptionFilter(name='')

Bases: Filter

filter(record)

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class firexapp.submit.console.DistlibWarningsFilter(name='')

Bases: Filter

filter(record)

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class firexapp.submit.console.FireXColoredConsoleFormatter(*args, **kwargs)

Bases: TTYColoredFormatter

format(record)

Format a message from a record object.

class firexapp.submit.console.RequeueingUndeliverableFilter(name='')

Bases: Filter

filter(record)

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

class firexapp.submit.console.RetryFilter(name='')

Bases: Filter

filter(record)

Determine if the specified record is to be logged.

Returns True if the record should be logged, or False otherwise. If deemed appropriate, the record may be modified in-place.

firexapp.submit.console.add_filter_to_console(log_filter)

firexapp.submit.console.set_console_log_level(log_level)

firexapp.submit.console.setup_console_logging(module=None, stdout_logging_level=20, console_logging_formatter='%(green)s[%(asctime)s]%(reset)s[%(hostname)s] %(log_color)s%(message)s', console_datefmt='%H:%M:%S', stderr_logging_level=40, module_logger_logging_level=None)

class firexapp.submit.reporting.ReportGenerator

Bases: ABC

abstract add_entry(key_name, value, priority, formatters, **extra)

filter_formatters(all_formatters)

formatters = ()

abstract post_run_report(root_id=None, **kwargs)

This could runs in the context of __main__ if –sync, other in the context of celery. So the instance cannot be assumed be the same as in pre_run_report()

static pre_run_report(**kwarg)

This runs in the context of __main__

class firexapp.submit.reporting.ReportersRegistry

Bases: object

classmethod get_generators()

classmethod post_run_report(results, kwargs)

classmethod pre_run_report(kwargs)

firexapp.submit.reporting.report(key_name=None, priority=-1, **formatters)

Use this decorator to indicate what returns to include in the report and how to format it

class firexapp.submit.submit.AdjustCeleryConcurrency(option_strings, dest, nargs=None, const=None, default=None, type=None, choices=None, required=False, help=None, metavar=None)

Bases: Action

exception firexapp.submit.submit.FireXReturnCodeException(error_msg, firex_returncode)

Bases: Exception

class firexapp.submit.submit.JsonFileAction(option_strings, dest, nargs=None, const=None, default=None, type=None, choices=None, required=False, help=None, metavar=None)

Bases: Action

class firexapp.submit.submit.OptionalBoolean(option_strings, dest, nargs=None, const=None, default=None, type=None, choices=None, required=False, help=None, metavar=None)

Bases: Action

class firexapp.submit.submit.SubmitBaseApp(submission_tmp_file=None)

Bases: object

DEFAULT_MICROSERVICE = None

SUBMISSION_LOGGING_FORMATTER = '[%(asctime)s %(levelname)s] %(message)s'

check_for_failures(root_task_result_promise, unsuccessful_services)

convert_chain_args(chain_args) → dict

copy_submission_log()

create_logs_link(logs_link)

create_submit_parser(sub_parser)

del_parser_attributes()

dump_environ()

static error_banner(err_msg, banner_title='ERROR', logf=<bound method Logger.error of <Logger firexapp.submit.submit (WARNING)>>)

format_results_str(chain_results)

graceful_exit_on_failure(failure_caption: str)

init_file_logging()

install_configs: FireXInstallConfigs

static log_firex_pkgs_versions()

log_preamble()

Overridable method to allow a firex application to log on startup

static log_results(results_str)

main_error_exit_handler(chain_details=None, reason=None, run_revoked=False)

process_other_chain_args(args, other_args) → {}

process_sync(root_task_result_promise, chain_args)

resolve_install_configs_args(args_from_first_pass: ~argparse.Namespace, other_args_from_first_pass: list) -> (<class 'argparse.Namespace'>, <class 'list'>)

run_submit(args, others)

self_destruct(chain_details=None, reason=None, run_revoked=False)

set_broker_in_app()

start_broker(args)

start_celery(args, plugins)

start_engine(args, chain_args, uid) → {}

start_tracking_services(args, **chain_args) → {}

store_parser_attributes(arg_parser, submit_parser)

submit(args_from_first_pass: Namespace, other_args_from_first_pass: list)

classmethod validate_argument_applicability(chain_args, args, all_tasks)

wait_tracking_services_pred(service_predicate, description, timeout) → None

wait_tracking_services_release_console_ready(timeout=5) → None

wait_tracking_services_task_ready(timeout=5) → None

firexapp.submit.submit.celery_worker_ready(sender, **_kwargs)

firexapp.submit.submit.format_unsuccessful_services(unsuccessful_services)

firexapp.submit.submit.get_firex_id_from_output(cmd_output: str) → str

firexapp.submit.submit.get_log_dir_from_output(cmd_output: str) → str

firexapp.submit.submit.get_unsuccessful_items(list_of_tasks, filters=None)

firexapp.submit.submit.safe_create_completed_run_json(uid, chain_result, run_revoked, chain_args)

firexapp.submit.submit.safe_create_initial_run_json(**kwargs)

class firexapp.submit.tracking_service.TrackingService

Bases: ABC

extra_cli_arguments(arg_parser)

get_pkg_version_info() → PkgVersionInfo | None

install_configs: FireXInstallConfigs

ready_for_tasks(**kwargs) → bool

ready_release_console(**kwargs) → bool

abstract start(args, install_configs: FireXInstallConfigs, **kwargs) → {}

firexapp.submit.tracking_service.get_service_name(service: TrackingService) → str

firexapp.submit.tracking_service.get_tracking_services() → tuple[TrackingService] | None

firexapp.submit.tracking_service.get_tracking_services_versions() → list[PkgVersionInfo]

firexapp.submit.tracking_service.has_flame() → bool

class firexapp.submit.uid.FireXIdParts(user: str, timestamp: datetime.datetime, random_int: int)

Bases: object

firex_id()

random_int: int

timestamp: datetime

user: str

class firexapp.submit.uid.Uid(identifier=None)

Bases: object

add_viewers(**attrs)

property base_logging_dir

copy_resources()

create_debug_dir()

create_logs_dir()

property debug_dir

debug_dirname = 'firex_internal'

classmethod get_resources_path(logs_dir)

property logs_dir

property logs_url

property resources_dir

property run_data

property viewers

firexapp.submit.uid.firex_id_str(user: str, timestamp: datetime, random_int: int) → str

firexapp.submit.uid.get_firex_id_parts(maybe_firex_id: str) → FireXIdParts | None

firexapp.submit.uid.is_firex_id(maybe_firex_id: str) → bool

firexapp.tasks.example.employee_names_to_list(employee_names)

firexapp.tasks.example.to_list(guests: str) → list[str]

class firexapp.testing.config_base.FlowTestConfiguration

Bases: object

abstract assert_expected_firex_output(cmd_output, cmd_err)

abstract assert_expected_return_code(ret_value)

cleanup()

abstract initial_firex_options() → list

property name

run_data: FireXRunData | None

class firexapp.testing.config_base.InterceptFlowTestConfiguration

Bases: FlowTestConfiguration

assert_expected_firex_output(cmd_output, cmd_err)

abstract assert_expected_options(captured_kwargs)

assert_expected_return_code(ret_value)

abstract intercept_service() → str

Name of the microservice that will be mocked into the validator capturing the options that get compared by assertExpectedOptions()

run_data: FireXRunData | None

firexapp.testing.config_base.assert_is_bad_run(ret_value)

firexapp.testing.config_base.assert_is_good_run(ret_value)

firexapp.testing.config_base.discover_tests(tests, config_filter='') → list

firexapp.testing.config_base.import_test_configs(path) → []

firexapp.testing.config_base.skip_test(cls)

class firexapp.testing.config_interpreter.ConfigInterpreter

Bases: object

cleanup_after_timeout(std_out, std_err)

collect_plugins(flow_test_config) → []

create_cmd(flow_test_config) → List[str]

static create_mock_file(results_folder, results_file, test_name, intercept_microservice)

document_viewer(file_path: str) → str

execution_directory = None

get_exe(flow_test_config) → List[str]

static get_intercept_results_file(flow_test_config)

static get_test_name(flow_test_config)

static is_instance_of_intercept(test_config: FlowTestConfiguration)

static is_submit_command(test_config: FlowTestConfiguration)

on_test_exit(std_out, std_err)

run_executable(cmd, flow_test_config)

run_integration_test(flow_test_config, results_folder)

tmp_json_file: str | None

firexapp.testing.coverage_plugin.find_in_stack(file_to_find) → bool

firexapp.testing.coverage_plugin.is_celery() → bool

firexapp.testing.coverage_plugin.is_running_under_coverage() → bool

firexapp.testing.coverage_plugin.restart_celery_under_coverage()

class firexapp.testing.test_infra.FlowTestInfra(methodName='runTest')

Bases: TestCase

config_interpreter = <firexapp.testing.config_interpreter.ConfigInterpreter object>

failures = 0

max_acceptable_failures = None

classmethod populate_tests()

results_dir = None

setUp()

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

tearDown()

Hook method for deconstructing the test fixture after testing it.

test_configs = []

firexapp.testing.test_infra.default_main()

firexapp.testing.test_infra.main(default_results_dir, default_test_dir)