Jobbergate Core Reference

jobbergate_core

Jobbergate-core contains key components that are shared among sub-projects.

AuthenticationError

Bases: Buzz

Base exception for errors related to authentication on Jobbergate.

JobbergateAuthHandler `dataclass`

High-level class used to manage authentication to requests to the Jobbergate-API

After an instance of this class is created, it can be used to authenticate requests from both requests and httpx packages by passing it to the auth parameter on the request (see examples below).

It just works out of the box. Behind the scenes, this procedure calls the :meth:JobbergateAuthHandler.acquire_access method to load the available tokens from the cache directory, it tries to refresh them if they are expired, or provides an URL to the user to login on the system.

Notice all steps above are also available individually as public methods, allowing a fine control for advanced users.

.. _requests: https://requests.readthedocs.io/en/latest/ .. _httpx: https://www.python-httpx.org/

Parameters:

Name	Type	Description	Default
`cache_directory`	`Path`	Directory to be used for the caching tokens.	required
`login_domain`	`str`	Domain used for the login.	required
`login_client_id`	`str`	Client ID used for login.	`'default'`

Note

These values depend on the identity provider used for authentication. Consult your system administrator or contact Omnivector support support@omnivector.solutions for further assistance.

Note

This class can interoperate with the tokens generated by the jobbergate-cli package, as long as they are stored in the same cache directory.

Examples:

The following example shows how to use the :meth:`JobbergateAuthHandler`
class to authenticate a request:

>>> from pathlib import Path
>>> import requests
>>> from jobbergate_core import JobbergateAuthHandler
>>> jobbergate_auth = JobbergateAuthHandler(
...     cache_directory=Path("."),
...     login_domain="http://keycloak.local:8080/realms/jobbergate-local",
...     login_client_id="cli",
... )
>>> jobbergate_base_url = "http://localhost:8000/jobbergate"
>>> response = requests.get(
...     f"{jobbergate_base_url}/applications",
...     auth=jobbergate_auth # this is the important part
)
Login Here: http://keycloak.local:8080/realms/jobbergate-local/device?user_code=LMVJ-XOLG
>>> response.raise_for_status()
>>> print(f"response = {response.json()}")

call

__call__(request)

This internal method allows the integration with the requests library.

It is called automatically when the instance is passed to the auth parameter, see the examples in the class docstring.

It adds the Authorization header to the request with the access token.

acquire_access

acquire_access() -> str

High-level method to acquire a valid access token.

This method will attempt, in order:

Use the internal access token from the instance
Load the tokens from the cache directory (see :meth:JobbergateAuthHandler.load_from_cache)
If the access token is unavailable or expired, refresh both tokens using the refresh token grant type (see :meth:JobbergateAuthHandler.refresh_tokens)
If the refresh token is unavailable or expired, login to generate both tokens (see :meth:JobbergateAuthHandler.login)

Returns:

Type	Description
`str`	The bearer access token.

Raises:

Type	Description
`AuthenticationError`	If all of the steps above fail to acquire a valid access token.

get_access_from_secret

get_access_from_secret() -> None

Get the access token using the client secret.

load_from_cache

load_from_cache() -> None

Load the tokens that are available at the cache directory.

login() -> None

Login to Jobbergate.

An URL will be printed to the console, the user must open it in a browser and provide their access credentials.

After the login is completed, the tokens will be saved to the cache directory.

logout

logout() -> None

Logout from Jobbergate by clearing the loaded tokens and their cache on the disk.

refresh_tokens

refresh_tokens() -> None

Refresh the tokens.

After the refresh operation is completed, the tokens will be saved to the cache directory.

Raises:

Type	Description
`AuthenticationError`	If the refresh token is missing or expired.

save_to_cache

save_to_cache() -> None

Save the tokens to the cache directory.

Note

This method will create the cache directory if it does not exist.

TokenError

Bases: AuthenticationError

Exception for errors related to tokens on Jobbergate.

auth

Utilities for handling auth in Jobbergate.

AuthenticationError

Bases: Buzz

Base exception for errors related to authentication on Jobbergate.

JobbergateAuthHandler `dataclass`

High-level class used to manage authentication to requests to the Jobbergate-API

After an instance of this class is created, it can be used to authenticate requests from both requests and httpx packages by passing it to the auth parameter on the request (see examples below).

It just works out of the box. Behind the scenes, this procedure calls the :meth:JobbergateAuthHandler.acquire_access method to load the available tokens from the cache directory, it tries to refresh them if they are expired, or provides an URL to the user to login on the system.

Notice all steps above are also available individually as public methods, allowing a fine control for advanced users.

.. _requests: https://requests.readthedocs.io/en/latest/ .. _httpx: https://www.python-httpx.org/

Parameters:

Name	Type	Description	Default
`cache_directory`	`Path`	Directory to be used for the caching tokens.	required
`login_domain`	`str`	Domain used for the login.	required
`login_client_id`	`str`	Client ID used for login.	`'default'`

Note

These values depend on the identity provider used for authentication. Consult your system administrator or contact Omnivector support support@omnivector.solutions for further assistance.

Note

This class can interoperate with the tokens generated by the jobbergate-cli package, as long as they are stored in the same cache directory.

Examples:

The following example shows how to use the :meth:`JobbergateAuthHandler`
class to authenticate a request:

>>> from pathlib import Path
>>> import requests
>>> from jobbergate_core import JobbergateAuthHandler
>>> jobbergate_auth = JobbergateAuthHandler(
...     cache_directory=Path("."),
...     login_domain="http://keycloak.local:8080/realms/jobbergate-local",
...     login_client_id="cli",
... )
>>> jobbergate_base_url = "http://localhost:8000/jobbergate"
>>> response = requests.get(
...     f"{jobbergate_base_url}/applications",
...     auth=jobbergate_auth # this is the important part
)
Login Here: http://keycloak.local:8080/realms/jobbergate-local/device?user_code=LMVJ-XOLG
>>> response.raise_for_status()
>>> print(f"response = {response.json()}")

call

__call__(request)

This internal method allows the integration with the requests library.

It is called automatically when the instance is passed to the auth parameter, see the examples in the class docstring.

It adds the Authorization header to the request with the access token.

acquire_access

acquire_access() -> str

High-level method to acquire a valid access token.

This method will attempt, in order:

