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_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.main()

class firexapp.celery_manager.CeleryManager(logs_dir: str, fx_env: FxEnvVars, plugins: None | str | list[str] = None, app='firexapp.engine.celery:app', env=None)

Bases: object

static cap_cpu_count(count, cap_concurrency)

property celery_pids_dir

find_all_procs()

classmethod get_celery_pid(logs_dir: str, worker_and_host: str)

static get_worker_logs_dir(logs_dir: str) → str

classmethod is_current_env_fx_celery_worker() → bool

kill_all_forked(pid_file)

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

shutdown(timeout=60)

start_celery_worker(workername: str, queues=None, wait=True, timeout=60, concurrency=None, cap_concurrency=None, cwd=None, soft_time_limit=None, autoscale: tuple | None = None, detach: bool = True, celery_cmd_log_level=10)

classmethod terminate(pid, timeout=60)

classmethod unset_start_fx_celery_worker_env()

update_env(env)

wait_for_shutdown(timeout=15)

exception firexapp.celery_manager.CeleryWorkerStartFailed

Bases: Exception

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

firexapp.common.create_link_async(src: str, target: str, **create_link_kwargs) → Thread

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

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

firexapp.common.find(keys, input_dict)

firexapp.common.get_available_port()

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

firexapp.common.poll_until_existing_file_not_empty(file_path, timeout: float = 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.qualify_firex_bin(bin_name)

firexapp.common.render_template(template_str, template_args)

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

firexapp.common.wait_until(predicate: Callable[[...], T], timeout: float, sleep_for: float, *args, **kwargs) → T

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) → list[str]

firexapp.discovery.find_firex_task_bundles() → list[str]

firexapp.discovery.get_all_pkg_versions() → list[PkgVersionInfo]

firexapp.discovery.get_all_pkg_versions_str() → str

firexapp.discovery.get_firex_dependant_package_versions() → list[PkgVersionInfo]

firexapp.discovery.get_firex_tracking_services_entry_points() → list[EntryPoint]

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) → list[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)

classmethod print_argument_used(plugins: str)

classmethod 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

class firexapp.plugins.FxPluginRegistry

Bases: object

create_replacement_task(fx_app, original, name_postfix, sigs)

classmethod find_plugin_file(file_path: str) → str

classmethod import_plugin_file(plugin_file: str, replace=False) → ModuleType | None

load_plugin_modules(fx_app, plugin_files: str | list[str], log_level: int)

classmethod resolve_plugin_paths(plugin_files: None | str | list[str]) → list[str]

classmethod set_plugins_env(plugin_files)

exception firexapp.plugins.PluginLoadError

Bases: Exception

firexapp.plugins.convert_plugins_to_list(plugin_files: None | str | list[str]) → list[str]

firexapp.plugins.get_active_plugins() → str

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

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

firexapp.plugins.plugins_has(plugins: str | list[str], query_basename: str) → bool

Check if a plugin basename is present in the plugins string or list.

Args:

plugins: Either a comma-separated string of plugin paths or a list of plugin paths query_basename: The basename of the plugin to search for (e.g., ‘sparse_build.py’)

Returns:

True if the query_basename is found in plugins, False otherwise

class firexapp.engine.default_celery_config.FxCeleryConfig(fx_env: firexapp.engine.default_celery_config.FxEnvVars, timestamp_format: str = '<small>[%(asctime)s]', process_format: str = '[%(levelname)s/%(processName)-13s]', task_format: str = '[%(task_id).8s-%(task_name)s]', message_format: str = ':</small> %(message)s', worker_log_format: str = '<small>[%(asctime)s][%(levelname)s/%(processName)-13s]:</small> %(message)s', worker_task_log_format: str = '<small>[%(asctime)s][%(levelname)s/%(processName)-13s][%(task_id).8s-%(task_name)s]:</small> %(message)s', broker_connection_retry_on_startup: bool = True, worker_prefetch_multiplier: int = 1, worker_redirect_stdouts_level: str = 'PRINT', task_soft_time_limit: int = 4320, link_for_logo: str | None = None, logs_url: str | None = None, resources_dir: str | None = None)

Bases: object

accept_content = ['pickle', 'json']

property broker: str

broker_connection_retry_on_startup: bool = True

property broker_url: str

fx_env: FxEnvVars

property imports: tuple[str, ...]

link_for_logo: str | None = None

load_install_config()

property logs_dir: str

logs_url: str | None = None

property mc: str

message_format: str = ':</small> %(message)s'

primary_worker_minimum_concurrency: ClassVar[int] = 4

primary_worker_name = 'mc'

process_format: str = '[%(levelname)s/%(processName)-13s]'

property redis_bin_dir: str

resources_dir: str | None = None

property result_backend: str

result_expires = None

result_serializer = 'pickle'

root_task = 'firexapp.tasks.root_tasks.RootTask'

task_acks_late = True

task_default_queue = 'mc'

task_format: str = '[%(task_id).8s-%(task_name)s]'

task_serializer = 'pickle'

task_soft_time_limit: int = 4320

task_track_started = True

timestamp_format: str = '<small>[%(asctime)s]'

property uid: str

