Jobbergate CLI Reference

jobbergate_cli

Jobbergate command-line interface and app library

__getattr__

__getattr__(name: str)

Overload module attribute lookup to warn if 'appform' is being imported because it is deprecated.

application_base

Provide a stub module to maintain compatibility with previous versions.

Issue a deprecation warning when this module is imported from if JOBBERGATE_COMPATIBILITY_MODE is enabled.

If JOBBERGATE_COMPATIBILITY_MODE is not enabled, raise an import error when this module is imported.

auth

Utilities for handling auth in jobbergate-cli.

open_on_browser

open_on_browser(url: str) -> bool

Open the url on the browser using webbrowser.

show_login_message

show_login_message(device_code_data: DeviceCodeData)

Show a message to the user with a link to the auth provider to login.

track_login_progress

track_login_progress(iterable: Iterable) -> Iterable

Track the progress of the login process on a progress bar.

compat

Provide compatibility to the previous version of Jobbergate CLI for users who have automation or are familiar with the old commands

add_legacy_compatible_commands

add_legacy_compatible_commands(app: typer.Typer)

Add commands from the restructured CLI under the previous names for the commands to the root typer app.

config

Configuration file, sets all the necessary environment variables. Can load configuration from a dotenv file if supplied.

Settings

Bases: BaseSettings

Provide a pydantic settings model to hold configuration values loaded from the environment.

is_onsite_mode property

is_onsite_mode: bool

Check if the SBATCH_PATH is set, indicating that the CLI is running in on-site mode.

build_settings

build_settings(*args, **kwargs)

Return a Setting object and handle ValidationError with a message to the user.

constants

Provide constants that may be used throughout the CLI modules.

FileType

Bases: str, Enum

File type enum.

PaginationChoices

Bases: str, Enum

Enum describing the type of pagination that is available for list commands.

__str__

__str__() -> str

Return the string representation of the enum.

SortOrder

Bases: str, Enum

Enum descring the type of sort orders that are available for list commands.

context

Provides a data object for context passed from the main entry point.

JobbergateContext dataclass

Bases: ContextProtocol

A data object describing context passed from the main entry point.

authentication_handler cached property

authentication_handler: JobbergateAuthHandler

The authentication handler for the context.

This is a cached property to ensure that the handler is only created when needed, so commands that require no authentication face no configuration errors.

client cached property

client: Client

Client for making requests to the Jobbergate API.

exceptions

Provide exceptions and custom handlers for the CLI.

Abort

Bases: Buzz

A special exception used to abort the Jobbergate CLI.

Collects information provided for use in the handle_abort context manager.

__init__

__init__(message, *args, subject=None, support=False, log_message=None, sentry_context=None, original_error=None, warn_only=False, **kwargs)

Initialize the Abort error.

JobbergateCliError

Bases: Buzz

A generic exception base class to use in Jobbergate CLI

handle_abort

handle_abort(func)

Apply a decorator to gracefully handle any Abort errors that happen within the context.

Will log the error, dispatch it to Sentry, show a helpful message to the user about the error, and exit.

handle_authentication_error

handle_authentication_error(func)

Adapter decorator that catches AuthenticationError exceptions and raises an appropriate Abort error.

jobberappslib

Provide a stub module to maintain compatibility with previous versions.

Issue a deprecation warning when this module is imported from if JOBBERGATE_COMPATIBILITY_MODE is enabled.

If JOBBERGATE_COMPATIBILITY_MODE is not enabled, raise an import error when this module is imported.

logging

Provide initializers for logging.

init_logs

init_logs(verbose=False)

Initialize logging.

If JOBBERGATE_LOG_PATH is set in the config, add a rotating file log handler. Logs will be retained for 1 week.

If verbose is supplied, add a stdout handler at the DEBUG level.

init_sentry

init_sentry()

Initialize Sentry if the SENTRY_DSN environment variable is present.

main

Provide main entry point for the Jobbergate CLI App.

login

login(ctx: typer.Context)

Log in to the jobbergate-cli by storing the supplied token argument in the cache.

logout

logout(ctx: typer.Context)

Logs out of the jobbergate-cli. Clears the saved user credentials.

main

main(ctx: typer.Context, verbose: bool = typer.Option(False, help='Enable verbose logging to the terminal'), full: bool = typer.Option(False, help='Print all fields from CRUD commands'), raw: bool = typer.Option(False, help='Print output from CRUD commands as raw json'), version: bool = typer.Option(False, help='Print the version of jobbergate-cli and exit'), ignore_extra_args: str = typer.Option(None, '--username', '-u', '--password', '-p', hidden=True, help='Ignore extra arguments passed to the command for backward compatibility with the legacy app.'))

Welcome to the Jobbergate CLI!

More information can be shown for each command listed below by running it with the --help option.

safe_entrypoint

safe_entrypoint()

Entrypoint for the app including custom error handling.

With this we ensure error handling is applied to all commands with no need to duplicate the decorators on each of them.

show_token

show_token(ctx: typer.Context, plain: bool = typer.Option(False, help='Show the token in plain text.'), refresh: bool = typer.Option(False, help='Show the refresh token instead of the access token.'), show_prefix: bool = typer.Option(False, '--prefix', help="Include the 'Bearer' prefix in the output."), show_header: bool = typer.Option(False, '--header', help='Show the token as it would appear in a request header.'), decode: bool = typer.Option(False, '--decode', help='Show the content of the decoded access token.'))

Show the token for the logged in user.

Token output is automatically copied to your clipboard.

render

Provide helpers to render output for users.

StyleMapper

Provide a mapper that can set rich styles for rendered output of data tables and dicts.