Use the internal access token from the instance
Load the tokens from the cache directory (see :meth:JobbergateAuthHandler.load_from_cache)
If the access token is unavailable or expired, refresh both tokens using the refresh token grant type (see :meth:JobbergateAuthHandler.refresh_tokens)
If the refresh token is unavailable or expired, login to generate both tokens (see :meth:JobbergateAuthHandler.login)

Returns:

Type	Description
`str`	The bearer access token.

Raises:

Type	Description
`AuthenticationError`	If all of the steps above fail to acquire a valid access token.

get_access_from_secret

get_access_from_secret() -> None

Get the access token using the client secret.

load_from_cache

load_from_cache() -> None

Load the tokens that are available at the cache directory.

login() -> None

Login to Jobbergate.

An URL will be printed to the console, the user must open it in a browser and provide their access credentials.

After the login is completed, the tokens will be saved to the cache directory.

logout

logout() -> None

Logout from Jobbergate by clearing the loaded tokens and their cache on the disk.

refresh_tokens

refresh_tokens() -> None

Refresh the tokens.

After the refresh operation is completed, the tokens will be saved to the cache directory.

Raises:

Type	Description
`AuthenticationError`	If the refresh token is missing or expired.

save_to_cache

save_to_cache() -> None

Save the tokens to the cache directory.

Note

This method will create the cache directory if it does not exist.

Token `dataclass`

Low-level class used to handle tokens.

Parameters:

Name	Type	Description	Default
`cache_directory`	`Path`	The directory used for cache.	required
`label`	`str`	The type of token.	required
`content`	`str`	The content of the token (default is `""`).	`''`

Attributes:

Name	Type	Description
`file_path`	`Path`	The path to the file associated with the token. It is computed as `<cache_directory>/<label>.token`.
`data`	`TokenData`	Metadata decoded from the token's content are available in this dictionary. Expiration date and permissions are some examples of data that can be found.

bearer_token `property`

bearer_token: str

Return the token with the Bearer prefix.

__post_init__

__post_init__()

Post init method.

clear_cache

clear_cache() -> None

Clear the token from cache by removing the file associated with it.

is_expired

is_expired() -> bool

Check if the token is expired.

Returns:

Type	Description
`bool`	True if the token is expired, False otherwise.

Raises:

Type	Description
`TokenError`	If the expiration date is not found.

is_valid

is_valid() -> bool

Verify if the token is valid, i.e., has content and is not expired.

load_from_cache

load_from_cache() -> Token

Load the token from the cache directory.

Parameters:

Name	Type	Description	Default
`cache_directory`		The path to the cache directory.	required
`label`		The type of token.	required

Returns:

Type	Description
`Token`	A new token with the content replaced.

replace

replace(**changes) -> Token

Create a new instance of the token with the changes applied.

Other Parameters:

Name	Type	Description
`content`		The content of the token.
`cache_directory`		The directory containing the cache.
`label`		The type of token.

save_to_cache

save_to_cache() -> None

Save the token to the cache file associated with it.

Raises:

Type	Description
`TokenError`	If the parent directory does not exist.
`TokenError`	If there is an unknown error while saving the token.

TokenError

Bases: AuthenticationError

Exception for errors related to tokens on Jobbergate.

TokenType

Bases: str, Enum

The types of tokens available in the system are access and refresh.

exceptions

AuthenticationError

Bases: Buzz

Base exception for errors related to authentication on Jobbergate.

TokenError

Bases: AuthenticationError

Exception for errors related to tokens on Jobbergate.

handler

Utilities for handling authentication in the Jobbergate system.

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.

JobbergateAuthHandler `dataclass`

High-level class used to manage authentication to requests to the Jobbergate-API

After an instance of this class is created, it can be used to authenticate requests from both requests and httpx packages by passing it to the auth parameter on the request (see examples below).

It just works out of the box. Behind the scenes, this procedure calls the :meth:JobbergateAuthHandler.acquire_access method to load the available tokens from the cache directory, it tries to refresh them if they are expired, or provides an URL to the user to login on the system.

Notice all steps above are also available individually as public methods, allowing a fine control for advanced users.

.. _requests: https://requests.readthedocs.io/en/latest/ .. _httpx: https://www.python-httpx.org/

Parameters:

Name	Type	Description	Default
`cache_directory`	`Path`	Directory to be used for the caching tokens.	required
`login_domain`	`str`	Domain used for the login.	required
`login_client_id`	`str`	Client ID used for login.	`'default'`

Note

These values depend on the identity provider used for authentication. Consult your system administrator or contact Omnivector support support@omnivector.solutions for further assistance.

Note

This class can interoperate with the tokens generated by the jobbergate-cli package, as long as they are stored in the same cache directory.

Examples:

The following example shows how to use the :meth:`JobbergateAuthHandler`
class to authenticate a request:

>>> from pathlib import Path
>>> import requests
>>> from jobbergate_core import JobbergateAuthHandler
>>> jobbergate_auth = JobbergateAuthHandler(
...     cache_directory=Path("."),
...     login_domain="http://keycloak.local:8080/realms/jobbergate-local",
...     login_client_id="cli",
... )
>>> jobbergate_base_url = "http://localhost:8000/jobbergate"
>>> response = requests.get(
...     f"{jobbergate_base_url}/applications",
...     auth=jobbergate_auth # this is the important part
)
Login Here: http://keycloak.local:8080/realms/jobbergate-local/device?user_code=LMVJ-XOLG
>>> response.raise_for_status()
>>> print(f"response = {response.json()}")

__call__

__call__(request)

This internal method allows the integration with the requests library.

It is called automatically when the instance is passed to the auth parameter, see the examples in the class docstring.

It adds the Authorization header to the request with the access token.

acquire_access

acquire_access() -> str

High-level method to acquire a valid access token.

This method will attempt, in order:

Use the internal access token from the instance
Load the tokens from the cache directory (see :meth:JobbergateAuthHandler.load_from_cache)
If the access token is unavailable or expired, refresh both tokens using the refresh token grant type (see :meth:JobbergateAuthHandler.refresh_tokens)
If the refresh token is unavailable or expired, login to generate both tokens (see :meth:JobbergateAuthHandler.login)

Returns:

Type	Description
`str`	The bearer access token.

Raises:

Type	Description
`AuthenticationError`	If all of the steps above fail to acquire a valid access token.