worker_autoscaler

alias of FireXAutoscaler

worker_log_format: str = '<small>[%(asctime)s][%(levelname)s/%(processName)-13s]:</small> %(message)s'

worker_prefetch_multiplier: int = 1

worker_redirect_stdouts_level: str = 'PRINT'

worker_task_log_format: str = '<small>[%(asctime)s][%(levelname)s/%(processName)-13s][%(task_id).8s-%(task_name)s]:</small> %(message)s'

class firexapp.engine.default_celery_config.FxEnvVars(*, CURRENT_RUN_FIREX_ID: str, firex_logs_dir: str, redis_bin_dir: str, BROKER: str, firex_plugins: Annotated[str, BeforeValidator(func=_plugins_to_csv, json_schema_input_type=PydanticUndefined)] = '')

Bases: BaseModel

BROKER: str

CURRENT_RUN_FIREX_ID: str

property broker_url: str

classmethod clear_os_fx_env()

classmethod create_no_task_exec_fx_env(plugins=None, logs_dir='') → Self

property firex_id: str

firex_logs_dir: str

firex_plugins: Annotated[str, BeforeValidator(func=_plugins_to_csv, json_schema_input_type=PydanticUndefined)]

get_plugin_files() → list[str]

get_redis_hostname() → str

classmethod load_firex_env_vars_from_env() → Self

classmethod load_firex_env_vars_from_logs_dir(logs_dir: str, plugins) → Self

property logs_dir: str

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

redis_bin_dir: str

classmethod select_minimal_fx_env_from_os_env() → dict[str, str]

set_fx_os_env()

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: dict[str, Any], ignore_list: list[str], all_tasks: dict[str, Any])

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.get_global_args_allowlist() → frozenset[str]

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=None, 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)

filter_loaders(all_loaders)

formatters = ()

abstract load_data(key_name, value, loaders, **extra)

loaders = ()

abstract post_run_report(root_async_result: FxAsyncResult, **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(*args, **kwarg)

This runs in the context of __main__

class firexapp.submit.reporting.ReportersRegistry

Bases: object

classmethod get_generators() → list[ReportGenerator]

classmethod post_run_report(root_async_result: FxAsyncResult, chain_args: dict[str, Any])

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

firexapp.submit.reporting.report_data(key_name=None, **loaders)

Use this decorator to indicate what returns to include in the report and how to load 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: str | None = None)

Bases: object

DEFAULT_MICROSERVICE = None

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

check_for_failures(root_task_result_promise: FxAsyncResult, unsuccessful_services: dict[str, list[FxAsyncResult]])

convert_chain_args(chain_args) → dict

copy_submission_log()

create_chain_sentinel_files(chain: str)

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()

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: tuple[FxAsyncResult, dict[str, Any]] | None = None, reason=None, run_revoked=False)

process_other_chain_args(args, other_args) → dict[str, Any]

process_sync(root_task_result_promise: FxAsyncResult, chain_args: dict[str, Any])

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: tuple[FxAsyncResult, dict[str, Any]] | None = None, reason: str | None = None, run_revoked: bool = False)

start_broker(logs_dir: str, args) → RedisManager

start_celery(fx_app: FireXCelery, args, plugins)

start_engine(fx_app_cls: type[T], args: Namespace, chain_args: dict[str, Any], uid: Uid) → tuple[T, dict[str, Any]]

start_tracking_services(args, **chain_args) → dict[str, Any]

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: dict[str, Any], arg_names: set[str], all_tasks: dict[str, Any]) → bool

wait_tracking_services_pred(service_predicate, description, timeout) → None

wait_tracking_services_release_console_ready(timeout=45) → None

wait_tracking_services_task_ready(fx_app, timeout=5) → None

static write_run_complete_file(log_path: str)

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

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

firexapp.submit.submit.get_unsuccessful_items(ars: list[FxAsyncResult], filters=None) → list[str]

firexapp.submit.submit.safe_create_initial_run_json(**kwargs) → FireXRunData | None

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, ...]

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()

static from_str(firex_id: str) → FireXIdParts

random_int: int

timestamp: datetime

user: str

class firexapp.submit.uid.Uid(identifier=None, firex_requester=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: str

property logs_url

property resources_dir

property viewers

firexapp.submit.uid.find_all_firex_ids_from_str(input_str: str | None) → list[str]

firexapp.submit.uid.find_single_firex_id_from_str(input_str) → str

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

assert_expected_firex_output(cmd_output, cmd_err)

assert_expected_return_code(ret_value)

cleanup()

property completed_run: FireXRunData

static get_extra_run_env()

abstract initial_firex_options() → list

property name

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) → list[FlowTestConfiguration]

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) → list[str]

create_cmd(flow_test_config: FlowTestConfiguration) → list[str]

document_viewer(file_path: str) → str

execution_directory = None

get_exe(flow_test_config) → list[str]

static get_test_name(flow_test_config)

static is_submit_command(test_config: FlowTestConfiguration)

on_test_exit(std_out, std_err)

run_executable(cmd, flow_test_config: FlowTestConfiguration)

run_integration_test(flow_test_config, results_folder)

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)