jobflow_remote.cli.utils module

class jobflow_remote.cli.utils.IndexDirection(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

ASC = 'asc'

DESC = 'desc'

property as_pymongo

class jobflow_remote.cli.utils.ReportInterval(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

DAYS = 'days'

HOURS = 'hours'

MONTHS = 'months'

WEEKS = 'weeks'

YEARS = 'years'

class jobflow_remote.cli.utils.ReprStr

Bases: str

Helper class that overrides the standard __repr__ to return the string itself and not its repr(). Used mainly to allow printing of strings with newlines instead of ‘n’ when repr is used in rich.

class jobflow_remote.cli.utils.SerializeFileFormat(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

JSON = 'json'

TOML = 'toml'

YAML = 'yaml'

class jobflow_remote.cli.utils.SortOption(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: str, Enum

CREATED_ON = 'created_on'

DB_ID = 'db_id'

UPDATED_ON = 'updated_on'

jobflow_remote.cli.utils.check_at_least_one_opt(d: dict) → None

jobflow_remote.cli.utils.check_incompatible_opt(d: dict) → None

jobflow_remote.cli.utils.check_only_one_opt(d: dict) → None

jobflow_remote.cli.utils.check_query_incompatibility(query, incompatible_options)

jobflow_remote.cli.utils.check_stopped_runner(error: bool = True) → None

jobflow_remote.cli.utils.check_valid_uuid(uuid_str, raise_on_error: bool = True) → bool

jobflow_remote.cli.utils.cleanup_job_controller() → None

jobflow_remote.cli.utils.cli_error_handler(func)

jobflow_remote.cli.utils.complete_profiling() → None

jobflow_remote.cli.utils.confirm_project_name(style: str | None = 'red', exit_on_error: bool = True) → bool

Ask the user to insert the name of the current project to confirm that a specific operation should be performed.

Parameters:

• style – The style to use for the message. If None, no style is used.

• exit_on_error – If True, exit the program with an error message if the user does not insert the correct project name. Otherwise, return False.

Returns:

True if the user inserted the correct project name, False otherwise.

Return type:

bool

jobflow_remote.cli.utils.execute_multi_jobs_cmd(single_cmd: Callable, multi_cmd: Callable, job_db_id: str | None = None, job_index: int | None = None, job_ids: list[str] | None = None, db_ids: str | list[str] | None = None, flow_ids: str | list[str] | None = None, states: JobState | list[JobState] | None = None, start_date: datetime | None = None, end_date: datetime | None = None, name: str | None = None, metadata: dict | None = None, days: int | None = None, hours: int | None = None, workers: list[str] | None = None, custom_query: dict | None = None, verbosity: int = 0, raise_on_error: bool = False, interactive: bool = True, **kwargs) → None

Utility function to execute a command on a single job or on a set of jobs. Checks the query options, determine whether the command is single or multi and call the corresponding function.

Parameters:

• single_cmd – A function that takes the job_id and job_index as arguments and performs an action on a single job.

• multi_cmd – A function that takes a set of standard arguments and performs an action on multiple jobs.

• job_db_id – The job id or db_id of a job to be considered. If specified, all the other query options should be disabled.

• job_index – The index of the job in the DB.

• job_ids – A list of job ids to be considered.

• db_ids – A list of db ids to be considered.

• flow_ids – A list of flow ids to be considered.

• states – The states of the jobs to be considered.

• start_date – The start date of the jobs to be considered.

• end_date – The end date of the jobs to be considered.

• name – The name of the jobs to be considered.

• metadata – The metadata of the jobs to be considered.

• days – Set the start_date based on the number of past days.

• hours – Set the start_date based on the number of past hours.

• workers – Workers associated with the jobs to be considered.

• custom_query – A custom query to be used to filter the jobs.

• verbosity – The verbosity of the output.

• raise_on_error – If True, an error will be raised if the operation fails.

• interactive – If True, a spinner will be shown even if the operation is not interactive while the operation is being performed.

• **kwargs (dict) – Additional arguments to be passed to the single_cmd and multi_cmd functions.

jobflow_remote.cli.utils.exit_with_error_msg(message: str | Text, code: int = 1, **kwargs) → NoReturn

jobflow_remote.cli.utils.exit_with_warning_msg(message: str | Text, code: int = 0, **kwargs) → NoReturn

jobflow_remote.cli.utils.get_command_tree(app: TyperGroup, tree: Tree, show_options: bool, show_docs: bool, show_hidden: bool, max_depth: int | None, current_depth: int = 0, max_doc_lines: int | None = None) → Tree

Recursively build a tree representation of the command structure.

Parameters:

• app – The Typer app or command group to process.

• tree – The Rich Tree object to add nodes to.

• show_options – Whether to display command options.

• show_docs – Whether to display command and option documentation.

• show_hidden – Whether to display hidden commands.

• max_depth – Maximum depth of the tree to display.

• current_depth – Current depth in the tree (used for recursion).

• max_doc_lines – If show_docs is True, the maximum number of lines showed from the documentation of each command/option.

Returns:

The populated Rich Tree object.

Return type:

Tree

jobflow_remote.cli.utils.get_config_manager() → ConfigManager

jobflow_remote.cli.utils.get_job_controller()

jobflow_remote.cli.utils.get_job_db_ids(job_db_id: str, job_index: int | None)

jobflow_remote.cli.utils.get_job_ids_indexes(job_ids: list[str] | None) → list[tuple[str, int]] | None

jobflow_remote.cli.utils.get_start_date(start_date: datetime | None, days: int | None, hours: int | None)

jobflow_remote.cli.utils.hide_progress(progress: Progress)

Hide the progress bar or spinning icon if an input is required from the user.

Adapted from a related github issue for rich: Textualize/rich#1535

Parameters:

progress – The Progress object in use

jobflow_remote.cli.utils.initialize_config_manager(*args, **kwargs) → None

jobflow_remote.cli.utils.loading_spinner(processing: bool = True)

jobflow_remote.cli.utils.print_success_msg(message: str | Text = 'operation completed', **kwargs) → None

jobflow_remote.cli.utils.start_profiling() → None

jobflow_remote.cli.utils.str_to_dict(string: str | None) → dict | None

jobflow_remote.cli.utils.tree_callback(ctx: Context, value: bool)