get_access_from_secret

get_access_from_secret() -> None

Get the access token using the client secret.

load_from_cache

load_from_cache() -> None

Load the tokens that are available at the cache directory.

login() -> None

Login to Jobbergate.

An URL will be printed to the console, the user must open it in a browser and provide their access credentials.

After the login is completed, the tokens will be saved to the cache directory.

logout

logout() -> None

Logout from Jobbergate by clearing the loaded tokens and their cache on the disk.

refresh_tokens

refresh_tokens() -> None

Refresh the tokens.

After the refresh operation is completed, the tokens will be saved to the cache directory.

Raises:

Type	Description
`AuthenticationError`	If the refresh token is missing or expired.

save_to_cache

save_to_cache() -> None

Save the tokens to the cache directory.

Note

This method will create the cache directory if it does not exist.

TimedIterator `dataclass`

An iterator that runs for a given time interval, yielding the current iteration number.

print_login_url

print_login_url(device_code_data: DeviceCodeData)

Basic way to handle the login url.

token

Utilities for handling tokens on Jobbergate.

Token `dataclass`

Low-level class used to handle tokens.

Parameters:

Name	Type	Description	Default
`cache_directory`	`Path`	The directory used for cache.	required
`label`	`str`	The type of token.	required
`content`	`str`	The content of the token (default is `""`).	`''`

Attributes:

Name	Type	Description
`file_path`	`Path`	The path to the file associated with the token. It is computed as `<cache_directory>/<label>.token`.
`data`	`TokenData`	Metadata decoded from the token's content are available in this dictionary. Expiration date and permissions are some examples of data that can be found.

bearer_token property

bearer_token: str

Return the token with the Bearer prefix.

__post_init__

__post_init__()

Post init method.

clear_cache

clear_cache() -> None

Clear the token from cache by removing the file associated with it.

is_expired

is_expired() -> bool

Check if the token is expired.

Returns:

Type	Description
`bool`	True if the token is expired, False otherwise.

Raises:

Type	Description
`TokenError`	If the expiration date is not found.

is_valid

is_valid() -> bool

Verify if the token is valid, i.e., has content and is not expired.

load_from_cache

load_from_cache() -> Token

Load the token from the cache directory.

Parameters:

Name	Type	Description	Default
`cache_directory`		The path to the cache directory.	required
`label`		The type of token.	required

Returns:

Type	Description
`Token`	A new token with the content replaced.

replace

replace(**changes) -> Token

Create a new instance of the token with the changes applied.

Other Parameters:

Name	Type	Description
`content`		The content of the token.
`cache_directory`		The directory containing the cache.
`label`		The type of token.

save_to_cache

save_to_cache() -> None

Save the token to the cache file associated with it.

Raises:

Type	Description
`TokenError`	If the parent directory does not exist.
`TokenError`	If there is an unknown error while saving the token.

TokenData

Bases: TypedDict

Expected data from the token to make type checking easier.

TokenType

Bases: str, Enum

The types of tokens available in the system are access and refresh.

sdk

Apps

Main class for Jobbergate SDK applications.

Parameters:

Name	Type	Description	Default
`client`		An instance of Client to handle API requests.	required
`request_handler_cls`		A class that handles requests, defaults to RequestHandler.	required

build `classmethod`

build(base_url: str = 'https://apis.vantagecompute.ai', login_url: str = 'https://auth.vantagecompute.ai/realms/vantage', token_dir: Path = Path('~/.local/share/jobbergate3/token'), client_kwargs: dict | None = None, auth_kwargs: dict | None = None) -> Apps

Alternative build method to set client and auth parameters.

clusters

ClusterStatus

SDK class to interact with the cluster status API.

get_list

get_list(size: PositiveInt = 50, page: PositiveInt = 1) -> ListResponseEnvelope[ClusterStatusView]

Get a list of cluster statuses.

Parameters:

Name	Type	Description	Default
`size`	`PositiveInt`	The number of items to return in the response.	`50`
`page`	`PositiveInt`	The page number to return.	`1`

Returns:

Type	Description
`ListResponseEnvelope[ClusterStatusView]`	A list of cluster statuses.

get_one

get_one(client_id: str) -> ClusterStatusView

Get a specific cluster status.

Parameters:

Name	Type	Description	Default
`client_id`	`str`	The client_id of the cluster to get the status for.	required

Returns:

Type	Description
`ClusterStatusView`	The status of the cluster.

app

SDK module to interact with cluster status API.

ClusterStatus

SDK class to interact with the cluster status API.

get_list

get_list(size: PositiveInt = 50, page: PositiveInt = 1) -> ListResponseEnvelope[ClusterStatusView]

Get a list of cluster statuses.

Parameters:

Name	Type	Description	Default
`size`	`PositiveInt`	The number of items to return in the response.	`50`
`page`	`PositiveInt`	The page number to return.	`1`

Returns:

Type	Description
`ListResponseEnvelope[ClusterStatusView]`	A list of cluster statuses.

get_one

get_one(client_id: str) -> ClusterStatusView

Get a specific cluster status.

Parameters:

Name	Type	Description	Default
`client_id`	`str`	The client_id of the cluster to get the status for.	required

Returns:

Type	Description
`ClusterStatusView`	The status of the cluster.

schemas

ClusterStatusView

Bases: BaseModel

Describes the status of a cluster.

constants

FileType

Bases: str, Enum

File type enum.

job_scripts

JobScripts

SDK class to interact with job scripts.

files cached property

files: JobScriptFiles

Get the JobScriptFiles instance.

Returns:

Type	Description
`JobScriptFiles`	The JobScriptFiles instance.

clone

clone(base_id: NonNegativeInt, *, name: str | None = None, description: str | None = None) -> JobScriptDetailedView

Clone a job script.

Parameters:

Name	Type	Description	Default
`base_id`	`NonNegativeInt`	The ID of the base job script.	required
`name`	`str \| None`	The name of the new job script.	`None`
`description`	`str \| None`	The description of the new job script.	`None`

Returns:

Type	Description
`JobScriptDetailedView`	The detailed view of the cloned job script.

create

create(name: str, description: str | None = None) -> JobScriptBaseView

Create a stand alone job script. Use file upload to add files.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the job script.	required
`identifier`		The identifier of the job script.	required
`description`	`str \| None`	The description of the job script.	`None`
`template_vars`		The template variables.	required