The subapps have list endpoints that return sets of values. These are rendered as tables in the output. The StyleMapper class provides a way to simply define styles that should be applied to the columns of the table.

Example:

The following code will print a table where the columns are colored according to the style_mapper

.. code-block: python

style_mapper = StyleMapper( a="bold green", b="red", c="blue", ) envelope = dict( results=[ dict(a=1, b=2, c=3), dict(a=4, b=5, c=6), dict(a=7, b=8, c=9), ], pagination=dict(total=3) ) render_list_results(jb_ctx, envelope, style_mapper)

__init__

__init__(**colors: str)

Initialize the StyleMapper.

map_style

map_style(column: str) -> Dict[str, Any]

Map a column name from the table to display to the style that should be used to render it.

render_demo

render_demo(pre_amble: str | None = None)

Show the demo for the jobbergate-cli.

render_dict

render_dict(data: dict[str, Any], title: str = 'Data', hidden_fields: list[str] | None = None)

Render a dictionary in a rich Table that shows the key and value of each item.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	The dictionary to render.	required
title	str	The title header to include above the Table output.	'Data'
hidden_fields	list[str] | None	Keys that should be hidden in the Table output.	None

render_json

render_json(data: Any)

Print nicely formatted representation of a JSON serializable python primitive.

render_list_results

render_list_results(ctx: ContextProtocol, envelope: ListResponseEnvelope, style_mapper: StyleMapper | None = None, hidden_fields: list[str] | None = None, title: str = 'Results List')

Render a list of result data items in a rich Table.

Parameters:

Name	Type	Description	Default
ctx	ContextProtocol	The JobbergateContext. This is needed to detect if full or raw output is needed.	required
envelope	ListResponseEnvelope	A ListResponseEnvelope containing the data items.	required
style_mapper	StyleMapper | None	The style mapper that should be used to apply styles to the columns of the table.	None
hidden_fields	list[str] | None	Columns that should (if not using full mode) be hidden in the Table output.	None
title	str	The title header to include above the Table output.	'Results List'

render_single_result

render_single_result(ctx: ContextProtocol, result: dict[str, Any] | pydantic.BaseModel, hidden_fields: list[str] | None = None, title: str = 'Result', value_mappers: dict[str, Callable[[Any], Any]] | None = None)

Render a single data item in a rich Table.

Parameters:

Name	Type	Description	Default
ctx	ContextProtocol	The JobbergateContext. This is needed to detect if full or raw output is needed.	required
result	dict[str, Any] | BaseModel	The data item to display. May be a dict or a pydantic model.	required
hidden_fields	list[str] | None	Rows that should (if not using full mode) be hidden in the Table output.	None
title	str	The title header to include above the Table output.	'Result'
value_mappers	dict[str, Callable[[Any], Any]] | None	Mapping functions to change fields before rendering.	None

terminal_message

terminal_message(message: str, subject: str | None = None, color: str = 'green', footer: str | None = None, indent: bool = True)

Print a nicely formatted message as output to the user using a rich Panel.

Parameters:

Name	Type	Description	Default
message	str	The message to print out.	required
subject	str | None	An optional subject line to add in the header of the Panel.	None
color	str	An optional color to style the subject header with. Defaults to "green".	'green'
footer	str | None	An optional message to display in the footer of the Panel.	None
indent	bool	Adds padding to the left of the message. Defaults to True.	True

requests

Provide utilities for making requests against the Jobbergate API.

format_response_error

format_response_error(response: httpx.Response, default_text) -> str

Format a response into a human-readable error message, including the cause, and a possible solution.

get_possible_solution_to_error

get_possible_solution_to_error(response: httpx.Response) -> str

Get a possible solution to an error code.

make_request

make_request(client: httpx.Client, url_path: str, method: str, *, expected_status: int | None = None, expect_response: bool = True, abort_message: str = 'There was an error communicating with the API', abort_subject: str = 'REQUEST FAILED', support: bool = True, response_model_cls: Type[ResponseModel] | None = None, request_model: pydantic.BaseModel | None = None, save_to_file: Path | None = None, **request_kwargs: Any) -> ResponseModel | dict | int

Make a request against the Jobbergate API.

Parameters:

Name	Type	Description	Default
client	Client	The Httpx client to use for the request.	required
url_path	str	The path to add to the base url of the client where the request should be sent.	required
method	str	The REST method to use for the request (GET, PUT, UPDATE, POST, DELETE, etc).	required
expected_status	int | None	The status code to expect on the response. If it is not received, raise an Abort.	None
expect_response	bool	Indicates if response data (JSON) is expected from the API endpoint.	True
abort_message	str	The message to show the user if there is a problem and the app must be aborted.	'There was an error communicating with the API'
abort_subject	str	The subject to use in Abort output to the user.	'REQUEST FAILED'
support	bool	If true, add a message to the output instructing the user to seek help.	True
response_model_cls	Type[ResponseModel] | None	If supplied, serialize the response data into this Pydantic model class.	None
request_model	BaseModel | None	Use a pydantic model instance as the data body for the request.	None
save_to_file	Path | None	If supplied, save the response data to this file.	None
request_kwargs	Any	Any additional keyword arguments to pass to the request.	{}

Returns:

Type	Description
ResponseModel | dict | int	The response from the API, either as a Pydantic model, a dictionary, or an integer status code.

schemas

Provide Pydantic models for various data items.

ApplicationResponse

Bases: BaseModel

Describes the format of data for applications retrieved from the Jobbergate API endpoints.

ClusterCacheData

Bases: BaseModel

Describes the format of data stored in the clusters cache file.

ContextProtocol

Bases: Protocol

A protocol describing context passed from the main entry point.

It aims to help static type checkers at the same time that prevents circular import issues on the actual implementation.

