Utilities

Set of utilities used in various parts of the code

pubtools._quay.utils.misc.setup_arg_parser(args: Dict[Any, Any]) → ArgumentParser

Set up ArgumentParser with the provided arguments.

Args:

args (dict)

Dictionary of argument aliases and options to be consumed by ArgumentParser.

Returns:

(ArgumentParser) Configured instance of ArgumentParser.

pubtools._quay.utils.misc.add_args_env_variables(parsed_args: Namespace, args: Dict[Any, Any]) → Namespace

Add argument values from environment variables.

Args:

parsed_args ():

Parsed arguments object.

args (dict):

Argument definition.

Returns:

Modified parsed arguments object.

pubtools._quay.utils.misc.capture_stdout() → Generator[StringIO, None, None]

Capture sys.stdout to stream buffer.

pubtools._quay.utils.misc.setup_entry_point_cli(entry_tuple: Tuple[str, str, str], name: Optional[str], args: List[str], environ_vars: Dict[str, Any]) → Generator[Callable[[], Any], None, None]

Set up an entrypoint as a context manager.

Args:

entry_tuple ((str, str, str)):

Tuple consisting of dependency, category, and entrypoint.

name: (str):

Entrypoint name.

args ([str]):

Entrypoint arguments.

environ_vars (dict):

Env variable names and values to set for the entrypoint.

pubtools._quay.utils.misc.run_entrypoint(entry_tuple: Tuple[str, str, str], name: Optional[str], args: List[str], environ_vars: Dict[str, Any], capture_out: bool = True) → Any

Run an entrypoint function and return its return value.

Args:

entry_tuple ((str, str, str)):

Tuple consisting of dependency, category, and entrypoint.

name: (str):

Entrypoint name.

args ([str]):

Entrypoint arguments.

environ_vars (dict):

Env variable names and values to set for the entrypoint.

capture_out (bool):

Whether to capture stdout.

Returns (str):

Data returned by the entrypoint.

pubtools._quay.utils.misc.get_internal_container_repo_name(external_name: str) → str

Transform a repository name to an internal form in which it exists on Quay.io.

Expected input format: <namespace>/<product> Generated output format: <namespace>—-<product>

NOTE: Repositories without a delimeter “/” may actually exist. In that case, the function simply returns the repo without any alterations.

Args:

external_name (str):

External repository name.

Returns:

Internal repository name.

pubtools._quay.utils.misc.get_external_container_repo_name(internal_name: str) → str

Transform a repository name to an external form in which it’s visible to customers.

Expected input format: <namespace>—-<product> Generated output format: <namespace>/<product>

NOTE: Repositories without a delimeter “—-” may actually exist. In that case, the function simply returns the repo without any alterations.

Args:

internal_name (str):

Internal repository name.

Returns:

External repository name.

pubtools._quay.utils.misc.task_status(event: str) → Dict[str, Dict[str, str]]

Helper function. Expand as necessary.

pubtools._quay.utils.misc.log_step(step_name: str) → Callable[[Any], Any]

Log status for methods which constitute an entire task step.

Args:

step_name (str):

Name of the task step, e.g., “Tag images”.

pubtools._quay.utils.misc.get_pyxis_ssl_paths(target_settings: Dict[str, Any]) → Tuple[str, str]

Get certificate and key paths for Pyxis SSL authentication.

First attempt is made by invoking the hook implementation ‘get_cert_key_paths’. If nothing is returned (no hook implementation is registered), fallback on target settings values of ‘pyxis_ssl_cert’ and ‘pyxis_ssl_key’. If multiple values are returned (multiple hook implementations are registered), take the first response (from the implementation which was registered last). Otherwise, raise an error.

Args:

target_settings (dict):

Dictionary containing various task-related settings.

Returns ((str, str)):

Paths to Pyxis SSL certificate and key.

pubtools._quay.utils.misc.timestamp() → str

Return now() timestamp.

pubtools._quay.utils.misc.run_with_retries(function: Callable[[], Any], message: str, tries: int = 4, wait_time_increase: int = 10) → Any

Run the specified function until it succeeds or maximum retries are reached.

Wait time will increase after every retry, up to a point defined by MAX_RETRY_WAIT.

Args:

function (callable):

Function that should be retried. It must be able to run with 0 parameters.

message (str):

Message describing the action performed by the function. For example, “tag images”.

tries (int):

Numbers of times to run the function before giving up.

wait_time_increase (int):

Time increase (in seconds) to wait before running the function again. Example (default): RUN -> WAIT 0 -> RUN -> WAIT 10 -> RUN -> WAIT 20 -> RUN

pubtools._quay.utils.misc.retry(message: str, tries: int = 4, wait_time_increase: int = 10) → Callable[[Callable[[...], Any]], Any]

Retry decorated function.

Args:

message (str):

Message describing the action performed by the function. For example, “tag images”.

tries (int):

Numbers of times to run the function before giving up.

wait_time_increase (int):

Time increase (in seconds) to wait before running the function again. Example (default): RUN -> WAIT 0 -> RUN -> WAIT 10 -> RUN -> WAIT 20 -> RUN