Returns:

Type	Description
`JobScriptBaseView`	The detailed view of the created job script.

delete

delete(id: NonNegativeInt) -> None

Delete a job script.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required

get_list

get_list(sort_ascending: bool = True, user_only: bool = False, search: str | None = None, sort_field: str | None = None, include_archived: bool = False, include_parent: bool = False, from_job_script_template_id: NonNegativeInt | None = None, size: PositiveInt = 50, page: PositiveInt = 1) -> ListResponseEnvelope[JobScriptListView]

List job scripts.

Parameters:

Name	Type	Description	Default
`include_null_identifier`		Whether to include scripts with null identifiers.	required
`sort_ascending`	`bool`	Whether to sort in ascending order.	`True`
`user_only`	`bool`	Whether to include only user-specific scripts.	`False`
`search`	`str \| None`	The search query.	`None`
`sort_field`	`str \| None`	The field to sort by.	`None`
`include_archived`	`bool`	Whether to include archived scripts.	`False`
`size`	`PositiveInt`	The number of scripts per page.	`50`
`page`	`PositiveInt`	The page number.	`1`

Returns:

Type	Description
`ListResponseEnvelope[JobScriptListView]`	The list response envelope containing job script list views.

get_one

get_one(id: NonNegativeInt) -> JobScriptDetailedView

Get a single job script.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required

Returns:

Type	Description
`JobScriptDetailedView`	The detailed view of the job script.

render_from_template

render_from_template(id_or_identifier: NonNegativeInt | str, name: str, *, template_output_name_mapping: dict[str, str], param_dict: dict[str, Any], description: str | None = None, sbatch_params: list[str] | None = None) -> JobScriptDetailedView

Render a job script from a template.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the base job script template.	required
`name`	`str`	The name of the new job script.	required
`template_output_name_mapping`	`dict[str, str]`	The mapping of template output names to new names.	required
`param_dict`	`dict[str, Any]`	The render parameters dictionary.	required
`description`	`str \| None`	The description of the new job script.	`None`
`sbatch_params`	`list[str] \| None`	The sbatch parameters.	`None`

Returns:

Type	Description
`JobScriptDetailedView`	The detailed view of the rendered job script.

update

update(id: NonNegativeInt, *, name: str | None = None, description: str | None = None, is_archived: bool | None = None) -> JobScriptBaseView

Update a job script.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required
`name`	`str \| None`	The name of the job script.	`None`
`identifier`		The identifier of the job script.	required
`description`	`str \| None`	The description of the job script.	`None`
`template_vars`		The template variables.	required
`is_archived`	`bool \| None`	Whether the job script is archived.	`None`

Returns:

Type	Description
`JobScriptBaseView`	The detailed view of the updated job script.

app

SDK module to interact with job scripts.

JobScriptFiles

SDK class to interact with job script files.

delete

delete(id: NonNegativeInt, filename: str) -> None

Delete a job script file.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required
`filename`	`str`	The name of the file to be deleted.	required

download

download(id: NonNegativeInt, filename: str, directory: Path = Path.cwd()) -> Path

Download a job script file.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required
`filename`	`str`	The name of the file to be downloaded.	required
`directory`	`Path`	The directory where the file will be saved.	`cwd()`

Returns:

Type	Description
`Path`	The path to the downloaded file.

upsert

upsert(id: NonNegativeInt, file_type: FileType, file_path: Path) -> JobScriptFileDetailedView

Upload or update a job script file.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required
`file_type`	`FileType`	The type of the file.	required
`file_path`	`Path`	The path to the file to be uploaded.	required

Returns:

Type	Description
`JobScriptFileDetailedView`	The detailed view of the job script file.

JobScripts

SDK class to interact with job scripts.

files cached property

files: JobScriptFiles

Get the JobScriptFiles instance.

Returns:

Type	Description
`JobScriptFiles`	The JobScriptFiles instance.

clone

clone(base_id: NonNegativeInt, *, name: str | None = None, description: str | None = None) -> JobScriptDetailedView

Clone a job script.

Parameters:

Name	Type	Description	Default
`base_id`	`NonNegativeInt`	The ID of the base job script.	required
`name`	`str \| None`	The name of the new job script.	`None`
`description`	`str \| None`	The description of the new job script.	`None`

Returns:

Type	Description
`JobScriptDetailedView`	The detailed view of the cloned job script.

create

create(name: str, description: str | None = None) -> JobScriptBaseView

Create a stand alone job script. Use file upload to add files.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the job script.	required
`identifier`		The identifier of the job script.	required
`description`	`str \| None`	The description of the job script.	`None`
`template_vars`		The template variables.	required

Returns:

Type	Description
`JobScriptBaseView`	The detailed view of the created job script.

delete

delete(id: NonNegativeInt) -> None

Delete a job script.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required

get_list

get_list(sort_ascending: bool = True, user_only: bool = False, search: str | None = None, sort_field: str | None = None, include_archived: bool = False, include_parent: bool = False, from_job_script_template_id: NonNegativeInt | None = None, size: PositiveInt = 50, page: PositiveInt = 1) -> ListResponseEnvelope[JobScriptListView]

List job scripts.

Parameters:

Name	Type	Description	Default
`include_null_identifier`		Whether to include scripts with null identifiers.	required
`sort_ascending`	`bool`	Whether to sort in ascending order.	`True`
`user_only`	`bool`	Whether to include only user-specific scripts.	`False`
`search`	`str \| None`	The search query.	`None`
`sort_field`	`str \| None`	The field to sort by.	`None`
`include_archived`	`bool`	Whether to include archived scripts.	`False`
`size`	`PositiveInt`	The number of scripts per page.	`50`
`page`	`PositiveInt`	The page number.	`1`

Returns:

Type	Description
`ListResponseEnvelope[JobScriptListView]`	The list response envelope containing job script list views.

get_one

get_one(id: NonNegativeInt) -> JobScriptDetailedView

Get a single job script.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required

Returns:

Type	Description
`JobScriptDetailedView`	The detailed view of the job script.

render_from_template

render_from_template(id_or_identifier: NonNegativeInt | str, name: str, *, template_output_name_mapping: dict[str, str], param_dict: dict[str, Any], description: str | None = None, sbatch_params: list[str] | None = None) -> JobScriptDetailedView