DeviceCodeData

Bases: BaseModel

A model representing the data that is returned from the OIDC provider's device code endpoint.

IdentityData

Bases: BaseModel

A model representing the identifying data for a user from an auth token.

JobScriptCreateRequest

Bases: BaseModel

Request model for creating JobScript instances.

JobScriptFile

Bases: BaseModel

Model containing job-script files.

JobScriptRenderRequestData

Bases: BaseModel

Describes the data that will be sent to the create endpoint of the Jobbergate API for job scripts.

JobScriptResponse

Bases: BaseModel

Describes the format of data for job_scripts retrieved from the Jobbergate API endpoints.

null_files classmethod

null_files(value)

Remap a None value in files to an empty list.

JobSubmissionCreateRequestData

Bases: BaseModel

Describes the data that will be sent to the create endpoint of the Jobbergate API for job submissions.

JobSubmissionResponse

Bases: BaseModel

Describes the format of data for job_submissions retrieved from the Jobbergate API endpoints.

JobbergateApplicationConfig

Bases: BaseModel

A data object describing the config data needed to instantiate a JobbergateApplication class.

JobbergateConfig

Bases: BaseModel

A data object describing the config values needed in the "jobbergate_config" section of the JobbergateApplicationConfig model.

compute_extra_settings

compute_extra_settings(values)

Compute missing values and extra operations to enhance the user experience and backward compatibility.

ListResponseEnvelope

Bases: BaseModel, Generic[EnvelopeT]

A model describing the structure of response envelopes from "list" endpoints.

LocalApplication

Bases: BaseModel

Application retrieved from a local folder.

LocalTemplateFile

Bases: BaseModel

Template file retrieved from a local folder.

LocalWorkflowFile

Bases: BaseModel

Workflow file retrived from a local folder.

Persona

Bases: BaseModel

A model representing a pairing of a TokenSet and user email. This is a convenience to combine all of the identifying data and credentials for a given user.

RenderFromTemplateRequest

Bases: BaseModel

Request model for creating a JobScript entry from a template.

TokenSet

Bases: BaseModel

A model representing a pairing of access and refresh tokens

subapps

Subapps that are added to the base Typer application.

applications

Provide a sub-app for interacting with Applications data.

app

Provide a typer app that can interact with Application data in a cruddy manner.

clone

clone(ctx: typer.Context, id_or_identifier: Optional[str] = typer.Argument(None, help='The specific id or identifier of the application to be selected.'), id: Optional[int] = typer.Option(None, '--id', '-i', help=f'Alternative way to specify id. {ID_NOTE}'), identifier: Optional[str] = typer.Option(None, help=f'Alternative way to specify identifier. {IDENTIFIER_NOTE}'), application_identifier: Optional[str] = typer.Option(None, help='\n        Optional new application identifier to override the original.\n\n        Notice this can not match an existing identifier, including the one this entry is going to be cloned from.\n        '), application_desc: Optional[str] = typer.Option(None, help='Optional new application description to override the original'), application_name: Optional[str] = typer.Option(None, help='Optional new application name to override the original'))

Clone an application, so the user can own and modify a copy of it.

create

create(ctx: typer.Context, name: str = typer.Option(..., '--name', '-n', help='The name of the application to create'), identifier: Optional[str] = typer.Option(None, help=f'The human-friendly identifier of the application. {IDENTIFIER_NOTE}'), application_path: pathlib.Path = typer.Option(..., '--application-path', '-a', help='The path to the directory where the application files are located'), application_desc: Optional[str] = typer.Option(None, help='A helpful description of the application'))

Create a new application.

delete

delete(ctx: typer.Context, id_or_identifier: Optional[str] = typer.Argument(None, help='The specific id or identifier of the application to be selected.'), id: Optional[int] = typer.Option(None, '--id', '-i', help=f'Alternative way to specify id. {ID_NOTE}'), identifier: Optional[str] = typer.Option(None, help=f'Alternative way to specify identifier. {IDENTIFIER_NOTE}'))

Delete an existing application.

download_files

download_files(ctx: typer.Context, id_or_identifier: Optional[str] = typer.Argument(None, help='The specific id or identifier of the application to be selected.'), id: Optional[int] = typer.Option(None, '--id', '-i', help=f'Alternative way to specify id. {ID_NOTE}'), identifier: Optional[str] = typer.Option(None, help=f'Alternative way to specify identifier. {IDENTIFIER_NOTE}'))

Download the files from an application to the current working directory.

get_one

get_one(ctx: typer.Context, id_or_identifier: Optional[str] = typer.Argument(None, help='The specific id number or identifier of the application to be selected.'), id: Optional[int] = typer.Option(None, '--id', '-i', help=f'Alternative way to specify id. {ID_NOTE}'), identifier: Optional[str] = typer.Option(None, help=f'Alternative way to specify identifier. {IDENTIFIER_NOTE}'))

Get a single application by id or identifier

list_all

list_all(ctx: typer.Context, show_all: bool = typer.Option(False, '--all', help='Show all applications, even the ones without identifier'), user_only: bool = typer.Option(False, '--user', help='Show only applications owned by the current user'), search: Optional[str] = typer.Option(None, help='Apply a search term to results'), sort_order: SortOrder = typer.Option(SortOrder.DESCENDING, help='Specify sort order'), sort_field: Optional[str] = typer.Option('id', help='The field by which results should be sorted'))

Show available applications

update

