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.

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

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.

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

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

tools

requests

JobbergateRequestError

JobbergateResponseError

RequestHandler dataclass

RequestModelError

deserialize_request_model

sbatch

InfoHandler dataclass

SubmissionHandler dataclass

inject_sbatch_params

version

JobbergateAuthHandler `dataclass`

call

JobbergateAuthHandler `dataclass`

call

Token `dataclass`

bearer_token `property`

JobbergateAuthHandler `dataclass`

TimedIterator `dataclass`

Token `dataclass`

RequestHandler `dataclass`

InfoHandler `dataclass`

SubmissionHandler `dataclass`