Render a job script from a template.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the base job script template.	required
`name`	`str`	The name of the new job script.	required
`template_output_name_mapping`	`dict[str, str]`	The mapping of template output names to new names.	required
`param_dict`	`dict[str, Any]`	The render parameters dictionary.	required
`description`	`str \| None`	The description of the new job script.	`None`
`sbatch_params`	`list[str] \| None`	The sbatch parameters.	`None`

Returns:

Type	Description
`JobScriptDetailedView`	The detailed view of the rendered job script.

update

update(id: NonNegativeInt, *, name: str | None = None, description: str | None = None, is_archived: bool | None = None) -> JobScriptBaseView

Update a job script.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job script.	required
`name`	`str \| None`	The name of the job script.	`None`
`identifier`		The identifier of the job script.	required
`description`	`str \| None`	The description of the job script.	`None`
`template_vars`		The template variables.	required
`is_archived`	`bool \| None`	Whether the job script is archived.	`None`

Returns:

Type	Description
`JobScriptBaseView`	The detailed view of the updated job script.

schemas

JobScriptBaseView

Bases: TableResource

Base schema for the request to an entry.

Omits parent relationship.

JobScriptDetailedView

Bases: JobScriptBaseView

Schema for the request to an entry.

Notice the files default to None, as they are not always requested, to differentiate between an empty list when they are requested, but no file is found.

JobScriptFileDetailedView

Bases: BaseModel

Schema for the response to get a job-script file.

JobScriptListView

Bases: JobScriptBaseView

Schema for the response to get a list of entries.

Notice files are omitted. Parent template can be included.

job_submissions

JobSubmissions

cancel

cancel(id: NonNegativeInt) -> JobSubmissionDetailedView

Cancel a job submission.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the canceled job submission.

clone

clone(id: NonNegativeInt) -> JobSubmissionDetailedView

Clone a job submission under the CREATED status for a new run on the cluster.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission to clone.	required

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the cloned job submission.

create

create(job_script_id: NonNegativeInt, name: str, *, description: str | None = None, slurm_job_id: NonNegativeInt | None = None, execution_directory: str | None = None, client_id: str | None = None, sbatch_arguments: list[str] | None = None) -> JobSubmissionDetailedView

Create a job submission.

Parameters:

Name	Type	Description	Default
`job_script_id`	`NonNegativeInt`	The ID of the base job script.	required
`name`	`str`	The name of the job submission.	required
`description`	`str \| None`	The description of the job submission.	`None`
`slurm_job_id`	`NonNegativeInt \| None`	The SLURM job ID.	`None`
`execution_directory`	`str \| None`	The execution directory.	`None`
`client_id`	`str \| None`	The client ID.	`None`
`sbatch_arguments`	`list[str] \| None`	The SLURM sbatch arguments.	`None`

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the created job submission.

delete

delete(id: NonNegativeInt) -> None

Delete a job submission.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required

get_list

get_list(sort_ascending: bool = True, user_only: bool = False, search: str | None = None, sort_field: str | None = None, include_archived: bool = False, include_parent: bool = False, slurm_job_ids: list[NonNegativeInt] | None = None, slurm_status: JobSubmissionStatus | None = None, from_job_script_id: NonNegativeInt | None = None, size: PositiveInt = 50, page: PositiveInt = 1) -> ListResponseEnvelope[JobSubmissionListView]

List job submissions.

Parameters:

Name	Type	Description	Default
`sort_ascending`	`bool`	Whether to sort in ascending order.	`True`
`user_only`	`bool`	Whether to include only user-specific submissions.	`False`
`search`	`str \| None`	The search query.	`None`
`sort_field`	`str \| None`	The field to sort by.	`None`
`include_archived`	`bool`	Whether to include archived submissions.	`False`
`include_parent`	`bool`	Whether to include parent submissions.	`False`
`slurm_job_ids`	`list[NonNegativeInt] \| None`	Filter by SLURM job IDs.	`None`
`slurm_status`	`JobSubmissionStatus \| None`	Filter by SLURM status.	`None`
`from_job_script_id`	`NonNegativeInt \| None`	Filter by job script ID.	`None`
`size`	`PositiveInt`	The number of submissions per page.	`50`
`page`	`PositiveInt`	The page number.	`1`

Returns:

Type	Description
`ListResponseEnvelope[JobSubmissionListView]`	The list response envelope containing job submission list views.

get_one

get_one(id: NonNegativeInt) -> JobSubmissionDetailedView

Get a single job submission.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the job submission.

get_one_ensure_slurm_id

get_one_ensure_slurm_id(id: NonNegativeInt, max_retries: PositiveInt = 8, waiting_interval: PositiveInt = 15) -> JobSubmissionDetailedView

Get a single job submission and ensure that the SLURM job ID is set.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required
`max_retries`	`PositiveInt`	The maximum number of retry attempts.	`8`
`waiting_interval`	`PositiveInt`	The interval in seconds to wait between checks.	`15`

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the job submission.

update

update(id: NonNegativeInt, *, name: str | None = None, description: str | None = None, execution_directory: str | None = None, status: JobSubmissionStatus | None = None) -> JobSubmissionDetailedView

Update a job submission.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required
`name`	`str \| None`	The name of the job submission.	`None`
`description`	`str \| None`	The description of the job submission.	`None`
`execution_directory`	`str \| None`	The execution directory.	`None`

Returns:

Type	Description
`JobSubmissionDetailedView`	The base view of the updated job submission.

app

JobSubmissions

cancel

cancel(id: NonNegativeInt) -> JobSubmissionDetailedView

Cancel a job submission.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the canceled job submission.

clone

clone(id: NonNegativeInt) -> JobSubmissionDetailedView

Clone a job submission under the CREATED status for a new run on the cluster.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission to clone.	required

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the cloned job submission.

create

create(job_script_id: NonNegativeInt, name: str, *, description: str | None = None, slurm_job_id: NonNegativeInt | None = None, execution_directory: str | None = None, client_id: str | None = None, sbatch_arguments: list[str] | None = None) -> JobSubmissionDetailedView

Create a job submission.

Parameters:

Name	Type	Description	Default
`job_script_id`	`NonNegativeInt`	The ID of the base job script.	required
`name`	`str`	The name of the job submission.	required
`description`	`str \| None`	The description of the job submission.	`None`
`slurm_job_id`	`NonNegativeInt \| None`	The SLURM job ID.	`None`
`execution_directory`	`str \| None`	The execution directory.	`None`
`client_id`	`str \| None`	The client ID.	`None`
`sbatch_arguments`	`list[str] \| None`	The SLURM sbatch arguments.	`None`

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the created job submission.

delete

delete(id: NonNegativeInt) -> None

Delete a job submission.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required

get_list

get_list(sort_ascending: bool = True, user_only: bool = False, search: str | None = None, sort_field: str | None = None, include_archived: bool = False, include_parent: bool = False, slurm_job_ids: list[NonNegativeInt] | None = None, slurm_status: JobSubmissionStatus | None = None, from_job_script_id: NonNegativeInt | None = None, size: PositiveInt = 50, page: PositiveInt = 1) -> ListResponseEnvelope[JobSubmissionListView]

List job submissions.

Parameters:

Name	Type	Description	Default
`sort_ascending`	`bool`	Whether to sort in ascending order.	`True`
`user_only`	`bool`	Whether to include only user-specific submissions.	`False`
`search`	`str \| None`	The search query.	`None`
`sort_field`	`str \| None`	The field to sort by.	`None`
`include_archived`	`bool`	Whether to include archived submissions.	`False`
`include_parent`	`bool`	Whether to include parent submissions.	`False`
`slurm_job_ids`	`list[NonNegativeInt] \| None`	Filter by SLURM job IDs.	`None`
`slurm_status`	`JobSubmissionStatus \| None`	Filter by SLURM status.	`None`
`from_job_script_id`	`NonNegativeInt \| None`	Filter by job script ID.	`None`
`size`	`PositiveInt`	The number of submissions per page.	`50`
`page`	`PositiveInt`	The page number.	`1`

Returns:

Type	Description
`ListResponseEnvelope[JobSubmissionListView]`	The list response envelope containing job submission list views.

get_one

get_one(id: NonNegativeInt) -> JobSubmissionDetailedView

Get a single job submission.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the job submission.

get_one_ensure_slurm_id

get_one_ensure_slurm_id(id: NonNegativeInt, max_retries: PositiveInt = 8, waiting_interval: PositiveInt = 15) -> JobSubmissionDetailedView

Get a single job submission and ensure that the SLURM job ID is set.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required
`max_retries`	`PositiveInt`	The maximum number of retry attempts.	`8`
`waiting_interval`	`PositiveInt`	The interval in seconds to wait between checks.	`15`

Returns:

Type	Description
`JobSubmissionDetailedView`	The detailed view of the job submission.

update

update(id: NonNegativeInt, *, name: str | None = None, description: str | None = None, execution_directory: str | None = None, status: JobSubmissionStatus | None = None) -> JobSubmissionDetailedView

Update a job submission.

Parameters:

Name	Type	Description	Default
`id`	`NonNegativeInt`	The ID of the job submission.	required
`name`	`str \| None`	The name of the job submission.	`None`
`description`	`str \| None`	The description of the job submission.	`None`
`execution_directory`	`str \| None`	The execution directory.	`None`

Returns:

Type	Description
`JobSubmissionDetailedView`	The base view of the updated job submission.

constants

JobSubmissionStatus

Bases: str, Enum

Defines the set of possible statuses for a Job Submission.

schemas

JobSubmissionBaseView

Bases: TableResource

Base model to match database for the JobSubmission resource.

Omits parent relationship.

JobSubmissionDetailedView

Bases: JobSubmissionBaseView

Model to match database for the JobSubmission resource.

JobSubmissionListView

Bases: JobSubmissionBaseView

Model to match database for the JobSubmission resource.

Notice files are omitted. Parent job-script can be included.

job_templates

JobTemplates

files cached property

files: Files

Get the Files instance.

Returns:

Type	Description
`Files`	The Files instance.

clone

clone(base_id_or_identifier: NonNegativeInt | str, *, name: str | None = None, identifier: str | None = None, description: str | None = None, template_vars: dict[str, Any] | None = None) -> JobTemplateDetailedView

Clone a job template.

Parameters:

Name	Type	Description	Default
`base_id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the base job template.	required
`name`	`str \| None`	The name of the new job template.	`None`
`identifier`	`str \| None`	The identifier of the new job template.	`None`
`description`	`str \| None`	The description of the new job template.	`None`
`template_vars`	`dict[str, Any] \| None`	The template variables.	`None`

Returns:

Type	Description
`JobTemplateDetailedView`	The detailed view of the cloned job template.

create

create(name: str, *, identifier: str | None = None, description: str | None = None, template_vars: dict[str, Any] | None = None) -> JobTemplateBaseDetailedView

Create a new job template.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the job template.	required
`identifier`	`str \| None`	The identifier of the job template.	`None`
`description`	`str \| None`	The description of the job template.	`None`
`template_vars`	`dict[str, Any] \| None`	The template variables.	`None`

Returns:

Type	Description
`JobTemplateBaseDetailedView`	The detailed view of the created job template.

delete

delete(id_or_identifier: NonNegativeInt | str) -> None

Delete a job template.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required

get_list

get_list(include_null_identifier: bool = False, sort_ascending: bool = True, user_only: bool = False, search: str | None = None, sort_field: str | None = None, include_archived: bool = False, size: PositiveInt = 50, page: PositiveInt = 1) -> ListResponseEnvelope[JobTemplateListView]

List job templates.

Parameters:

Name	Type	Description	Default
`include_null_identifier`	`bool`	Whether to include templates with null identifiers.	`False`
`sort_ascending`	`bool`	Whether to sort in ascending order.	`True`
`user_only`	`bool`	Whether to include only user-specific templates.	`False`
`search`	`str \| None`	The search query.	`None`
`sort_field`	`str \| None`	The field to sort by.	`None`
`include_archived`	`bool`	Whether to include archived templates.	`False`
`size`	`PositiveInt`	The number of templates per page.	`50`
`page`	`PositiveInt`	The page number.	`1`

Returns:

Type	Description
`ListResponseEnvelope[JobTemplateListView]`	The list response envelope containing job template list views.

get_one

get_one(id_or_identifier: NonNegativeInt | str) -> JobTemplateDetailedView

Get a single job template.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required

Returns:

Type	Description
`JobTemplateDetailedView`	The detailed view of the job template.

update

update(id_or_identifier: NonNegativeInt | str, *, name: str | None = None, identifier: str | None = None, description: str | None = None, template_vars: dict[str, Any] | None = None, is_archived: bool | None = None) -> JobTemplateBaseDetailedView

Update a job template.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required
`name`	`str \| None`	The name of the job template.	`None`
`identifier`	`str \| None`	The identifier of the job template.	`None`
`description`	`str \| None`	The description of the job template.	`None`
`template_vars`	`dict[str, Any] \| None`	The template variables.	`None`
`is_archived`	`bool \| None`	Whether the job template is archived.	`None`

Returns:

Type	Description
`JobTemplateBaseDetailedView`	The detailed view of the updated job template.

app

Files

template cached property

template: TemplateFiles

Get the TemplateFiles instance.

Returns:

Type	Description
`TemplateFiles`	The TemplateFiles instance.

workflow cached property

workflow: WorkflowFiles

Get the WorkflowFiles instance.

Returns:

Type	Description
`WorkflowFiles`	The WorkflowFiles instance.

JobTemplates

files cached property

files: Files

Get the Files instance.

Returns:

Type	Description
`Files`	The Files instance.

clone

clone(base_id_or_identifier: NonNegativeInt | str, *, name: str | None = None, identifier: str | None = None, description: str | None = None, template_vars: dict[str, Any] | None = None) -> JobTemplateDetailedView

Clone a job template.

Parameters:

Name	Type	Description	Default
`base_id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the base job template.	required
`name`	`str \| None`	The name of the new job template.	`None`
`identifier`	`str \| None`	The identifier of the new job template.	`None`
`description`	`str \| None`	The description of the new job template.	`None`
`template_vars`	`dict[str, Any] \| None`	The template variables.	`None`