update(ctx: typer.Context, id_or_identifier: Optional[str] = typer.Argument(None, help='The specific id or identifier of the application to be selected.'), id: Optional[int] = typer.Option(None, '--id', '-i', help=f'Alternative way to specify id. {ID_NOTE}'), identifier: Optional[str] = typer.Option(None, help=f'Alternative way to specify identifier. {IDENTIFIER_NOTE}'), application_path: Optional[pathlib.Path] = typer.Option(None, '--application-path', '-a', help='The path to the directory where the application files are located'), update_identifier: Optional[str] = typer.Option(None, help='Optional new application identifier to be set'), application_desc: Optional[str] = typer.Option(None, help='Optional new application description to be set'), application_name: Optional[str] = typer.Option(None, help='Optional new application name to be set'))

Update an existing application.

application_base

ApplicationBase.

JobbergateApplicationBase

JobbergateApplicationBase.

__init__

__init__(jobbergate_yaml: Dict[str, Any])

Initialize class attributes.

find_templates staticmethod

find_templates(application_path: pathlib.Path) -> List[pathlib.Path]

Finds templates a given application path.

mainflow

mainflow(data: Dict[str, Any])

Implements the main question asking workflow.

application_helpers

Helper functions that may be used inside of Jobbergate applications.

get_file_list

get_file_list(path=None, search_term='*.*')

Return a list of input files in a directory that match a search term.

Ignore casing when comparing against the search term.

Default to searching for all files in the current directory.

get_running_jobs

get_running_jobs(user_only=True)

Return a list of the user's currently running jobs, as given by SLURM's squeue command.

The format returned is: [job ID, 8 chars] [job name]

questions

Abstraction layer for questions. Each class represents different question types.

The questions describe literal questions that are asked of the user in an interactive mode via the inquirer package.

Questions will be skipped and use the default value if the ignore property resolves to True.

Questions will also resolve to their default values if running in "fast mode".

BooleanList

Bases: Confirm

Asks a confirmation question that is followed up by a certain question list when true and a different list if false.

__init__

__init__(variablename: str, message: str, whentrue=None, whenfalse=None, **kwargs)

Initialize the Checkbox question.

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required
message	str	The message to show the user that describes what the question is gathering.	required
whentrue		List of questions to ask if the user answers 'true' on this question.	None
whenfalse		List of questions to show if the user answers 'false' on this question.	None

ignore_child

ignore_child(child: QuestionBase, answers: dict[str, Any]) -> bool

Dynamically check if a child question should be ignored based on the questions that have already been answered.

Parameters:

Name	Type	Description	Default
child	QuestionBase	The child question that might be ignored.	required
answers	dict[str, Any]	Answer values to previously asked questions.	required

make_ignore_partial

make_ignore_partial(child: QuestionBase) -> Callable[[Dict[str, Any]], bool]

Build a partial method for checking if a child should be ignored.

This method just makes the code more readable so that a non-descriptive lambda does not need to be used inline.

make_prompts

make_prompts(**override_kwargs)

Create inquirer prompts from this instance of BooleanList and for all its child questions.

Parameters:

Name	Type	Description	Default
override_kwargs		A collection of keyword arguments to override in the base make_prompts method.	{}

Checkbox

Bases: QuestionBase

Gives the user a list to choose multiple entries from.

__init__

__init__(variablename: str, message: str, choices: list, **kwargs)

Initialize the Checkbox question.

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required
message	str	The message to show the user that describes what the question is gathering.	required
choices	list	A list of the possible values from which the Question will allow the user to select many.	required

Confirm

Bases: QuestionBase

Asks a question with a boolean answer (true/false).

__init__

__init__(variablename: str, message: str, **kwargs)

Initialize the Confirm question.

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required
message	str	The message to show the user that describes what the question is gathering.	required

Const

Bases: Text

Sets the variable to the default value. Doesn't show anything.

__init__

__init__(variablename: str, **kwargs)

Initialize the Const "question".

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required

make_prompts

make_prompts()

Create inquirer prompts from this instance of Const.

Directory

Bases: QuestionBase

Asks for a directory name. If exists is True, it checks if the path exists and is a directory.

Parameters:

Name	Type	Description	Default
exists	bool | None	Checks if the given directory exists.	None

__init__

__init__(variablename: str, message: str, exists: bool | None = None, **kwargs)

Initialize the Directory question.

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required
message	str	The message to show the user that describes what the question is gathering.	required
exists	bool | None	If True, ensure that the directory exists on the system.	None

File

Bases: QuestionBase

Asks for a file name.

__init__

__init__(variablename: str, message: str, exists: bool | None = None, **kwargs)

Initialize the File question.

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required
message	str	The message to show the user that describes what the question is gathering.	required
exists	bool | None	If True, ensure that the file path exists on the system.	None

Integer

Bases: QuestionBase

Asks for an integer value. Could have min and/or max constrains.

__init__

__init__(variablename: str, message: str, minval: int | None = None, maxval: int | None = None, **kwargs)

Initialize the Integer question.

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required
message	str	The message to show the user that describes what the question is gathering.	required
minval	int | None	The minimum value the integer may be set to. If not specified, use negative infinity.	None
maxval	int | None	The maximum value the integer may be set to. If not specified, use infinity.	None

List

Bases: QuestionBase

Gives the user a list to choose one from.

__init__

__init__(variablename: str, message: str, choices: list, **kwargs)

Initialize the List question.

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required
message	str	The message to show the user that describes what the question is gathering.	required
choices	list	A list of the possible values from which the Question will allow the user to select one.	required

QuestionBase

Baseclass for questions.

All questions have variablename, message and an optional default.

__init__

__init__(variablename: str, message: str, ignore: bool = False, default: Any | None = None, inquirer_type: Type[TInquirerType] = inquirer.Text)

Initialize the Question.

Parameters:

Name	Type	Description	Default
variablename	str	The key in the config dictionary that this question will set.	required
message	str	The message to show the user that describes what the question is gathering.	required
ignore	bool	If true, do not ask the question and just use the default value instead.	False
default	Any | None	The default value for the variablename in the answers dict.	None
inquirer_type	Type[TInquirerType]	The inquirer question type that this QuestionBase wraps.	Text

make_prompts

make_prompts(**override_kwargs)

Create inquirer prompts from this instance of QuestionBase.

Parameters:

Name	Type	Description	Default
override_kwargs		A collection of keyword arguments to override in initializing the inquirer question.	{}

Text

Bases: QuestionBase

Asks for a text value.

tools

Provide tool functions for working with Application data.

ApplicationRuntime dataclass

Prepare and execute a Jobbergate application gathering the answers to the questions.

Parameters:

Name	Type	Description	Default
app_data	Union[ApplicationResponse, LocalApplication]	The application data, can be either an ApplicationResponse or a LocalApplication.	required
app_source_code	str	The source code of the application, often coming from jobbergate.py file.	required
supplied_params	Dict[str, Any]	The parameters supplied to the application, defaults to an empty dictionary.	field(default_factory=dict)
fast_mode	bool	A flag indicating whether the application is in fast mode, defaults to False.	False

as_flatten_param_dict

as_flatten_param_dict() -> Dict[str, Any]

Flatten the internal data to support the rendering process.

execute_application

execute_application()

Execute the jobbergate application python module.

fetch_application_data

fetch_application_data(jg_ctx: ContextProtocol, id_or_identifier: int | str) -> ApplicationResponse

Retrieve an application from the API by id or identifier.

Parameters:

Name	Type	Description	Default
jg_ctx	ContextProtocol	The JobbergateContext. Needed to access the Httpx client with which to make API calls.	required
id		The id of the application to fetch.	required
identifier		If supplied, look for an application instance with the provided identifier.	required

Returns:

Type	Description
ApplicationResponse	An instance of ApplicationResponse containing the application data.

fetch_application_data_locally

fetch_application_data_locally(application_path: pathlib.Path) -> LocalApplication

Retrieve an application from a local directory.

Parameters:

Name	Type	Description	Default
application_path	Path	The directory containing the application files.	required

Returns:

Type	Description
LocalApplication	A LocalApplication instance containing the application data.

get_upload_files

get_upload_files(application_path: pathlib.Path)

Context manager to build the files parameter.

Open the supplied file(s) and build a files param appropriate for using multi-part file uploads with the client.

Parameters:

Name	Type	Description	Default
application_path	Path	The directory where the application files are located.	required

load_application_config_from_source

load_application_config_from_source(config_source: str) -> JobbergateApplicationConfig

Load the JobbergateApplicationConfig from a text string containing the config as YAML.

Parameters:

Name	Type	Description	Default
config_source	str	The YAML containing the config	required

Returns:

Type	Description
JobbergateApplicationConfig	A JobbergateApplicationConfig instance with the config values

load_application_data

load_application_data(app_data: Union[ApplicationResponse, LocalApplication], application_source_file: str) -> Tuple[JobbergateApplicationConfig, JobbergateApplicationBase]

Validates and loads the data for an application returned from the API's applications GET endpoint.

As part of the Jobbergate data restructure, sections of the legacy jobbergate.yaml are now stored in different tables in the backend. This function reconstructs them from app_data.workflow_file.runtime_config and app_data.template_vars for backward compatibility.

Parameters:

Name	Type	Description	Default
app_data	Union[ApplicationResponse, LocalApplication]	A dictionary containing the application data.	required
application_source_file	str	The source file of the application.	required

Returns:

Type	Description
Tuple[JobbergateApplicationConfig, JobbergateApplicationBase]	A tuple containing the application config and the application module.

load_application_from_source

load_application_from_source(app_source: str, app_config: JobbergateApplicationConfig) -> JobbergateApplicationBase

Load the JobbergateApplication class from a text string containing the source file.

Creates the module in a temporary file and imports it with importlib.

Adapted from: https://docs.python.org/3/library/importlib.html#importing-a-source-file-directly

Parameters:

Name	Type	Description	Default
app_source	str	The JobbergateApplication source code to load	required
app_config	JobbergateApplicationConfig	The JobbergateApplicationConfig needed to instantiate the JobbergateApplication	required

load_default_config

load_default_config() -> Dict[str, Any]

Load the default config for an application.

save_application_files

save_application_files(jg_ctx: ContextProtocol, application_data: ApplicationResponse, destination_path: pathlib.Path) -> List[pathlib.Path]

Save the application files from the API response into a local destination.

upload_application

upload_application(jg_ctx: ContextProtocol, application_path: pathlib.Path, id_or_identifier: int | str)

Upload an application given an application path and the application id.

Parameters:

Name	Type	Description	Default
jg_ctx	ContextProtocol	The JobbergateContext. Needed to access the Httpx client with which to make API calls.	required
application_path	Path	The directory where the application files to upload may be found.	required
application_id		The id of the application for which to upload data.	required
application_identifier		The identifier of the application for which to upload data.	required

clusters

Provide a sub-app for interacting with Cluster data.

app

Provide a typer app that can interact with Cluster data in a cruddy manner.

list_all

list_all(ctx: typer.Context)

Show available clusters

tools

Provide tool functions for working with Cluster data

job_scripts

Provide a sub-app for interacting with Job Script data.

app

Provide a typer app that can interact with Job Script data in a cruddy manner.

clone

