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_audience`	`str`	Audience of 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_audience="https://local.omnivector.solutions",
...     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.

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_audience`	`str`	Audience of 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_audience="https://local.omnivector.solutions",
...     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.

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 handling 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`	`Dict[str, Any]`	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.

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_audience`	`str`	Audience of 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_audience="https://local.omnivector.solutions",
...     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.

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

Utilities for handling tokens on Jobbergate.

Token `dataclass`

Low-level class used to handling 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`	`Dict[str, Any]`	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.

TokenType

Bases: str, Enum

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

tools

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.

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.

Jobbergate Core Reference

jobbergate_core

AuthenticationError

JobbergateAuthHandler dataclass

__call__

acquire_access

load_from_cache

login

logout

refresh_tokens

save_to_cache

TokenError

auth

AuthenticationError

JobbergateAuthHandler dataclass

__call__

acquire_access

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

JobbergateAuthHandler dataclass

token

Token dataclass

TokenType

tools

sbatch

InfoHandler dataclass

SubmissionHandler dataclass

inject_sbatch_params

version

JobbergateAuthHandler `dataclass`

call

JobbergateAuthHandler `dataclass`

call

Token `dataclass`

bearer_token `property`

JobbergateAuthHandler `dataclass`

Token `dataclass`

InfoHandler `dataclass`

SubmissionHandler `dataclass`