Returns:

Type	Description
`JobTemplateDetailedView`	The detailed view of the cloned job template.

create

create(name: str, *, identifier: str | None = None, description: str | None = None, template_vars: dict[str, Any] | None = None) -> JobTemplateBaseDetailedView

Create a new job template.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the job template.	required
`identifier`	`str \| None`	The identifier of the job template.	`None`
`description`	`str \| None`	The description of the job template.	`None`
`template_vars`	`dict[str, Any] \| None`	The template variables.	`None`

Returns:

Type	Description
`JobTemplateBaseDetailedView`	The detailed view of the created job template.

delete

delete(id_or_identifier: NonNegativeInt | str) -> None

Delete a job template.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required

get_list

get_list(include_null_identifier: bool = False, sort_ascending: bool = True, user_only: bool = False, search: str | None = None, sort_field: str | None = None, include_archived: bool = False, size: PositiveInt = 50, page: PositiveInt = 1) -> ListResponseEnvelope[JobTemplateListView]

List job templates.

Parameters:

Name	Type	Description	Default
`include_null_identifier`	`bool`	Whether to include templates with null identifiers.	`False`
`sort_ascending`	`bool`	Whether to sort in ascending order.	`True`
`user_only`	`bool`	Whether to include only user-specific templates.	`False`
`search`	`str \| None`	The search query.	`None`
`sort_field`	`str \| None`	The field to sort by.	`None`
`include_archived`	`bool`	Whether to include archived templates.	`False`
`size`	`PositiveInt`	The number of templates per page.	`50`
`page`	`PositiveInt`	The page number.	`1`

Returns:

Type	Description
`ListResponseEnvelope[JobTemplateListView]`	The list response envelope containing job template list views.

get_one

get_one(id_or_identifier: NonNegativeInt | str) -> JobTemplateDetailedView

Get a single job template.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required

Returns:

Type	Description
`JobTemplateDetailedView`	The detailed view of the job template.

update

update(id_or_identifier: NonNegativeInt | str, *, name: str | None = None, identifier: str | None = None, description: str | None = None, template_vars: dict[str, Any] | None = None, is_archived: bool | None = None) -> JobTemplateBaseDetailedView

Update a job template.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required
`name`	`str \| None`	The name of the job template.	`None`
`identifier`	`str \| None`	The identifier of the job template.	`None`
`description`	`str \| None`	The description of the job template.	`None`
`template_vars`	`dict[str, Any] \| None`	The template variables.	`None`
`is_archived`	`bool \| None`	Whether the job template is archived.	`None`

Returns:

Type	Description
`JobTemplateBaseDetailedView`	The detailed view of the updated job template.

TemplateFiles

delete

delete(id_or_identifier: NonNegativeInt | str, filename: str) -> None

Delete a template file.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required
`filename`	`str`	The name of the file to be deleted.	required

download

download(id_or_identifier: NonNegativeInt | str, filename: str, directory: Path = Path.cwd()) -> Path

Download a template file.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required
`filename`	`str`	The name of the file to be downloaded.	required
`directory`	`Path`	The directory where the file will be saved.	`cwd()`

Returns:

Type	Description
`Path`	The path to the downloaded file.

upsert

upsert(id_or_identifier: NonNegativeInt | str, file_type: FileType, file_path: Path) -> TemplateFileDetailedView

Upload or update a template file.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required
`file_type`	`FileType`	The type of the file.	required
`file_path`	`Path`	The path to the file to be uploaded.	required

Returns:

Type	Description
`TemplateFileDetailedView`	The detailed view of the template file.

WorkflowFiles

delete

delete(id_or_identifier: NonNegativeInt | str) -> None

Delete a workflow file.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required

download

download(id_or_identifier: NonNegativeInt | str, directory: Path = Path.cwd()) -> Path

Download a workflow file.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required
`directory`	`Path`	The directory where the file will be saved.	`cwd()`

Returns:

Type	Description
`Path`	The path to the downloaded file.

upsert

upsert(id_or_identifier: NonNegativeInt | str, file_path: Path | None = None, runtime_config: dict[str, Any] | None = None) -> WorkflowFileDetailedView

Upload or update a workflow file.

Parameters:

Name	Type	Description	Default
`id_or_identifier`	`NonNegativeInt \| str`	The ID or identifier of the job template.	required
`file_path`	`Path \| None`	The path to the file to be uploaded.	`None`
`runtime_config`	`dict[str, Any] \| None`	The runtime configuration.	`None`

Returns:

