diff --git a/docs/source/item_processor.py b/docs/source/item_processor.py new file mode 100644 index 00000000..b2ed7a82 --- /dev/null +++ b/docs/source/item_processor.py @@ -0,0 +1,27 @@ +Item Processor +============== + +Class which process input push items and based on their metadata produce various information +needed for the push process. + +.. py.module pubtools._quay.item_processor + +.. autoclass:: SignEntry + :members: + +.. autoclass:: ManifestArchDigest + :members: + +.. autoclass:: ContentExtractor + :members: + +.. autoclass:: ReferenceProcessorInternal + :members: + +.. autoclass:: ReferenceProcessorExternal + :members: + +.. autoclass ItemProcessor + :members: + +""" diff --git a/docs/source/modules_reference.rst b/docs/source/modules_reference.rst index d6cce425..1d844d9b 100644 --- a/docs/source/modules_reference.rst +++ b/docs/source/modules_reference.rst @@ -17,9 +17,9 @@ API reference to modules used by pubtools-quay. Isolated usage outside of pubtoo quay_api_client quay_client quay_session - signature_handler - signature_remover + item_processor + signer_wrapper security_manifest_pusher tag_docker utilities - exceptions \ No newline at end of file + exceptions diff --git a/docs/source/signature_handler.rst b/docs/source/signature_handler.rst deleted file mode 100644 index 692b8eae..00000000 --- a/docs/source/signature_handler.rst +++ /dev/null @@ -1,36 +0,0 @@ -Signature Handler -================= - -Classes which handle container image signing. Base class, "SignatureHandler" contains common operations for all signing workflows. Children classes contain specific implementations for given workflows. - -.. py:module:: pubtools._quay.signature_handler - -.. autoclass:: SignatureHandler - - .. automethod:: __init__ - .. automethod:: src_quay_client - .. automethod:: create_manifest_claim_message - .. automethod:: get_tagged_image_digests - .. automethod:: get_signatures_from_pyxis - .. automethod:: remove_duplicate_claim_messages - .. automethod:: filter_claim_messages - .. automethod:: get_signatures_from_radas - .. automethod:: upload_signatures_to_pyxis - .. automethod:: validate_radas_messages - -.. autoclass:: ContainerSignatureHandler - - .. automethod:: construct_item_claim_messages - .. automethod:: construct_variant_claim_messages - .. automethod:: sign_container_images - -.. autoclass:: OperatorSignatureHandler - - .. automethod:: construct_index_image_claim_messages - .. automethod:: sign_operator_images - .. automethod:: sign_task_index_image - -.. autoclass:: BasicSignatureHandler - - .. automethod:: __init__ - .. automethod:: sign_claim_messages diff --git a/docs/source/signature_remover.rst b/docs/source/signature_remover.rst deleted file mode 100644 index de976bde..00000000 --- a/docs/source/signature_remover.rst +++ /dev/null @@ -1,18 +0,0 @@ -Signature Remover -================= - -Class used for removing unnecessary signatures. Multiple ways to delete signatures exist, based on a particular use-case. - -.. py:module:: pubtools._quay.signature_remover - -.. autoclass:: SignatureRemover - - .. automethod:: __init__ - .. automethod:: quay_client - .. automethod:: set_quay_client - .. automethod:: get_signatures_from_pyxis - .. automethod:: remove_signatures_from_pyxis - .. automethod:: get_repository_digests - .. automethod:: remove_repository_signatures - .. automethod:: remove_tag_signatures - .. automethod:: get_index_image_signatures diff --git a/docs/source/signer_wrapper.rst b/docs/source/signer_wrapper.rst new file mode 100644 index 00000000..76b47116 --- /dev/null +++ b/docs/source/signer_wrapper.rst @@ -0,0 +1,20 @@ +Signer Wrapper +============== + +Set of signing wrappers which wraps functionality of pubtools-sign projects and provides extra methods needed for the signign process. + +.. py:module:: pubtools._quay.signer_wrapper + +.. autoclass:: MsgSignerSettingsSchema + :members: + +.. autoclass:: SignerWrapper + :members: + +.. autoclass:: MsgSignerWrapper + :members: + +.. autoclass:: CosignSignerSettingsSchema + :members: + +.. autoclass:: CosignSignerWrapper diff --git a/pubtools/_quay/clear_repo.py b/pubtools/_quay/clear_repo.py index f3ea859f..1dc636b2 100644 --- a/pubtools/_quay/clear_repo.py +++ b/pubtools/_quay/clear_repo.py @@ -1,7 +1,7 @@ import argparse import logging import os -from typing import Any, Dict, List, cast +from typing import Any, Dict, List, cast, Optional from pubtools.pluggy import task_context, pm @@ -90,7 +90,7 @@ } -def clear_repositories(repositories: str, settings: dict[str, Any]) -> None: +def clear_repositories(repositories: str, settings: Dict[str, Any]) -> None: """ Clear Quay repository. @@ -178,7 +178,7 @@ def setup_args() -> argparse.ArgumentParser: return setup_arg_parser(CLEAR_REPO_ARGS) -def clear_repositories_main(sysargs: list[str] | None = None) -> None: +def clear_repositories_main(sysargs: Optional[List[str]] = None) -> None: """Entrypoint for clearing repositories.""" logging.basicConfig(level=logging.INFO) diff --git a/pubtools/_quay/command_executor.py b/pubtools/_quay/command_executor.py index 36ebc96a..67d87e94 100644 --- a/pubtools/_quay/command_executor.py +++ b/pubtools/_quay/command_executor.py @@ -11,7 +11,7 @@ import time import textwrap from types import TracebackType -from typing import Any, Optional, Type, Generator +from typing import Any, Optional, Type, Generator, List, Dict, Tuple from typing_extensions import Self import docker @@ -61,15 +61,15 @@ def __exit__( def _run_cmd( self, cmd: str, - err_msg: str | None = None, + err_msg: Optional[str] = None, tolerate_err: bool = False, - stdin: str | None = None, - ) -> tuple[str, str]: + stdin: Optional[str] = None, + ) -> Tuple[str, str]: """Run a bash command.""" raise NotImplementedError # pragma: no cover" def skopeo_login( - self, host: str = "quay.io", username: str | None = None, password: str | None = None + self, host: str = "quay.io", username: Optional[str] = None, password: Optional[str] = None ) -> None: """ Attempt to login to Quay if no login credentials are present. @@ -105,7 +105,7 @@ def skopeo_login( "STDOUT: '{0}', STDERR: '{1}'".format(out, err) ) - def tag_images(self, source_ref: str, dest_refs: list[str], all_arch: bool = False) -> None: + def tag_images(self, source_ref: str, dest_refs: List[str], all_arch: bool = False) -> None: """ Copy image from source to destination(s) using skopeo. @@ -157,7 +157,7 @@ def skopeo_inspect(self, image_ref: str, raw: bool = False) -> Any: class LocalExecutor(Executor): """Run commands locally.""" - def __init__(self, params: dict[str, Any] = {}) -> None: + def __init__(self, params: Dict[str, Any] = {}) -> None: """ Initialize. @@ -175,10 +175,10 @@ def __init__(self, params: dict[str, Any] = {}) -> None: def _run_cmd( self, cmd: str, - err_msg: str | None = None, + err_msg: Optional[str] = None, tolerate_err: bool = False, - stdin: str | None = None, - ) -> tuple[str, str]: + stdin: Optional[str] = None, + ) -> Tuple[str, str]: """ Run a command locally. @@ -215,10 +215,10 @@ class RemoteExecutor(Executor): def __init__( self, hostname: str, - username: str | None = None, - key_filename: str | None = None, - password: str | None = None, - port: int | None = None, + username: Optional[str] = None, + key_filename: Optional[str] = None, + password: Optional[str] = None, + port: Optional[int] = None, accept_unknown_host: bool = True, ) -> None: """ @@ -253,10 +253,10 @@ def __init__( def _run_cmd( self, cmd: str, - err_msg: str | None = None, + err_msg: Optional[str] = None, tolerate_err: bool = False, - stdin: str | None = None, - ) -> tuple[str, str]: + stdin: Optional[str] = None, + ) -> Tuple[str, str]: """ Run a command remotely via SSH. @@ -310,11 +310,11 @@ def __init__( self, image: str, base_url: str = "unix://var/run/docker.sock", - timeout: int | None = None, + timeout: Optional[int] = None, verify_tls: bool = False, - cert_path: str | None = None, - registry_username: str | None = None, - registry_password: str | None = None, + cert_path: Optional[str] = None, + registry_username: Optional[str] = None, + registry_password: Optional[str] = None, ) -> None: """ Initialize. @@ -340,7 +340,7 @@ def __init__( """ self.image = image - kwargs: dict[Any, Any] = {} + kwargs: Dict[Any, Any] = {} kwargs["base_url"] = base_url kwargs["version"] = "auto" if timeout: @@ -382,10 +382,10 @@ def __exit__( def _run_cmd( self, cmd: str, - err_msg: str | None = None, + err_msg: Optional[str] = None, tolerate_err: bool = False, - stdin: str | None = None, - ) -> tuple[str, str]: + stdin: Optional[str] = None, + ) -> Tuple[str, str]: """ Run a command locally. @@ -456,7 +456,7 @@ def _add_file(self, data: str, file_name: str) -> None: raise RuntimeError("File was not successfully added to the container") def skopeo_login( - self, host: str = "quay.io", username: str | None = None, password: str | None = None + self, host: str = "quay.io", username: Optional[str] = None, password: Optional[str] = None ) -> None: """ Attempt to login to Quay if no login credentials are present. diff --git a/pubtools/_quay/container_image_pusher.py b/pubtools/_quay/container_image_pusher.py index 095aa551..85f07739 100644 --- a/pubtools/_quay/container_image_pusher.py +++ b/pubtools/_quay/container_image_pusher.py @@ -2,7 +2,7 @@ import logging from concurrent import futures from concurrent.futures.thread import ThreadPoolExecutor -from typing import Any, cast +from typing import Any, cast, Dict, Optional, Union import requests @@ -33,7 +33,7 @@ class ContainerImagePusher: No validation is performed, push items are expected to be correct. """ - def __init__(self, push_items: list[Any], target_settings: dict[str, Any]) -> None: + def __init__(self, push_items: list[Any], target_settings: Dict[str, Any]) -> None: """ Initialize. @@ -47,8 +47,8 @@ def __init__(self, push_items: list[Any], target_settings: dict[str, Any]) -> No self.target_settings = target_settings self.quay_host = self.target_settings.get("quay_host", "quay.io").rstrip("/") - self._src_quay_client: QuayClient | None = None - self._dest_quay_client: QuayClient | None = None + self._src_quay_client: Optional[QuayClient] = None + self._dest_quay_client: Optional[QuayClient] = None @property def src_quay_client(self) -> QuayClient: @@ -74,7 +74,7 @@ def dest_quay_client(self) -> QuayClient: @classmethod def run_tag_images( - cls, source_ref: str, dest_refs: list[str], all_arch: bool, target_settings: dict[str, Any] + cls, source_ref: str, dest_refs: list[str], all_arch: bool, target_settings: Dict[str, Any] ) -> None: """ Prepare the "tag images" entrypoint with all the necessary arguments and run it. @@ -248,7 +248,7 @@ def copy_multiarch_push_item(self, push_item: Any, source_ml: ManifestList) -> N ) try: dest_ml = cast( - Manifest | ManifestList, self.dest_quay_client.get_manifest(dest_ref) + Union[Manifest, ManifestList], self.dest_quay_client.get_manifest(dest_ref) ) if dest_ml.get("mediaType") != QuayClient.MANIFEST_LIST_TYPE: LOG.warning( diff --git a/pubtools/_quay/hooks.py b/pubtools/_quay/hooks.py index 6f9183e1..a33b2c68 100644 --- a/pubtools/_quay/hooks.py +++ b/pubtools/_quay/hooks.py @@ -1,4 +1,5 @@ import sys +from typing import List from pubtools.pluggy import pm, hookspec @@ -7,7 +8,7 @@ @hookspec -def quay_repositories_cleared(repository_ids: list[str]) -> None: +def quay_repositories_cleared(repository_ids: List[str]) -> None: """Invoked after repositories have been cleared on Quay. :param repository_ids: ID of each cleared repository. @@ -16,7 +17,7 @@ def quay_repositories_cleared(repository_ids: list[str]) -> None: @hookspec -def quay_repositories_removed(repository_ids: list[str]) -> None: +def quay_repositories_removed(repository_ids: List[str]) -> None: """Invoked after repositories have been removed from Quay. :param repository_ids: ID of each removed repository. @@ -25,7 +26,7 @@ def quay_repositories_removed(repository_ids: list[str]) -> None: @hookspec -def quay_images_tagged(source_ref: str, dest_refs: list[str]) -> None: +def quay_images_tagged(source_ref: str, dest_refs: List[str]) -> None: """Invoked after tagging image(s) on Quay. :param source_ref: Source image reference. @@ -36,7 +37,7 @@ def quay_images_tagged(source_ref: str, dest_refs: list[str]) -> None: @hookspec -def quay_images_untagged(untag_refs: list[str], lost_refs: list[str]) -> None: +def quay_images_untagged(untag_refs: List[str], lost_refs: List[str]) -> None: """Invoked after untagging image(s) on Quay. :param untag_refs: Image references for which tags were removed. diff --git a/pubtools/_quay/iib_operations.py b/pubtools/_quay/iib_operations.py index a2875585..6a4c6ef3 100644 --- a/pubtools/_quay/iib_operations.py +++ b/pubtools/_quay/iib_operations.py @@ -1,7 +1,7 @@ import json import logging import sys -from typing import Any, List, Tuple, cast +from typing import Any, List, Tuple, cast, Dict from .container_image_pusher import ContainerImagePusher from .exceptions import InvalidTargetSettings @@ -106,7 +106,7 @@ def _index_image_to_sign_entries( dest_operator_quay_client = _get_operator_quay_client(target_settings) ret = cast( - Tuple[str, dict[str, str]], + Tuple[str, Dict[str, str]], dest_operator_quay_client.get_manifest( src_index_image, media_type=QuayClient.MANIFEST_LIST_TYPE, return_headers=True, raw=True ), @@ -178,7 +178,7 @@ def _sign_index_image( task_id: str, target_settings: dict[str, Any], pre_push: bool = False, -) -> list[tuple[str, str, str]]: +) -> List[Tuple[str, str, str]]: """Sign index image with configured signers for the target. Args: @@ -194,7 +194,7 @@ def _sign_index_image( to_sign_entries = _index_image_to_sign_entries( built_index_image, dest_tags, signing_keys, target_settings, internal=not pre_push ) - current_signatures: list[tuple[str, str, str]] = [ + current_signatures: List[Tuple[str, str, str]] = [ (e.reference, e.digest, e.signing_key) for e in to_sign_entries ] set_aws_kms_environment_variables(target_settings, "cosign_signer") diff --git a/pubtools/_quay/image_untagger.py b/pubtools/_quay/image_untagger.py index 7452e23d..0db3a971 100644 --- a/pubtools/_quay/image_untagger.py +++ b/pubtools/_quay/image_untagger.py @@ -8,7 +8,7 @@ from .security_manifest_pusher import SecurityManifestPusher from .types import Manifest, ManifestList -from typing import cast, List, Dict +from typing import cast, List, Dict, Optional, Tuple, Union LOG = logging.getLogger("pubtools.quay") @@ -21,9 +21,9 @@ def __init__( references: list[str], quay_api_token: str, remove_last: bool = False, - quay_user: str | None = None, - quay_password: str | None = None, - host: str | None = None, + quay_user: Optional[str] = None, + quay_password: Optional[str] = None, + host: Optional[str] = None, ) -> None: """ Initialize. @@ -56,7 +56,9 @@ def __init__( self.remove_last = remove_last if quay_user and quay_password: - self._quay_client: QuayClient | None = QuayClient(quay_user, quay_password, self.host) + self._quay_client: Optional[QuayClient] = QuayClient( + quay_user, quay_password, self.host + ) else: self._quay_client = None self._quay_api_client = QuayApiClient(quay_api_token, self.host) @@ -90,7 +92,7 @@ def get_repository_tags_mapping(self) -> dict[str, list[str]]: def construct_tag_digest_mappings( self, repository: str - ) -> tuple[dict[str, list[str]], dict[str, list[str]]]: + ) -> Tuple[Dict[str, List[str]], Dict[str, List[str]]]: """ Create a mappings of tags->digests as well as digests->tags. @@ -118,7 +120,8 @@ def construct_tag_digest_mappings( image = image_schema.format(self.host, repository, tag) try: manifest = cast( - ManifestList | Manifest, cast(QuayClient, self._quay_client).get_manifest(image) + Union[ManifestList, Manifest], + cast(QuayClient, self._quay_client).get_manifest(image), ) except requests.exceptions.HTTPError as e: # Just removed tags could still be in tags list while manifests are removed @@ -185,7 +188,10 @@ def get_lost_digests( return lost_digests def get_repo_cosign_images( - self, repo_images: list[str], repo_tags: list[str], image_types: Iterable[str] | None = None + self, + repo_images: list[str], + repo_tags: list[str], + image_types: Optional[Iterable[str]] = None, ) -> set[str]: """ Get a list of images generated by cosign associated with the provided images. diff --git a/pubtools/_quay/item_processor.py b/pubtools/_quay/item_processor.py index b0407b2f..caa3e19f 100644 --- a/pubtools/_quay/item_processor.py +++ b/pubtools/_quay/item_processor.py @@ -1,7 +1,7 @@ from dataclasses import dataclass import hashlib import logging -from typing import List, Dict, Any, Optional, TypeAlias, Tuple, cast, Callable +from typing import List, Dict, Any, Optional, TypeAlias, Tuple, cast, Callable, Union import requests import time import json @@ -172,7 +172,8 @@ def _extract_manifest( _MEDIA_TYPES_PROCESS: dict[ str, Callable[ - [Any, str, str, str, Dict[str, Any]], ManifestArchDigest | List[ManifestArchDigest] + [Any, str, str, str, Dict[str, Any]], + Union[ManifestArchDigest, List[ManifestArchDigest]], ], ] = { QuayClient.MANIFEST_LIST_TYPE: _extract_ml_manifest, @@ -183,7 +184,8 @@ def _extract_manifest( _MEDIA_TYPES_PROCESS_FULL: dict[ str, Callable[ - [Any, str, str, str, Dict[str, Any]], ManifestArchDigest | List[ManifestArchDigest] + [Any, str, str, str, Dict[str, Any]], + Union[ManifestArchDigest, List[ManifestArchDigest]], ], ] = { QuayClient.MANIFEST_LIST_TYPE: _extract_ml_manifest_full, @@ -254,9 +256,9 @@ def extract_manifests( if not manifest: continue else: - mret: ManifestArchDigest | List[ManifestArchDigest] = MEDIA_TYPES_PROCESS[mtype]( - self, image_ref, manifest, mtype, ret_headers - ) + mret: Union[ManifestArchDigest, List[ManifestArchDigest]] = MEDIA_TYPES_PROCESS[ + mtype + ](self, image_ref, manifest, mtype, ret_headers) if isinstance(mret, list): results.extend(mret) else: @@ -264,7 +266,7 @@ def extract_manifests( seen = set() return [x for x in results if all([x not in seen, not (seen.add(x))])] # type: ignore - def extract_tags(self, repo_ref: str, tolerate_missing: bool = True) -> list[str]: + def extract_tags(self, repo_ref: str, tolerate_missing: bool = True) -> List[str]: """Fetch list of tags for given repo reference. Args: @@ -387,7 +389,7 @@ class ItemProcesor: """ extractor: ContentExtractor - reference_processor: ReferenceProcessorExternal | ReferenceProcessorInternal + reference_processor: Union[ReferenceProcessorExternal, ReferenceProcessorInternal] reference_registries: List[str] source_registry: Optional[str] @@ -548,7 +550,7 @@ def generate_to_unsign(self, item: PushItem) -> List[Dict[str, Any]]: def generate_existing_tags( self, item: PushItem, tolerate_missing: bool = True - ) -> List[Tuple[str, str, str | None]]: + ) -> List[Tuple[str, str, Union[str, None]]]: """Generate list of existing tags for given push item. Args: @@ -557,7 +559,7 @@ def generate_existing_tags( Returns: list: List of tuples containing registry, repository and tag. """ - existing_tag_entries: list[tuple[str, str, str | None]] = [] + existing_tag_entries: List[Tuple[str, str, Union[str, None]]] = [] for repo in self._generate_src_repo(item): ref_repo = self.reference_processor.replace_repo(repo) tags = self.extractor.extract_tags(ref_repo, tolerate_missing=tolerate_missing) @@ -568,7 +570,7 @@ def generate_existing_tags( return existing_tag_entries def _generate_existing_manifests( - self, item: PushItem, only_media_types: list[str] | None = None + self, item: PushItem, only_media_types: Union[List[str], None] = None ) -> List[Tuple[str, str, Optional[ManifestArchDigest]]]: """Generate list of existing manifests data for given push item. @@ -578,7 +580,7 @@ def _generate_existing_manifests( Returns: list: List of tuples containing repository, tag and ManifestArchDigest. """ - existing_manifests: list[tuple[str, str, ManifestArchDigest | None]] = [] + existing_manifests: List[Tuple[str, str, Union[ManifestArchDigest, None]]] = [] if not only_media_types: media_types = [ QuayClient.MANIFEST_LIST_TYPE, @@ -611,8 +613,8 @@ def _generate_existing_manifests( return existing_manifests def generate_existing_manifests_map( - self, item: PushItem, only_media_types: list[str] | None = None - ) -> Dict[str, Dict[str, Dict[str, List[ManifestArchDigest] | None]]]: + self, item: PushItem, only_media_types: Union[List[str], None] = None + ) -> Dict[str, Dict[str, Dict[str, Union[List[ManifestArchDigest], None]]]]: """Generate existing manifests map for given push item. Args: @@ -621,7 +623,9 @@ def generate_existing_manifests_map( Returns: dict: Dict of {registry: {repo: {tag: []}}} """ - mapping_existing: dict[str, dict[str, dict[str, List[ManifestArchDigest] | None]]] = {} + mapping_existing: Dict[str, Dict[str, dict[str, Union[List[ManifestArchDigest], None]]]] = ( + {} + ) for repo, tag, mad in self._generate_existing_manifests( item, only_media_types=only_media_types ): @@ -640,8 +644,8 @@ def generate_existing_manifests_map( return mapping_existing def generate_existing_manifests_metadata( - self, item: PushItem, only_media_types: list[str] | None = None - ) -> List[Tuple[str, str, ManifestArchDigest | None]]: + self, item: PushItem, only_media_types: Union[List[str], None] = None + ) -> List[Tuple[str, str, Union[ManifestArchDigest, None]]]: """Generate list of existing manifests for given push item. Args: @@ -659,7 +663,7 @@ def generate_existing_manifests_metadata( def generate_all_existing_manifests_metadata( self, item: PushItem - ) -> List[Tuple[str, str, ManifestArchDigest | None]]: + ) -> List[Tuple[str, str, Union[ManifestArchDigest, None]]]: """Return manifests for all existing tags in all repositories for given push item. Args: @@ -667,7 +671,7 @@ def generate_all_existing_manifests_metadata( Returns: list: List of tuples containing repository, tag and ManifestArchDigest. """ - repo_tags_map: dict[str, list[str | None]] = {} + repo_tags_map: dict[str, List[Union[str, None]]] = {} for registry, repo, tag in self.generate_existing_tags(item): repo_tags_map.setdefault(repo, []).append(tag) item2 = VirtualPushItem( @@ -678,7 +682,7 @@ def generate_all_existing_manifests_metadata( def item_processor_for_external_data( - quay_client: QuayClient, external_registries: list[str], retry_sleep_time: int + quay_client: QuayClient, external_registries: List[str], retry_sleep_time: int ) -> ItemProcesor: """Get instance of item processor configured to produce destination data. diff --git a/pubtools/_quay/manifest_claims_handler.py b/pubtools/_quay/manifest_claims_handler.py index db2698b6..9e6de608 100644 --- a/pubtools/_quay/manifest_claims_handler.py +++ b/pubtools/_quay/manifest_claims_handler.py @@ -2,8 +2,7 @@ import logging import os import json -from typing import Any, Callable, Sequence, Mapping, cast, Optional - +from typing import Any, Callable, Sequence, Mapping, cast, Optional, List, Dict import monotonic import proton from proton.handlers import MessagingHandler @@ -38,7 +37,7 @@ class UMBSettings(object): def __init__( self, - broker_urls: list[str], + broker_urls: List[str], radas_address: str = "VirtualTopic.eng.robosignatory.container.sign", pub_cert: str = "/etc/pub/umb-pub-cert-key.pem", ca_cert: str = "/etc/pki/tls/certs/ca-bundle.crt", @@ -90,7 +89,7 @@ def __str__(self) -> str: return repr(self) -def _get_endpoint_error_condition(endpoint: proton.Endpoint) -> proton.Condition | None: +def _get_endpoint_error_condition(endpoint: proton.Endpoint) -> Optional[proton.Condition]: """Return the error condition on this endpoint if there is one. If there's no local error condition, return the remote error @@ -160,7 +159,7 @@ class ManifestClaimsHandler(MessagingHandler): # type: ignore def __init__( self, settings: UMBSettings, - claim_messages: list[dict[str, Any]], + claim_messages: List[dict[str, Any]], message_sender_callback: Callable[[Any], Any], on_message_callback: Callable[[Any], None] = do_nothing, on_error_callback: Callable[[Any], None] = raise_error, @@ -341,7 +340,7 @@ def _endpoint_error_if_unhandled(self, event: proton.Event, endpoint: proton.End if error_tag in self._PROTON_KNOWN_UNHANDLED_ERRORS: self._endpoint_error(event, endpoint) - def _send_message(self, count: int | None = None) -> None: + def _send_message(self, count: Optional[int] = None) -> None: if not self.to_send: # pragma: no cover return @@ -384,7 +383,7 @@ class _ManifestClaimsRunner(object): def __init__( self, settings: UMBSettings, - claim_messages: list[dict[str, Any]], + claim_messages: List[Dict[str, Any]], send_action: Callable[[Sequence[Mapping[str, Any]]], None], ): self._settings = settings @@ -433,7 +432,7 @@ def start(self) -> None: # pragma: no cover """Start manifest claim messaging for the first time.""" self._run(self._claim_messages) - def _run(self, claims: list[dict[str, Any]]) -> None: # pragma: no cover + def _run(self, claims: List[Dict[str, Any]]) -> None: # pragma: no cover """Run manifest claim messaging with the given set of claims.""" handler = ManifestClaimsHandler( self._settings, diff --git a/pubtools/_quay/manifest_list_merger.py b/pubtools/_quay/manifest_list_merger.py index 02fb9afd..b634cec8 100644 --- a/pubtools/_quay/manifest_list_merger.py +++ b/pubtools/_quay/manifest_list_merger.py @@ -1,6 +1,6 @@ from copy import deepcopy import logging -from typing import List, cast +from typing import List, cast, Optional import requests @@ -17,12 +17,12 @@ def __init__( self, src_image: str, dest_image: str, - src_quay_host: str | None = None, - src_quay_username: str | None = None, - src_quay_password: str | None = None, - dest_quay_username: str | None = None, - dest_quay_password: str | None = None, - host: str | None = None, + src_quay_host: Optional[str] = None, + src_quay_username: Optional[str] = None, + src_quay_password: Optional[str] = None, + dest_quay_username: Optional[str] = None, + dest_quay_password: Optional[str] = None, + host: Optional[str] = None, ) -> None: """ Initialize. @@ -48,13 +48,13 @@ def __init__( self.src_image = src_image self.dest_image = dest_image if src_quay_username and src_quay_password: - self._src_quay_client: QuayClient | None = QuayClient( + self._src_quay_client: Optional[QuayClient] = QuayClient( src_quay_username, src_quay_password, src_quay_host or host ) else: self._src_quay_client = None if dest_quay_username and dest_quay_password: - self._dest_quay_client: QuayClient | None = QuayClient( + self._dest_quay_client: Optional[QuayClient] = QuayClient( dest_quay_username, dest_quay_password, host ) else: @@ -155,7 +155,7 @@ def _add_missing_architectures( return new_manifest_list def merge_manifest_lists_selected_architectures( - self, eligible_archs: list[str] + self, eligible_archs: List[str] ) -> ManifestList: """ Merge manifests lists. Only specified archs are eligible for merging. diff --git a/pubtools/_quay/merge_manifest_list.py b/pubtools/_quay/merge_manifest_list.py index 931b8198..2c3d9e87 100644 --- a/pubtools/_quay/merge_manifest_list.py +++ b/pubtools/_quay/merge_manifest_list.py @@ -1,5 +1,6 @@ import logging import argparse +from typing import Optional, List from pubtools.pluggy import task_context @@ -62,7 +63,7 @@ def setup_args() -> argparse.ArgumentParser: return setup_arg_parser(MERGE_MANIFEST_LIST_ARGS) -def merge_manifest_list_main(sysargs: list[str] | None = None) -> None: +def merge_manifest_list_main(sysargs: Optional[List[str]] = None) -> None: """Entrypoint for manifest list merging.""" logging.basicConfig(level=logging.INFO) diff --git a/pubtools/_quay/models.py b/pubtools/_quay/models.py index 68e95feb..cb13ed08 100644 --- a/pubtools/_quay/models.py +++ b/pubtools/_quay/models.py @@ -1,14 +1,14 @@ import dataclasses -from typing import Dict, Any, List +from typing import Dict, Any, List, Optional @dataclasses.dataclass class BuildIndexImageParam: """Parameter data for building index image and part of data required by iib_results.""" - bundles: list[str] + bundles: List[str] index_image: str - deprecation_list: List[str] | None + deprecation_list: Optional[List[str]] build_tags: List[str] target_settings: Dict[str, Any] tag: str diff --git a/pubtools/_quay/operator_pusher.py b/pubtools/_quay/operator_pusher.py index db333a8f..556d6d8a 100644 --- a/pubtools/_quay/operator_pusher.py +++ b/pubtools/_quay/operator_pusher.py @@ -5,7 +5,7 @@ from concurrent import futures from concurrent.futures.thread import ThreadPoolExecutor import os -from typing import Any +from typing import Any, Union, Optional, List, Dict, Tuple import requests from requests.adapters import HTTPAdapter @@ -59,7 +59,7 @@ def __init__( self.ocp_versions_resolved: dict[str, list[str]] = {} @staticmethod - def _get_immutable_tag(push_item: Any) -> str | Any: + def _get_immutable_tag(push_item: Any) -> Union[str, Any]: """ Return immutable tag from operator push item production tags. @@ -169,7 +169,7 @@ def version_items_mapping(self) -> dict[str, list[Any]]: return self._version_items_mapping - def get_deprecation_list(self, version: str) -> list[str] | None: + def get_deprecation_list(self, version: str) -> Optional[List[str]]: """ Get bundles to be deprecated in the index image. @@ -236,8 +236,8 @@ def _get_requests_session() -> requests.Session: @classmethod def pubtools_iib_get_common_args( - cls, target_settings: dict[str, Any] - ) -> tuple[list[str], dict[str, str]]: + cls, target_settings: Dict[str, Any] + ) -> Tuple[list[str], Dict[str, str]]: """ Create an argument list common for all pubtools-iib operations. @@ -272,11 +272,11 @@ def pubtools_iib_get_common_args( @tx.instrument_func() def iib_add_bundles( cls, - bundles: list[str] | None = None, - archs: list[str] | None = None, - index_image: str | None = None, - deprecation_list: list[str] | str | None = None, - build_tags: list[str] | None = None, + bundles: Optional[List[str]] = None, + archs: Optional[List[str]] = None, + index_image: Optional[str] = None, + deprecation_list: Union[List[str], str, None] = None, + build_tags: Optional[List[str]] = None, target_settings: dict[str, Any] = {}, ) -> Any: """ @@ -335,11 +335,11 @@ def iib_add_bundles( @tx.instrument_func() def iib_remove_operators( cls, - operators: list[str] | None = None, - archs: list[str] | None = None, - index_image: str | None = None, - build_tags: list[str] | None = None, - target_settings: dict[str, Any] = {}, + operators: Optional[List[str]] = None, + archs: Optional[List[str]] = None, + index_image: Optional[str] = None, + build_tags: Optional[List[str]] = None, + target_settings: Dict[str, Any] = {}, ) -> Any: """ Construct and execute pubtools-iib command to remove operators from index image. @@ -385,7 +385,7 @@ def iib_remove_operators( env_vars, ) - def get_existing_index_images(self, quay_client: QuayClient) -> list[tuple[str, str, str]]: + def get_existing_index_images(self, quay_client: QuayClient) -> List[Tuple[str, str, str]]: """ Return existing index images for push items. @@ -457,7 +457,7 @@ def ensure_bundles_present(self) -> bool: LOG.info("Bundle {0} is present".format(bundle)) return True - def _get_fbc_opted_in_items(self) -> tuple[dict[int, bool], dict[int, bool]]: + def _get_fbc_opted_in_items(self) -> Tuple[Dict[int, bool], Dict[int, bool]]: """Get items that are opted in for fbc. An item needs to be targeted for repos with fbc_opt_in set to True and @@ -749,7 +749,9 @@ def build_index_images(self) -> dict[str, dict[Any, Any]]: return iib_results @log_step("Push index images to Quay") - def push_index_images(self, iib_results: dict[Any, Any], tag_suffix: str | None = None) -> None: + def push_index_images( + self, iib_results: dict[Any, Any], tag_suffix: Optional[str] = None + ) -> None: """ Push index images which were built in the previous stage to Quay. diff --git a/pubtools/_quay/push_docker.py b/pubtools/_quay/push_docker.py index 1ef0b3e4..91b5451c 100644 --- a/pubtools/_quay/push_docker.py +++ b/pubtools/_quay/push_docker.py @@ -2,7 +2,7 @@ import logging import sys import json -from typing import Tuple, List, Dict, Any, cast +from typing import Tuple, List, Dict, Any, cast, Optional import requests import urllib3 @@ -51,7 +51,7 @@ class PushDocker: def __init__( self, - push_items: list[Any], + push_items: List[Any], hub: Any, task_id: str, target_name: str, @@ -82,10 +82,10 @@ def __init__( self.quay_host = self.target_settings.get("quay_host", "quay.io").rstrip("/") - self._src_quay_client: QuayClient | None = None - self._dest_quay_client: QuayClient | None = None - self._dest_operator_quay_client: QuayClient | None = None - self._dest_quay_api_client: QuayApiClient | None = None + self._src_quay_client: Optional[QuayClient] = None + self._dest_quay_client: Optional[QuayClient] = None + self._dest_operator_quay_client: Optional[QuayClient] = None + self._dest_quay_api_client: Optional[QuayApiClient] = None self._index_image_quay_client = None self.dest_registries = self.target_settings["docker_settings"]["docker_reference_registry"] self.dest_registries = ( @@ -172,7 +172,7 @@ def verify_target_settings(self) -> None: ) @log_step("Get container push items") - def get_docker_push_items(self) -> list[Any]: + def get_docker_push_items(self) -> List[Any]: """ Filter push items to only include docker ones. @@ -212,7 +212,7 @@ def get_docker_push_items(self) -> list[Any]: return docker_push_items @log_step("Get operator push items") - def get_operator_push_items(self) -> list[Any]: + def get_operator_push_items(self) -> List[Any]: """ Filter out push items to only include operator ones. @@ -250,7 +250,7 @@ def get_operator_push_items(self) -> list[Any]: @classmethod def check_repos_validity( - cls, push_items: list[Any], hub: Any, target_settings: dict[str, Any] + cls, push_items: List[Any], hub: Any, target_settings: dict[str, Any] ) -> None: """ Check if specified repos are valid and pushing to them is allowed. @@ -382,8 +382,8 @@ def generate_backup_mapping( ] else: arch_mads = cast(List[ManifestArchDigest], man_arch_digs) - v2list_mads: List[ManifestArchDigest | None] = cast( - List[ManifestArchDigest | None], + v2list_mads: List[Optional[ManifestArchDigest]] = cast( + List[Optional[ManifestArchDigest]], [ m for m in cast(List[ManifestArchDigest], man_arch_digs) @@ -391,13 +391,13 @@ def generate_backup_mapping( ] or [None], ) - v2s1_mads: List[ManifestArchDigest | None] = cast( - List[ManifestArchDigest | None], + v2s1_mads: List[Optional[ManifestArchDigest]] = cast( + List[Optional[ManifestArchDigest]], [m for m in arch_mads if m.type_ == QuayClient.MANIFEST_V2S1_TYPE] or [None], ) - v2s2_mads: List[ManifestArchDigest | None] = cast( - List[ManifestArchDigest | None], + v2s2_mads: List[Optional[ManifestArchDigest]] = cast( + List[Optional[ManifestArchDigest]], [m for m in arch_mads if m.type_ == QuayClient.MANIFEST_V2S2_TYPE] or [None], ) @@ -436,7 +436,7 @@ def generate_backup_mapping( return (backup_tags, rollback_tags) @log_step("Perform rollback") - def rollback(self, backup_tags: dict[ImageData, str], rollback_tags: list[ImageData]) -> None: + def rollback(self, backup_tags: dict[ImageData, str], rollback_tags: List[ImageData]) -> None: """ Perform a rollback. @@ -487,8 +487,8 @@ def get_outdated_manifests( return outdated_signatures def fetch_missing_push_items_digests( - self, push_items: list[Any] - ) -> dict[str, dict[str, dict[str, dict[str, tuple[str, Any]]]]]: + self, push_items: List[Any] + ) -> Dict[str, Dict[str, Dict[str, Dict[str, Tuple[str, Any]]]]]: """Fetch digests for media types which weren't originally pushed. In order to be able to sign v2s1 for images which were pushed as @@ -507,7 +507,7 @@ def fetch_missing_push_items_digests( self.target_settings["quay_namespace"], ) - new_digests: dict[str, dict[str, dict[str, dict[str, tuple[str, Any]]]]] = {} + new_digests: Dict[str, Dict[str, Dict[str, Dict[str, Tuple[str, Any]]]]] = {} for item in push_items: missing_media_types = set( [QuayClient.MANIFEST_V2S2_TYPE, QuayClient.MANIFEST_V2S1_TYPE] @@ -532,7 +532,7 @@ def fetch_missing_push_items_digests( ) return new_digests - def sign_new_manifests(self, docker_push_items: list[Any]) -> list[tuple[str, str, Any]]: + def sign_new_manifests(self, docker_push_items: List[Any]) -> List[tuple[str, str, Any]]: """Sign newly pushed images with signers enabled in target settings. Args: @@ -841,7 +841,7 @@ def run(self) -> None: def mod_entry_point( - push_items: list[Any], hub: Any, task_id: str, target_name: str, target_settings: dict[str, Any] + push_items: List[Any], hub: Any, task_id: str, target_name: str, target_settings: dict[str, Any] ) -> None: """Entry point for use in another python code.""" push = PushDocker(push_items, hub, task_id, target_name, target_settings) diff --git a/pubtools/_quay/quay_api_client.py b/pubtools/_quay/quay_api_client.py index dbf40ddb..30028341 100644 --- a/pubtools/_quay/quay_api_client.py +++ b/pubtools/_quay/quay_api_client.py @@ -1,4 +1,5 @@ import logging +from typing import Optional import requests @@ -10,7 +11,7 @@ class QuayApiClient: """Class for performing Quay REST API queries.""" - def __init__(self, token: str, host: str | None = None) -> None: + def __init__(self, token: str, host: Optional[str] = None) -> None: """ Initialize. diff --git a/pubtools/_quay/quay_client.py b/pubtools/_quay/quay_client.py index e30d6e24..7674b743 100644 --- a/pubtools/_quay/quay_client.py +++ b/pubtools/_quay/quay_client.py @@ -6,7 +6,7 @@ from urllib3.util.retry import Retry from urllib import request import threading -from typing import Any, cast, Dict, List, Tuple +from typing import Any, cast, Dict, List, Tuple, Optional, Union from .exceptions import ManifestTypeError, RegistryAuthError, ManifestNotFoundError from .quay_session import QuaySession @@ -25,7 +25,9 @@ class QuayClient: MANIFEST_OCI_LIST_TYPE = "application/vnd.oci.image.index.v1+json" MANIFEST_OCI_V2S2_TYPE = "application/vnd.oci.image.manifest.v1+json" - def __init__(self, username: str | None, password: str | None, host: str | None = None) -> None: + def __init__( + self, username: Optional[str], password: Optional[str], host: Optional[str] = None + ) -> None: """ Initialize. @@ -43,7 +45,7 @@ def __init__(self, username: str | None, password: str | None, host: str | None self.thread_local = threading.local() @property - def session(self) -> Any | QuaySession: + def session(self) -> Union[Any, QuaySession]: """Create QuaySession object per thread.""" if not hasattr(self.thread_local, "session"): self.thread_local.session = QuaySession(hostname=self.host, api="docker") @@ -53,9 +55,9 @@ def get_manifest( self, image: str, raw: bool = False, - media_type: str | None = None, + media_type: Optional[str] = None, return_headers: bool = False, - ) -> ManifestList | Manifest | str | Tuple[str, CaseInsensitiveDict[str]]: + ) -> Union[ManifestList, Manifest, str, Tuple[str, CaseInsensitiveDict[str]]]: """ Get manifest of given media type. @@ -126,7 +128,7 @@ def get_manifest( else: return cast(ManifestList, response.json()) - def get_manifest_digest(self, image: str, media_type: str | None = None) -> str: + def get_manifest_digest(self, image: str, media_type: Optional[str] = None) -> str: """ Get manifest of the specified image and calculate its digest by hashing it. @@ -153,7 +155,9 @@ def get_manifest_digest(self, image: str, media_type: str | None = None) -> str: return "sha256:{0}".format(digest) - def upload_manifest(self, manifest: ManifestList | str, image: str, raw: bool = False) -> None: + def upload_manifest( + self, manifest: Union[ManifestList, str], image: str, raw: bool = False + ) -> None: """ Upload manifest to a specified image. @@ -186,7 +190,9 @@ def upload_manifest(self, manifest: ManifestList | str, image: str, raw: bool = } self._request_quay("PUT", endpoint, kwargs) - def get_repository_tags(self, repository: str, raw: bool = False) -> str | Dict[str, List[str]]: + def get_repository_tags( + self, repository: str, raw: bool = False + ) -> Union[str, Dict[str, List[str]]]: """ Get tags of a provided repository. @@ -253,7 +259,7 @@ def _request_quay( return r def _authenticate_quay( - self, headers: dict[Any, Any] | requests.structures.CaseInsensitiveDict[Any] + self, headers: Union[dict[Any, Any], requests.structures.CaseInsensitiveDict[Any]] ) -> None: """ Attempt to perform an authentication with registry's authentication server. @@ -306,7 +312,7 @@ def _authenticate_quay( raise RegistryAuthError("Authentication server response doesn't contain a token.") self.session.set_auth_token(r.json()["token"]) - def _parse_and_validate_image_url(self, image: str) -> tuple[str, str]: + def _parse_and_validate_image_url(self, image: str) -> Tuple[str, str]: """ Extract image repository + reference from an image and validate its data. diff --git a/pubtools/_quay/quay_session.py b/pubtools/_quay/quay_session.py index 4c05645e..35252095 100644 --- a/pubtools/_quay/quay_session.py +++ b/pubtools/_quay/quay_session.py @@ -1,4 +1,4 @@ -from typing import Any +from typing import Any, Union import requests from requests.adapters import HTTPAdapter @@ -11,7 +11,7 @@ class QuaySession(object): def __init__( self, - hostname: str | None = None, + hostname: Union[str, None] = None, retries: int = 3, backoff_factor: int = 2, verify: bool = False, diff --git a/pubtools/_quay/remove_repo.py b/pubtools/_quay/remove_repo.py index 110cb186..25dfa44b 100644 --- a/pubtools/_quay/remove_repo.py +++ b/pubtools/_quay/remove_repo.py @@ -1,7 +1,7 @@ import os import logging import argparse -from typing import Any +from typing import Any, Optional, List from pubtools.pluggy import pm, task_context @@ -165,7 +165,7 @@ def setup_args() -> argparse.ArgumentParser: return setup_arg_parser(REMOVE_REPO_ARGS) -def remove_repositories_main(sysargs: list[str] | None = None) -> None: +def remove_repositories_main(sysargs: Optional[List[str]] = None) -> None: """Entrypoint for removing repositories.""" logging.basicConfig(level=logging.INFO) diff --git a/pubtools/_quay/security_manifest_pusher.py b/pubtools/_quay/security_manifest_pusher.py index 6ac97da0..fc981a45 100644 --- a/pubtools/_quay/security_manifest_pusher.py +++ b/pubtools/_quay/security_manifest_pusher.py @@ -8,7 +8,7 @@ from dataclasses import dataclass from concurrent import futures from concurrent.futures.thread import ThreadPoolExecutor -from typing import Any, cast +from typing import Any, cast, List, Union, Optional from .quay_client import QuayClient from .utils.misc import get_internal_container_repo_name, log_step @@ -33,7 +33,7 @@ class SecurityManifestPusher: COSIGN_TRIANGULATE_TYPES = ("attestation", "sbom", "signature") - def __init__(self, push_items: list[object], target_settings: dict[str, Any]) -> None: + def __init__(self, push_items: List[object], target_settings: dict[str, Any]) -> None: """ Initialize. @@ -50,9 +50,9 @@ def __init__(self, push_items: list[object], target_settings: dict[str, Any]) -> self.cosign_public_key_path = self.target_settings["cosign_public_key_path"] self.quay_host = self.target_settings.get("quay_host", "quay.io").rstrip("/") - self._src_quay_client: QuayClient | None = None - self._dest_quay_client: QuayClient | None = None - self._dest_quay_api_client: QuayApiClient | None = None + self._src_quay_client: Optional[QuayClient] = None + self._dest_quay_client: Optional[QuayClient] = None + self._dest_quay_api_client: Optional[QuayApiClient] = None @property def src_quay_client(self) -> QuayClient: @@ -112,7 +112,7 @@ def cosign_get_existing_attestation( self, image_ref: str, output_file: str, - rekor_url: str | None = None, + rekor_url: Optional[str] = None, skip_verify_rekor: bool = False, ) -> bool: """ @@ -158,7 +158,7 @@ def cosign_attest_security_manifest( self, security_manifest_path: str, image_ref: str, - rekor_url: str | None = None, + rekor_url: Optional[str] = None, skip_upload_rekor: bool = False, ) -> None: """ @@ -243,7 +243,7 @@ def cosign_triangulate_image( with open(reference_file, "r") as f: return f.read().strip() - def get_security_manifest_from_attestation(self, file_path: str) -> dict[Any, Any] | Any: + def get_security_manifest_from_attestation(self, file_path: str) -> Union[dict[Any, Any], Any]: """ Parse image attestation and extract a security manifest. diff --git a/pubtools/_quay/signer_wrapper.py b/pubtools/_quay/signer_wrapper.py index a196abf5..15fbb964 100644 --- a/pubtools/_quay/signer_wrapper.py +++ b/pubtools/_quay/signer_wrapper.py @@ -49,7 +49,7 @@ class SignerWrapper: entry_point_conf = ["signer", "group", "signer"] def __init__( - self, config_file: Optional[str] = None, settings: Dict[str, Any] | None = None + self, config_file: Optional[str] = None, settings: Optional[Dict[str, Any]] = None ) -> None: """Initialize SignerWrapper. @@ -194,7 +194,7 @@ def sign_containers( with redirect_stdout(io.StringIO()): self._sign_containers(to_sign_entries_filtered, task_id) - def validate_settings(self, settings: Dict[str, Any] | None = None) -> None: + def validate_settings(self, settings: Optional[Dict[str, Any]] = None) -> None: """Validate provided settings for the SignerWrapper.""" settings = settings or self.settings schema = self.SCHEMA(unknown=EXCLUDE) diff --git a/pubtools/_quay/tag_docker.py b/pubtools/_quay/tag_docker.py index e9b8f8cf..b0d8864d 100644 --- a/pubtools/_quay/tag_docker.py +++ b/pubtools/_quay/tag_docker.py @@ -4,7 +4,7 @@ import json import logging import urllib3 -from typing import Any, cast +from typing import Any, cast, Optional, Union, Tuple, List import requests @@ -72,7 +72,7 @@ def __init__( self.target_name = target_name self.target_settings = target_settings - self._quay_client: QuayClient | None = None + self._quay_client: Optional[QuayClient] = None self.quay_host = self.target_settings.get("quay_host", "quay.io").rstrip("/") @@ -197,7 +197,7 @@ def check_input_validity(self) -> None: ) ) - def get_image_details(self, reference: str, executor: Executor) -> ImageDetails | None: + def get_image_details(self, reference: str, executor: Executor) -> Optional[ImageDetails]: """ Create an ImageDetails namedtuple for the given image reference. @@ -206,12 +206,12 @@ def get_image_details(self, reference: str, executor: Executor) -> ImageDetails Image reference. executor (Executor): Instance of Executor subclass used for skopeo inspect. - Returns (ImageDetails|None): + Returns (Optional[ImageDetails]): Namedtuple filled with images data, or None if image doesn't exist. """ LOG.info("Getting image details of {0}".format(reference)) try: - manifest = cast(Manifest | ManifestList, self.quay_client.get_manifest(reference)) + manifest = cast(Union[Manifest, ManifestList], self.quay_client.get_manifest(reference)) except requests.exceptions.HTTPError as e: if e.response.status_code == 404 or e.response.status_code == 401: LOG.info("Image '{0}' doesn't exist".format(reference)) @@ -259,7 +259,7 @@ def is_arch_relevant(self, push_item: Any, arch: str) -> bool: def tag_remove_calculate_archs( self, push_item: Any, tag: str, executor: Executor - ) -> tuple[list[str], list[str]]: + ) -> Tuple[List[str], List[str]]: """ Calculate which architectures would be removed, and which would remain from a given tag. @@ -321,8 +321,8 @@ def tag_remove_calculate_archs( return ([], []) # pragma: no cover def tag_remove_calculate_archs_source_image( - self, push_item: Any, source_details: ImageDetails | None, dest_details: ImageDetails - ) -> tuple[list[str], list[str]]: + self, push_item: Any, source_details: Optional[ImageDetails], dest_details: ImageDetails + ) -> Tuple[List[str], List[str]]: """ Calculate which archs would be removed if the specified images were source images. @@ -331,7 +331,7 @@ def tag_remove_calculate_archs_source_image( Args: push_item (ContainerPushItem): Push item to perform the workflow with. - source_details (ImageDetails|None): + source_details (Optional[ImageDetails]): ImageDetails of source image, or None if it wasn't specified. dest_details (ImageDetails): ImageDetails of destination image. @@ -361,8 +361,8 @@ def tag_remove_calculate_archs_source_image( return ([], ["amd64"]) def tag_remove_calculate_archs_multiarch_image( - self, push_item: Any, source_details: ImageDetails | None, dest_details: ImageDetails - ) -> tuple[list[str], list[str]]: + self, push_item: Any, source_details: Optional[ImageDetails], dest_details: ImageDetails + ) -> Tuple[List[str], List[str]]: """ Calculate which archs would be removed if the specified images were multiarch images. @@ -371,7 +371,7 @@ def tag_remove_calculate_archs_multiarch_image( Args: push_item (ContainerPushItem): Push item to perform the workflow with. - source_details (ImageDetails|None): + source_details (Optional[ImageDetails]): ImageDetails of source image, or None if it wasn't specified. dest_details (ImageDetails): ImageDetails of destination image. @@ -419,7 +419,7 @@ def tag_remove_calculate_archs_multiarch_image( def tag_add_calculate_archs( self, push_item: Any, tag: str, executor: Executor - ) -> list[str] | None: + ) -> Optional[list[str]]: """ Calculate which architectures are present in a given tag, and which ones would be added. @@ -430,7 +430,7 @@ def tag_add_calculate_archs( Tag, for which an 'add' operation will be performed. executor (Executor): Instance of Executor subclass used for skopeo inspect. - Returns ([str]|None): + Returns (Optional[List[str]]): In case of multiarch image, arches which would be copied to the destination. In case of a source image, None if the copy operation is relevant or [] otherwise. """ diff --git a/pubtools/_quay/tag_images.py b/pubtools/_quay/tag_images.py index 61b50790..40f7a0e0 100644 --- a/pubtools/_quay/tag_images.py +++ b/pubtools/_quay/tag_images.py @@ -1,7 +1,7 @@ import functools import logging import argparse -from typing import Any +from typing import Any, Optional, List, Dict from pubtools.pluggy import pm, task_context @@ -141,7 +141,7 @@ } -def construct_kwargs(args: argparse.Namespace) -> dict[Any, Any]: +def construct_kwargs(args: argparse.Namespace) -> Dict[Any, Any]: """ Construct a kwargs dictionary based on the entered command line arguments. @@ -168,28 +168,28 @@ def construct_kwargs(args: argparse.Namespace) -> dict[Any, Any]: def tag_images( source_ref: str, - dest_refs: list[str], + dest_refs: List[str], all_arch: bool = False, - quay_user: str | None = None, - quay_password: str | None = None, - source_quay_host: str | None = None, - source_quay_user: str | None = None, - source_quay_password: str | None = None, + quay_user: Optional[str] = None, + quay_password: Optional[str] = None, + source_quay_host: Optional[str] = None, + source_quay_user: Optional[str] = None, + source_quay_password: Optional[str] = None, remote_exec: bool = False, - ssh_remote_host: str | None = None, - ssh_remote_host_port: str | None = None, + ssh_remote_host: Optional[str] = None, + ssh_remote_host_port: Optional[str] = None, ssh_reject_unknown_host: bool = False, - ssh_username: str | None = None, - ssh_password: str | None = None, - ssh_key_filename: str | None = None, + ssh_username: Optional[str] = None, + ssh_password: Optional[str] = None, + ssh_key_filename: Optional[str] = None, container_exec: bool = False, - container_image: str | None = None, + container_image: Optional[str] = None, docker_url: str = "unix://var/run/docker.sock", - docker_timeout: int | None = None, + docker_timeout: Optional[int] = None, docker_verify_tls: bool = False, - docker_cert_path: str | None = None, - registry_username: str | None = None, - registry_password: str | None = None, + docker_cert_path: Optional[str] = None, + registry_username: Optional[str] = None, + registry_password: Optional[str] = None, ) -> None: """ Tag images in Quay. @@ -298,14 +298,14 @@ def tag_images( def verify_tag_images_args( - quay_user: str | None, - quay_password: str | None, - source_quay_user: str | None, - source_quay_password: str | None, + quay_user: Optional[str], + quay_password: Optional[str], + source_quay_user: Optional[str], + source_quay_password: Optional[str], remote_exec: bool, - ssh_remote_host: str | None, + ssh_remote_host: Optional[str], container_exec: bool, - container_image: str | None, + container_image: Optional[str], ) -> None: """Verify the presence of input parameters.""" if remote_exec: @@ -332,7 +332,7 @@ def setup_args() -> argparse.ArgumentParser: return setup_arg_parser(TAG_IMAGES_ARGS) -def tag_images_main(sysargs: list[str] | None = None) -> None: +def tag_images_main(sysargs: Optional[List[str]] = None) -> None: """Entrypoint for image tagging.""" logging.basicConfig(level=logging.INFO) diff --git a/pubtools/_quay/types.py b/pubtools/_quay/types.py index 33e301ae..2414178d 100644 --- a/pubtools/_quay/types.py +++ b/pubtools/_quay/types.py @@ -1,4 +1,5 @@ -from typing import List, TypedDict +from typing_extensions import TypedDict +from typing import List Platform = TypedDict( diff --git a/pubtools/_quay/untag_images.py b/pubtools/_quay/untag_images.py index 35f61ea0..91cc7b58 100644 --- a/pubtools/_quay/untag_images.py +++ b/pubtools/_quay/untag_images.py @@ -1,6 +1,6 @@ import logging import argparse -from typing import Any +from typing import Any, Optional, List from pubtools.pluggy import pm, task_context @@ -68,7 +68,7 @@ def construct_kwargs(args: argparse.Namespace) -> dict[Any, Any]: def verify_untag_images_args( - references: list[str], quay_user: str | None, quay_password: str | None + references: List[str], quay_user: Optional[str], quay_password: Optional[str] ) -> None: """ Verify the presence and correctness of input parameters. @@ -90,11 +90,11 @@ def verify_untag_images_args( def untag_images( - references: list[str], + references: List[str], quay_api_token: str, remove_last: bool = False, - quay_user: str | None = None, - quay_password: str | None = None, + quay_user: Optional[str] = None, + quay_password: Optional[str] = None, ) -> None: """ Untag images from Quay. @@ -126,7 +126,7 @@ def setup_args() -> argparse.ArgumentParser: return setup_arg_parser(UNTAG_IMAGES_ARGS) -def untag_images_main(sysargs: list[str] | None = None) -> None: +def untag_images_main(sysargs: Optional[List[str]] = None) -> None: """Entrypoint for untagging images.""" logging.basicConfig(level=logging.INFO) diff --git a/pubtools/_quay/utils/misc.py b/pubtools/_quay/utils/misc.py index cd092585..e412fe45 100644 --- a/pubtools/_quay/utils/misc.py +++ b/pubtools/_quay/utils/misc.py @@ -11,7 +11,7 @@ import sys import textwrap import time -from typing import Any, Callable, Dict, Generator, cast, Optional +from typing import Any, Callable, Dict, Generator, cast, Optional, List, Union, Tuple from concurrent import futures @@ -42,7 +42,7 @@ class FData: kwargs: Dict[str, Any] = field(default_factory=dict) -def run_in_parallel(func: Callable[..., Any], data: list[Any], threads: int = 10) -> dict[Any, Any]: +def run_in_parallel(func: Callable[..., Any], data: List[Any], threads: int = 10) -> Dict[Any, Any]: """Run method on data in parallel. Args: @@ -65,7 +65,7 @@ def run_in_parallel(func: Callable[..., Any], data: list[Any], threads: int = 10 return dict(sorted(results.items(), key=lambda kv: kv[0])) -def setup_arg_parser(args: dict[Any, Any]) -> argparse.ArgumentParser: +def setup_arg_parser(args: Dict[Any, Any]) -> argparse.ArgumentParser: """ Set up ArgumentParser with the provided arguments. @@ -76,7 +76,7 @@ def setup_arg_parser(args: dict[Any, Any]) -> argparse.ArgumentParser: (ArgumentParser) Configured instance of ArgumentParser. """ parser = argparse.ArgumentParser() - arg_groups: dict[Any, Any] = {} + arg_groups: Dict[Any, Any] = {} for aliases, arg_data in args.items(): holder = parser if "group" in arg_data: @@ -102,7 +102,7 @@ def setup_arg_parser(args: dict[Any, Any]) -> argparse.ArgumentParser: def add_args_env_variables( - parsed_args: argparse.Namespace, args: dict[Any, Any] + parsed_args: argparse.Namespace, args: Dict[Any, Any] ) -> argparse.Namespace: """ Add argument values from environment variables. @@ -137,10 +137,10 @@ def capture_stdout() -> Generator[StringIO, None, None]: @contextlib.contextmanager def setup_entry_point_cli( - entry_tuple: tuple[str, str, str], + entry_tuple: Tuple[str, str, str], name: Optional[str], - args: list[str], - environ_vars: dict[str, Any], + args: List[str], + environ_vars: Dict[str, Any], ) -> Generator[Callable[[], Any], None, None]: """ Set up an entrypoint as a context manager. @@ -186,10 +186,10 @@ def setup_entry_point_cli( def run_entrypoint( - entry_tuple: tuple[str, str, str], + entry_tuple: Tuple[str, str, str], name: Optional[str], - args: list[str], - environ_vars: dict[str, Any], + args: List[str], + environ_vars: Dict[str, Any], capture_out: bool = True, ) -> Any: """ @@ -295,7 +295,7 @@ def get_external_container_repo_name(internal_name: str) -> str: return internal_name.replace(INTERNAL_DELIMITER, "/") -def task_status(event: str) -> dict[str, dict[str, str]]: +def task_status(event: str) -> Dict[str, Dict[str, str]]: """Helper function. Expand as necessary.""" # noqa: D401 return dict(event={"type": event}) @@ -330,7 +330,7 @@ def fn_wrapper(*args: Any, **kwargs: Any) -> Any: return decorate -def get_pyxis_ssl_paths(target_settings: dict[str, Any]) -> tuple[str, str]: +def get_pyxis_ssl_paths(target_settings: Dict[str, Any]) -> Tuple[str, str]: """ Get certificate and key paths for Pyxis SSL authentication. @@ -433,7 +433,7 @@ def wrapper_func(*args: Any, **kwargs: Any) -> Any: return inner_retry -def parse_index_image(build_details: Any) -> tuple[str, str, str]: +def parse_index_image(build_details: Any) -> Tuple[str, str, str]: """ Get registry, namespace and repository of a resolved internal index image. @@ -449,7 +449,7 @@ def parse_index_image(build_details: Any) -> tuple[str, str, str]: return (registry, namespace, repo) -def get_basic_auth(host: str) -> tuple[str | None, str | None] | list[str]: +def get_basic_auth(host: str) -> Union[Tuple[Optional[str], Optional[str]], List[str]]: """ Look for container config file for username and password. @@ -470,7 +470,7 @@ def get_basic_auth(host: str) -> tuple[str | None, str | None] | list[str]: return None, None -def pyxis_get_repo_metadata(repo: str, target_settings: dict[str, Any]) -> Any: +def pyxis_get_repo_metadata(repo: str, target_settings: Dict[str, Any]) -> Any: """ Invoke the 'get-repo-metadata' entrypoint from pubtools-pyxis. @@ -490,7 +490,7 @@ def pyxis_get_repo_metadata(repo: str, target_settings: dict[str, Any]) -> Any: args += ["--pyxis-ssl-keyfile", key] args += ["--repo-name", repo] - env_vars: dict[Any, Any] = {} + env_vars: Dict[Any, Any] = {} metadata = run_entrypoint( ("pubtools-pyxis", "console_scripts", "pubtools-pyxis-get-repo-metadata"), "pubtools-pyxis-get-repo-metadata", @@ -500,7 +500,7 @@ def pyxis_get_repo_metadata(repo: str, target_settings: dict[str, Any]) -> Any: return metadata -def set_aws_kms_environment_variables(target_settings: dict[str, Any], profile_name: str) -> None: +def set_aws_kms_environment_variables(target_settings: Dict[str, Any], profile_name: str) -> None: """ Set environment variables required to use an AWS KMS key.