Pyxis Client
Class for performing actual Pyxis queries and handling related logic.
- class pubtools._pyxis.pyxis_client.PyxisClient(hostname: str, retries: int = 5, auth: PyxisAuth | None = None, backoff_factor: int = 5, verify: bool = True, threads: int = 16)[source]
Pyxis requests wrapper.
- __init__(hostname: str, retries: int = 5, auth: PyxisAuth | None = None, backoff_factor: int = 5, verify: bool = True, threads: int = 16) None[source]
Initialize.
- Args:
- hostname (str)
Pyxis service hostname.
- retries (int)
number of http retries for Pyxis requests.
- auth (PyxisAuth)
PyxisAuth subclass instance.
- backoff_factor (int)
backoff factor to apply between attempts after the second try.
- verify (bool)
enable/disable SSL CA verification.
- threads (int)
the number of threads to use for parallel requests.
- get_operator_indices(ocp_versions_range: str, organization: str | None = None) list[str] | Any[source]
Get a list of index images satisfying versioning and organization conditions.
- Args:
- ocp_versions_range (str)
Supported OCP versions range.
- organization (str)
Organization understood by IIB.
- Returns:
list: List of index images satisfying the conditions.
- get_repository_metadata(repo_name: str, custom_registry: str | None = None, only_internal: bool = False, only_partner: bool = False) dict[Any, Any] | Any[source]
Get metadata of a Comet repository.
If checking only one registry hasn’t been specified, check both with precedence on the internal registry.
- Args:
- repo_name (str):
Name of the repository.
- custom_registry (str):
Use a custom registry address instead of the default ones.
- only_internal (bool):
Whether to only check internal registry.
- only_partner (bool):
Whether to only check partner registry.
- Returns (dict):
Metadata of the repository.
- upload_signatures(signatures: list[str]) list[Any][source]
Upload signatures from given JSON string.
- Args:
- signatures [str]
JSON with signatures to upload. See Pyxis API for details.
- Returns:
list: List of uploaded signatures including auto-populated fields.
- _do_parallel_requests(make_request: Callable[[Any], Any], data_items: list[Any]) list[Any] | Any[source]
Call given function with given data items in parallel, collect responses.
- Args:
- make_request (function): a function that does the actual request.
Must accept a single argument: a data item. Must return a requests.models.Response object.
- data_items (list): a list of arbitrary objects to be passed
individually to make_request().
The number of parallel requests is defined by DEFAULT_REQUEST_THREADS_LIMIT (can be overridden by the user) and of course by the number of actually available threads.
If a response fails consistently (see PyxisSession for retry policy), the execution is terminated and an informative error is raised. See PyxisClient._handle_json_response() for details.
- Returns:
list(dict): list of dictionaries extracted from responses.
- _handle_json_response(response: Response) dict[Any, Any] | Any[source]
Get JSON from given response or raise an informative exception.
Uses requests.raise_for_status() but tries to extract more details.
- get_container_signatures(manifest_digests: str | None = None, references: str | None = None) list[str][source]
Get a list of signature metadata matching given fields.
- Args:
- manifest_digests (comma separated str)
manifest_digest used for searching in signatures.
- references (comma separated str)
pull reference for image of signature stored.
- Returns:
list: List of signature metadata matching given fields.