Type	Description
`WorkflowFileDetailedView`	The detailed view of the workflow file.

schemas

Provide schemas for the job script templates component.

JobTemplateBaseDetailedView

Bases: JobTemplateListView

Schema for the request to an entry.

Notice the files are omitted.

JobTemplateDetailedView

Bases: JobTemplateBaseDetailedView

Schema for the request to an entry.

Notice the files default to None, as they are not always requested, to differentiate between an empty list when they are requested, but no file is found.

JobTemplateListView

Bases: TableResource

Schema for the response to get a list of entries.

TemplateFileDetailedView

Bases: BaseModel

Schema for the response to get a template file.

WorkflowFileDetailedView

Bases: BaseModel

Schema for the response to get a workflow file.

schemas

ListResponseEnvelope

Bases: BaseModel, Generic[EnvelopeT]

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

PydanticDateTime

Bases: DateTime

A pendulum.DateTime object. At runtime, this type decomposes into pendulum.DateTime automatically.

This type exists because Pydantic throws a fit on unknown types.

This code is borrowed and enhanced from the pydantic-extra-types module but provides conversion from standard datetimes as well.

__get_pydantic_core_schema__ classmethod

__get_pydantic_core_schema__(source: Type[Any], handler: GetCoreSchemaHandler) -> core_schema.CoreSchema

Return a Pydantic CoreSchema with the Datetime validation.

Parameters:

Name	Type	Description	Default
`source`	`Type[Any]`	The source type to be converted.	required
`handler`	`GetCoreSchemaHandler`	The handler to get the CoreSchema.	required

Returns:

Type	Description
`CoreSchema`	A Pydantic CoreSchema with the Datetime validation.

TableResource

Bases: BaseModel

Describes a base for table models that include basic, common info.

tools

requests

JobbergateRequestError

Bases: RequestError

Jobbergate specific exceptions that may occur when preparing a request.

JobbergateResponseError

Bases: HTTPStatusError

Jobbergate specific exceptions that may occur when handling a response.

RequestHandler `dataclass`

Provide utilities for making requests and handling responses.

Notice most methods return self as a syntax sugar to allow chaining.

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
`request_model`	`BaseModel \| None`	Use a pydantic model instance as the data body for the request	`None`
`request_kwargs`	`dict[str, Any]`	Any additional keyword arguments that need to be passed on to the client	`field(default_factory=dict)`

Attributes:

Name	Type	Description
`response`	`Response`	The response object from the request is kept for reference.

__post_init__

__post_init__()

Post init method.

check_status_code

check_status_code(*statuses: int) -> RequestHandler

Check if the response status code is in the provided set of status codes.

raise_for_status

raise_for_status() -> RequestHandler

Raise the HTTPStatusError if one occurred.

to_file

to_file(file_path: Path) -> Path

Write the response content to a file.

to_json

to_json() -> dict[str, Any]

Unpack the response content as json.

to_model

to_model(model: Type[ResponseModel]) -> ResponseModel

Unpack the response content as json and validate it against a pydantic model.

RequestModelError

Bases: JobbergateRequestError

An error occurred while preparing the request data from a model.

deserialize_request_model

deserialize_request_model(request_model: BaseModel, request_kwargs: dict[str, Any]) -> None

Deserialize a pydantic model instance into request_kwargs for an httpx client request in place.

sbatch

InfoHandler `dataclass`

Get info from jobs on the cluster.

get_job_info

get_job_info(slurm_id: int) -> dict[str, Any]

Gets job info as the user.

ScancelHandler `dataclass`

Cancels jobs on the cluster.

cancel_job

cancel_job(slurm_id: int) -> None

Cancels a job with the given slurm id.

SubmissionHandler `dataclass`

Submits sbatch jobs to the cluster.

copy_file_to_submission_directory

copy_file_to_submission_directory(source_file: Path) -> Path

Copies the job file to the submission directory as the user.

submit_job

submit_job(job_script_path: Path) -> int

Runs sbatch as the user to submit a job script and returns the slurm id assigned to it.

inject_sbatch_params

inject_sbatch_params(job_script_data_as_string: str, sbatch_params: list[str], header: str | None = None) -> str

Injects sbatch parameters into a job script.

This function takes a job script as a string, a list of sbatch parameters, and an optional header.

Parameters:

Name	Type	Description	Default
`job_script_data_as_string`	`str`	The job script as a string.	required
`sbatch_params`	`list[str]`	A list of sbatch parameters to be inserted.	required
`header`	`str \| None`	A comment to be inserted before the parameters (i.e., "Injected at runtime by Jobbergate").	`None`

Returns:

Type	Description
`str`	The job script with the sbatch parameters inserted.

version

Provide the version of the package.

get_version

get_version()

Get the version of the package.

Jobbergate Core Reference

jobbergate_core

AuthenticationError

JobbergateAuthHandler dataclass

__call__

acquire_access

get_access_from_secret

load_from_cache

login

logout

refresh_tokens

save_to_cache

TokenError

auth

AuthenticationError

JobbergateAuthHandler dataclass

__call__

acquire_access

get_access_from_secret

load_from_cache

login

logout

refresh_tokens

save_to_cache

Token dataclass

bearer_token property

__post_init__

clear_cache

is_expired

is_valid

load_from_cache

replace

save_to_cache

TokenError

TokenType

exceptions

AuthenticationError

TokenError

handler

DeviceCodeData

IdentityData

JobbergateAuthHandler dataclass

TimedIterator dataclass

print_login_url

token

Token dataclass

TokenData

TokenType

sdk

Apps

build classmethod

clusters

ClusterStatus

app

schemas

constants

FileType

job_scripts

JobScripts

app

schemas

job_submissions

JobSubmissions

app

constants

schemas

job_templates

JobTemplates

app

schemas

schemas

ListResponseEnvelope

PydanticDateTime

TableResource

tools

requests

JobbergateRequestError

JobbergateResponseError

RequestHandler dataclass

RequestModelError

JobbergateAuthHandler `dataclass`

call

JobbergateAuthHandler `dataclass`

call

Token `dataclass`

bearer_token `property`

JobbergateAuthHandler `dataclass`

TimedIterator `dataclass`

Token `dataclass`

build `classmethod`

RequestHandler `dataclass`

InfoHandler `dataclass`

ScancelHandler `dataclass`

SubmissionHandler `dataclass`