clone(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The specific id of the job script to be updated.'), id_option: Optional[int] = typer.Option(None, '--id', '-i', help='Alternative way to specify id.'), name: Optional[str] = typer.Option(None, help='Optional new name of the job script.'), description: Optional[str] = typer.Option(None, help='Optional new text describing the job script.'))

Clone an existing job script, so the user can own and modify a copy of it.

create

create(ctx: typer.Context, id_or_identifier: Optional[str] = typer.Argument(None, help='The specific id or identifier of the application from which to create the job script.'), name: Optional[str] = typer.Option(None, '--name', '-n', help=dedent('\n            The name of the job script to create.\n            If this is not supplied, the name will be derived from the base application.\n            ')), application_id: Optional[int] = typer.Option(None, '--application-id', '-i', help='Alternative way to specify the application id.'), application_identifier: Optional[str] = typer.Option(None, help='Alternative way to specify the application identifier.'), description: Optional[str] = typer.Option(None, help='Optional text describing the job script.'), sbatch_params: Optional[List[str]] = typer.Option(None, help='Optional parameter to submit raw sbatch parameters.'), param_file: Optional[pathlib.Path] = typer.Option(None, help=dedent('\n            Supply a json file that contains the parameters for populating templates.\n            If this is not supplied, the question asking in the application is triggered.\n            ')), fast: bool = typer.Option(False, '--fast', '-f', help='Use default answers (when available) instead of asking the user.'), download: Optional[bool] = typer.Option(None, help='Download the job script files to the current working directory'), submit: Optional[bool] = typer.Option(None, help='Do not ask the user if they want to submit a job.'), cluster_name: Optional[str] = typer.Option(None, help="The name of the cluster where the job should be submitted to (i.g. 'nash-staging')"), execution_directory: Optional[pathlib.Path] = typer.Option(None, help=dedent('\n            The path on the cluster where the job script should be executed.\n            If provided as a relative path, it will be converted as an absolute path from your current\n            working directory. If you use "~" to denote your home directory, the path will be expanded to an\n            absolute path for your home directory on *this* machine.\n            ').strip()))

Create a new job script from an application.

create_locally

create_locally(ctx: typer.Context, application_path: pathlib.Path = typer.Argument(pathlib.Path('.'), help='The path to the application directory to use as a template for the job script.', dir_okay=True), job_script_name: str = typer.Option('job_script', help='The name of the job script to render locally.'), output_path: pathlib.Path = typer.Option(pathlib.Path('.'), help='The path to the directory where the rendered job script should be saved.', dir_okay=True), sbatch_params: Optional[List[str]] = typer.Option(None, help='Optional parameter to submit raw sbatch parameters.'), param_file: Optional[pathlib.Path] = typer.Option(None, help=dedent('\n            Supply a json file that contains the parameters for populating templates.\n            If this is not supplied, the question asking in the application is triggered.\n            ')), fast: bool = typer.Option(False, '--fast', '-f', help='Use default answers (when available) instead of asking the user.'))

Create a job-script from local application files (ideal for development and troubleshooting).

The templates will be overwritten with the rendered files.

create_stand_alone

create_stand_alone(ctx: typer.Context, job_script_path: Optional[pathlib.Path] = typer.Argument(None, help='The path to the job script file to upload', file_okay=True, readable=True), name: str = typer.Option(..., help=dedent('The name of the job script to create.')), description: Optional[str] = typer.Option(None, help='Optional text describing the job script.'), job_script_path_option: Optional[pathlib.Path] = typer.Option(None, '--job-script-path', help='The path to the job script file to upload', file_okay=True, readable=True), supporting_file_path: Optional[List[pathlib.Path]] = typer.Option(None, '--supporting-file', '-s', help='A path for one of the supporting files to upload', file_okay=True, readable=True))

Create and upload files for a standalone job script (i.e., unrelated to any application).

delete

delete(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The id of the job script to delete'), id_option: Optional[int] = typer.Option(None, '--id', '-i', help='Alternative way to specify the job script id'))

Delete an existing job script.

download_files

download_files(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The specific id of the job script to be downloaded.'), id_option: Optional[int] = typer.Option(None, '--id', '-i', help='Alternative way to specify id.'))

Download the files from a job script to the current working directory.

get_one

get_one(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The specific id of the job script to be selected.'), id_option: Optional[int] = typer.Option(None, '--id', '-i', help='Alternative way to specify id.'))

Get a single job script by id.

list_all

list_all(ctx: typer.Context, show_all: bool = typer.Option(False, '--all', help='Show all job scripts, even the ones owned by others'), search: Optional[str] = typer.Option(None, help='Apply a search term to results'), sort_order: SortOrder = typer.Option(SortOrder.DESCENDING, help='Specify sort order'), sort_field: Optional[str] = typer.Option('id', help='The field by which results should be sorted'), from_application_id: Optional[int] = typer.Option(None, help='Filter job-scripts by the application-id they were created from.'))

Show available job scripts

show_files

show_files(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The specific id of the job script to be cloned.'), id_option: Optional[int] = typer.Option(None, '--id', '-i', help='Alternative way to specify id.'), plain: bool = typer.Option(False, help='Show the files in plain text.'))

Show the files for a single job script by id.

update

update(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The id of the job script to update'), id_option: Optional[int] = typer.Option(None, '--id', '-i', help='Alternative way to specify the id of the job script to update'), name: Optional[str] = typer.Option(None, help='Optional new name of the job script.'), description: Optional[str] = typer.Option(None, help='Optional new text describing the job script.'))

Update an existing job script.

tools

Provide tool functions for working with Job Script data

download_job_script_files

download_job_script_files(id: int, jg_ctx: ContextProtocol, destination_path: pathlib.Path) -> List[JobScriptFile]

Download all job script files from the API and save them to the destination path.

fetch_job_script_data

fetch_job_script_data(jg_ctx: ContextProtocol, id: int) -> JobScriptResponse

Retrieve a job_script from the API by id

get_template_output_name_mapping

get_template_output_name_mapping(config: JobbergateConfig, job_name: str) -> Dict[str, str]

Get the mapping of template names to output names.

This provides the mapping as expected by the API v4 from the configuration on CLI v3.

inject_sbatch_params

inject_sbatch_params(job_script_data_as_string: str, sbatch_params: List[str]) -> str

Inject sbatch params into job script.

Given the job script as job_script_data_as_string, inject the sbatch params in the correct location.

question_helper

question_helper(question_func: Callable, text: str, default: Any, fast: bool, actual_value: Optional[Any])

Helper function for asking questions to the user.

Parameters:

Name	Type	Description	Default
question_func	Callable	The function to use to ask the question	required
text	str	The text of the question to ask	required
default	Any	The default value to use if the user does not provide one	required
fast	bool	Whether to use default answers (when available) instead of asking the user	required
actual_value	Optional[Any]	The actual value provided by the user, if any	required

Returns:

Type	Description
	actual_value or default or the value provided by the user

Notes

The actual_value has the most priority and will be returned if it is not None. After evaluating the actual_value, the fast mode will determine if the default value will be used. Otherwise, the question will be prompted to the user.

remove_prefix

remove_prefix(s: str) -> str

Remove the prefix 'templates/' from a string

remove_prefix_suffix

remove_prefix_suffix(s: str) -> str

Remove the prefix 'templates/' and suffixes '.j2' and '.jinja2' from a string

render_job_script

render_job_script(jg_ctx: ContextProtocol, id_or_identifier: int | str, name: Optional[str] = None, description: Optional[str] = None, sbatch_params: Optional[List[str]] = None, param_file: Optional[pathlib.Path] = None, fast: bool = False) -> JobScriptResponse

Render a new job script from an application.

Parameters:

Name	Type	Description	Default
jg_ctx	ContextProtocol	The Jobbergate context.	required
name	Optional[str]	Name of the new job script.	None
application_id		Id of the base application.	required
application_identifier		Identifier of the base application.	required
description	Optional[str]	Description of the new job script.	None
sbatch_params	Optional[List[str]]	List of sbatch parameters.	None
param_file	Optional[Path]	Path to a parameters file.	None
fast	bool	Whether to use default answers (when available) instead of asking the user.	False

Returns:

Type	Description
JobScriptResponse	The new job script.

render_job_script_locally

render_job_script_locally(jg_ctx: ContextProtocol, job_script_name: str, application_path: pathlib.Path, output_path: pathlib.Path, sbatch_params: Optional[List[str]] = None, param_file: Optional[pathlib.Path] = None, fast: bool = False)

Render a new job script from an application in a local directory.

Parameters:

Name	Type	Description	Default
jg_ctx	ContextProtocol	The Jobbergate context.	required
job_script_name	str	Name of the new job script.	required
application_path	Path	Path to the base application.	required
output_path	Path	Path to the output the rendered job script.	required
sbatch_params	Optional[List[str]]	List of sbatch parameters.	None
param_file	Optional[Path]	Path to a parameters file.	None
fast	bool	Whether to use default answers (when available) instead of asking the user.	False

render_template

render_template(template_path: pathlib.Path, parameters: Dict[str, Any]) -> str

Render a template file and save it to the output directory.

Parameters:

Name	Type	Description	Default
template_path	Path	The path to the template file.	required
parameters	Dict[str, Any]	The parameters to use for rendering the template.	required

Returns:

Type	Description
str	The rendered template as a string.

save_job_script_file

save_job_script_file(jg_ctx: ContextProtocol, destination_path: pathlib.Path, job_script_file: JobScriptFile) -> pathlib.Path

Save a job script file from the API response to the destination path.

upload_job_script_files

upload_job_script_files(jg_ctx: ContextProtocol, job_script_id: int, job_script_path: pathlib.Path, supporting_file_paths: list[pathlib.Path] | None = None)

Upload a job-script and its supporting files given their paths and the job-script id.

Parameters:

Name	Type	Description	Default
jg_ctx	ContextProtocol	The JobbergateContext needed to access the Httpx client with which to make API calls	required
job_script_path	Path	The path to the job-script file to upload	required
supporting_file_paths	list[Path] | None	The paths to any supporting files to upload with the job-scritpt	None
job_script_id	int	The id of the job-script for which to upload data	required

validate_parameter_file

validate_parameter_file(parameter_path: pathlib.Path) -> Dict[str, Any]

Validate parameter file at the supplied path and returns the parsed dict.

Parameters:

Name	Type	Description	Default
parameter_path	Path	Path to the parameter file.	required

Raises:

Type	Description
Abort	If file does not exist or is not valid JSON.

Returns:

Type	Description
Dict[str, Any]	Parsed dictionary from the parameter file.

job_submissions

Provide a sub-app for interacting with Job Submission data.

app

Provide a typer app that can interact with Job Submission data in a cruddy manner.

clone

clone(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The specific id of the job script to be updated.'), id_option: Optional[int] = typer.Option(None, '--id', '-i', help='Alternative way to specify id.'))

Clone an existing job submission under the CREATED status, so it is re-submitted to the cluster.

create

create(ctx: typer.Context, job_script_id: Optional[int] = typer.Argument(None, help='The id of the job_script from which to create the job submission'), name: str = typer.Option(..., '--name', '-n', help='The name of the job submission to create'), description: Optional[str] = typer.Option(None, help='A helpful description of the job submission'), job_script_id_option: Optional[int] = typer.Option(None, '--job-script-id', '-i', help='Alternative way to specify the job script id'), cluster_name: str = typer.Option(None, help="The name of the cluster where the job should be submitted (i.g. 'nash-staging')"), execution_directory: Optional[Path] = typer.Option(None, help=dedent('\n            The path on the cluster where the job script should be executed.\n            If provided as a relative path, it will be converted as an absolute path from your current\n            working directory. If you use "~" to denote your home directory, the path will be expanded to an\n            absolute path for your home directory on *this* machine.\n            ').strip()), sbatch_arguments: Optional[list[str]] = typer.Option(None, '--sbatch-arguments', '-s', help=dedent('\n            Additional arguments to pass as sbatch directives. These should be provided as a list of strings.\n            See more details at: https://slurm.schedmd.com/sbatch.html\n            ').strip()), download: bool = typer.Option(False, help='Download the job script files to the current working directory'))

Create a new job submission.

delete

delete(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The id of the job submission to delete'), id_option: Optional[int] = typer.Option(..., '--id', '-i', help='Alternative way to specify id'))

Delete an existing job submission.

get_one

get_one(ctx: typer.Context, id: Optional[int] = typer.Argument(None, help='The specific id of the job submission to be selected.'), id_option: Optional[int] = typer.Option(None, '--id', '-i', help='Alternative way to specify id'))

Get a single job submission by id

list_all

list_all(ctx: typer.Context, show_all: bool = typer.Option(False, '--all', help='Show all job submissions, even the ones owned by others'), search: Optional[str] = typer.Option(None, help='Apply a search term to results'), sort_order: SortOrder = typer.Option(SortOrder.DESCENDING, help='Specify sort order'), sort_field: Optional[str] = typer.Option('id', help='The field by which results should be sorted'), from_job_script_id: Optional[int] = typer.Option(None, help='Filter job-submissions by the job-script-id they were created from.'))

Show available job submissions.

tools

Provide tool functions for working with Job Submission data

JobSubmissionABC dataclass

Bases: ABC

A dataclass representing a job submission for Jobbergate.

Parameters:

Name	Type	Description	Default
jg_ctx	ContextProtocol	The JobbergateContext. Used to retrieve the client for requests and the email of the submitting user.	required
job_script_id	int	The id of the Job Script to submit to Slurm.	required
name	str	The name to attach to the Job Submission.	required
description	Optional[str]	An optional description that may be added to the Job Submission.	None
cluster_name	str | None	An optional cluster_name for the cluster where the job should be executed. If left off, it will default to the DEFAULT_CLUSTER_NAME from the settings. If no default is set, an exception will be raised.	None
execution_directory	Path | None	An optional directory where the job should be executed. If provided as a relative path, it will be constructed as an absolute path relative to the current working directory.	None
download	bool	A flag indicating whether the job script files should be downloaded to the execution directory.	False
sbatch_arguments	Optional[list[str]]	An optional list of arguments to pass to inject into the job script.	None

__post_init__

__post_init__()

Post-init hook to ensure that the cluster_name and execution_directory are set correctly.

get_request_data

get_request_data() -> JobSubmissionCreateRequestData

Return the data to be used in the POST request to the API.

make_post_request

make_post_request(job_submission_data: JobSubmissionCreateRequestData) -> JobSubmissionResponse

Make the POST request to the API to create the job submission.

process_submission abstractmethod

process_submission()

Process the job submission. This method should be overridden by subclasses.

run

run() -> JobSubmissionResponse

Run all required steps to create the job submission.

OnsiteJobSubmission

Bases: JobSubmissionABC

get_request_data

get_request_data() -> JobSubmissionCreateRequestData

Return the data to be used in the POST request to the API.

inject_sbatch_params

inject_sbatch_params(job_script_path: Path)

Inject sbatch parameters into the job script.

process_submission

process_submission()

Process the job submission in on-site mode.

RemoteJobSubmission

Bases: JobSubmissionABC

process_submission

process_submission()

Process the job submission in remote mode.

fetch_job_submission_data

fetch_job_submission_data(jg_ctx: ContextProtocol, job_submission_id: int) -> JobSubmissionResponse

Retrieve a job submission from the API by id

job_submissions_factory

job_submissions_factory(jg_ctx: ContextProtocol, job_script_id: int, name: str, execution_directory: Path | None = None, cluster_name: str | None = None, download: bool = False, description: Optional[str] = None, sbatch_arguments: Optional[list[str]] = None) -> JobSubmissionABC

Job submission factory function. Returns the correct job submission class based on the current mode.

pagination

tools

Tools for the subapps.

resolve_application_selection

resolve_application_selection(id_or_identifier: int | str | None = None, id: int | None = None, identifier: str | None = None, prefix: str = '') -> int | str

Resolve the application selection parameters.

resolve_selection

resolve_selection(*args: SelectionType | None, option_name: str = 'id') -> SelectionType

Resolve the multiple optional selection parameters to ensure one value is indeed selected.

text_tools

Provide some basic tools for manipulating text.

conjoin

conjoin(*items: str, join_str: str = '\n') -> str

Joins strings supplied as args.

Helper that wraps str.join() without having to pack strings in an iterable.

copy_to_clipboard

copy_to_clipboard(text: str) -> bool

Copy the provided text to the clipboard.

If the clipboard is not available, return False. Otherwise, return True.

dedent

dedent(text: str) -> str

Dedents a paragraph after removing leading and trailing whitespace.

dedent_all

dedent_all(*texts: str, join_str: str = '\n') -> str

Dedents each blob supplied as an argument and then joins them.

indent

indent(text: str, prefix: str = '    ', **kwargs) -> str

Simple wrapper for the textwrap.indent() method but includes a default prefix.

unwrap

unwrap(text: str) -> str

Unwraps a paragraph of text into a single line.

The text may be indented.