Add type information to pushsource

setup.py

@@ -24,6 +24,7 @@ def get_requirements():
     version="2.13.3",
     packages=find_packages("src"),
     package_dir={"": "src"},
+    package_data={"pushsource": ["py.typed"]},
     include_package_data=True,
     url="https://github.com/release-engineering/pushsource",
     license="GNU General Public License",

src/pushsource/_impl/backend/errata_source/errata_client.pyi

@@ -0,0 +1,29 @@
+import xmlrpc.client
+from collections import Sequence
+from typing import Any, TypeVar
+
+from pushsource._impl.type_aliases import JsonObject
+
+ErrataRaw_co = TypeVar("ErrataRaw_co", bound="ErrataRaw", covariant=True)
+
+class ErrataRaw(object):
+    advisory_cdn_metadata: JsonObject
+    advisory_cdn_file_list: JsonObject
+    advisory_cdn_docker_file_list: JsonObject
+    ftp_paths: JsonObject
+    def __init__(
+        self,
+        advisory_cdn_metadata: JsonObject,
+        advisory_cdn_file_list: JsonObject,
+        advisory_cdn_docker_file_list: JsonObject,
+        ftp_paths: JsonObject,
+    ) -> None: ...
+
+class ErrataClient(object):
+    _errata_service: xmlrpc.client.ServerProxy
+    def __init__(self, threads: int, url: str, **retry_args: Any) -> None: ...
+    def shutdown(self) -> None: ...
+    def _log_queried_et(self, response: JsonObject, advisory_id: str) -> JsonObject: ...
+    def get_raw_f(self, advisory_id: str) -> Sequence[ErrataRaw_co]: ...
+    # TODO: narrow return type if possible: JsonObject maybe?
+    def _call_et(self, method_name: str, advisory_id: str) -> Any: ...

src/pushsource/_impl/backend/errata_source/errata_source.pyi

@@ -0,0 +1,67 @@
+from collections import Mapping, Iterator
+from collections.abc import Sequence
+from types import TracebackType
+from typing import Optional, Type, List, Any
+
+from pushsource import (
+    Source,
+    ErratumPushItem,
+    ContainerImagePushItem,
+)
+from pushsource._impl.backend.errata_source.errata_client import ErrataRaw
+from pushsource._impl.model.base import PushItem_co, PushItem_contra
+from pushsource._impl.type_aliases import MaybeString
+
+# TODO: is the value type a model type or just a Mapping?
+DockerFileList = Mapping[str, Any]
+
+# TODO: is the value type a model type or just a Mapping?
+RpmList = Mapping[str, Mapping[Any, ...]]
+
+class ErrataSource(Source):
+    _errata_service_url: str
+    _advisory_ids: List[str]
+    def __init__(
+        self,
+        url: str,
+        errata: MaybeString,
+        koji_source: Optional[str] = ...,
+        rpm_filter_arch: Optional[MaybeString] = ...,
+        legacy_container_repos: bool = ...,
+        threads: int = ...,
+        timeout: int = ...,
+    ) -> None: ...
+    def __enter__(self) -> "ErrataSource": ...
+    def __exit__(
+        self,
+        exc_type: Type[BaseException],
+        exc_val: BaseException,
+        exc_tb: TracebackType,
+    ) -> None: ...
+    def __iter__(self) -> Iterator[PushItem_co]: ...
+    def _koji_source(**kwargs: Any) -> Source: ...
+    def _push_items_from_raw(self, raw: ErrataRaw) -> Sequence[PushItem_co]: ...
+    def _push_items_from_container_manifests(
+        self, erratum: ErratumPushItem, docker_file_list: DockerFileList
+    ) -> Sequence[PushItem_co]: ...
+    def _enrich_container_push_item(
+        self,
+        erratum: ErratumPushItem,
+        docker_file_list: DockerFileList,
+        item: ContainerImagePushItem,
+    ) -> ContainerImagePushItem: ...
+    def _push_items_from_rpms(
+        self, erratum: ErratumPushItem, rpm_list: RpmList
+    ) -> Sequence[PushItem_co]: ...
+    def _module_push_items_from_build(
+        self, erratum: ErratumPushItem, build_nvr: str, build_info: Mapping[Any, ...]
+    ) -> Sequence[PushItem_co]: ...
+    def _filter_rpms_by_arch(
+        self, erratum: ErratumPushItem, rpm_filenames: Sequence[str]
+    ) -> Sequence[str]: ...
+    def _rpm_push_items_from_build(
+        self, erratum: ErratumPushItem, build_nvr: str, build_info: Mapping[Any, ...]
+    ) -> Sequence[PushItem_co]: ...
+    def _add_ftp_paths(
+        self, items: Sequence[PushItem_contra], erratum: ErratumPushItem, raw: ErrataRaw
+    ) -> Sequence[PushItem_co]: ...

src/pushsource/_impl/backend/koji_containers.pyi

@@ -0,0 +1,24 @@
+from collections import Mapping, Sequence
+from typing import Final, Any
+
+from pushsource import ContainerImagePullInfo
+from pushsource._impl.model.container import ContainerImagePullSpec_co
+
+MIME_TYPE_MANIFEST_LIST: Final[str] = ...
+
+# TODO: a lot of the properties of this class are ambiguous
+#  Mapping[Any] (i.e. Mapping[Any, Any]): can this be narrowed?
+#  Perhaps the JsonObject type would be more helpful.
+class ContainerArchiveHelper(object):
+    build_image: Mapping[Any]
+    build_index: Mapping[Any]
+    archive_extra: Mapping[Any]
+    archive_docker: Mapping[Any]
+    source_tags: Sequence[str]
+    arch: str
+    labels: Mapping[Any]
+    pull_info: ContainerImagePullInfo
+
+def get_tag_specs(raw_specs: Sequence[str]) -> Sequence[ContainerImagePullSpec_co]: ...
+def get_digest_specs(raw_specs: Sequence[str], digests_map):
+    Sequence[ContainerImagePullSpec_co]: ...

src/pushsource/_impl/backend/koji_source.pyi

@@ -0,0 +1,101 @@
+from collections import Mapping, Sequence, Iterator, Iterable, MutableSequence
+from concurrent.futures import Executor, Future
+from queue import Queue
+from threading import RLock as ReentrantLock
+from types import TracebackType
+from typing import Any, Optional, Union, Type, ClassVar, Final, NoReturn
+
+from koji import ClientSession, VirtualCall
+
+from pushsource import Source, RpmPushItem, OperatorManifestPushItem
+from pushsource._impl.model.base import PushItem_co
+from pushsource._impl.type_aliases import MaybeString, JsonObject
+
+Id = Union[str, int]
+
+CACHE_LOCK: ReentrantLock
+
+RETRY_ARGS: Mapping[str, Any]
+
+class ListArchivesCommand(object):
+    def __init__(self, build: Mapping[str, Any]) -> None: ...
+    build: Mapping[str, Any]
+    # TODO: is VirtualCall correct for this type?
+    call: Optional[VirtualCall] = ...
+    def execute(self, source: "KojiSource", session: ClientSession) -> int: ...
+    def save(self, source: "KojiSource", koji_queue: Queue) -> None: ...
+
+class GetBuildCommand(object):
+    def __init__(self, ident: int, list_archives: bool = ...) -> None: ...
+    indent: int
+    list_archives: bool = ...
+    # TODO: is VirtualCall correct for this type?
+    call: Optional[VirtualCall] = ...
+    def execute(self, source: "KojiSource", session: ClientSession) -> int: ...
+    def save(self, source: "KojiSource", koji_queue: Queue) -> None: ...
+
+class GetRpmCommand(object):
+    def __init__(self, indent: int) -> None: ...
+    indent: int
+    # TODO: is VirtualCall correct for this type?
+    call: Optional[VirtualCall] = ...
+    def execute(self, source: "KojiSource", session: ClientSession) -> int: ...
+    def save(self, source: "KojiSource", koji_queue: Queue) -> None: ...
+
+class KojiSource(Source):
+    _BATCH_SIZE: Final[ClassVar[int]]
+    _koji_session: ClientSession
+    def __init__(
+        self,
+        url: str,
+        dest: MaybeString,
+        rpm: Optional[Sequence[Id]] = ...,
+        module_build: Optional[Sequence[Id]] = ...,
+        module_filter_filename: Optional[Sequence[str]] = ...,
+        container_build: Optional[Sequence[Id]] = ...,
+        signing_key: Optional[Sequence[str]] = ...,
+        basedir: Optional[str] = ...,
+        threads: int = ...,
+        timeout: int = ...,
+        cache: Optional[MutableMapping[str, Any]] = ...,
+        executor: Optional[Executor] = ...,
+    ) -> None: ...
+    def __enter__(self) -> "KojiSource": ...
+    def __exit__(
+        self,
+        exc_type: Type[BaseException],
+        exc_val: BaseException,
+        exc_tb: TracebackType,
+    ) -> None: ...
+    def __iter__(self) -> Iterator[PushItem_co]: ...
+    # TODO: the return type here feels hacky, but is accurate
+    def _koji_check(self) -> None: ...
+    def _koji_get_version(self) -> str: ...
+    # TODO: reasonably sure based on the Fedora Koji XML-RPC
+    #  docs for getRPM, getBuild, and getArchive that what is
+    #  the map returned by those API calls is what's stored in
+    #  the cache. Since it is not unlike a JSON object I am sticking
+    #  with a JsonObject return type for now
+    def _get_rpm(self, rpm: str) -> JsonObject: ...
+    def _get_build(self, build_id: Id) -> JsonObject: ...
+    def _get_archives(self, build_id: Id) -> JsonObject: ...
+    # TODO: maybe JsonObject is appropriate for the meta parameter type
+    def _push_items_from_rpm_meta(
+        self, rpm: str, meta: Mapping[str, Any]
+    ) -> Sequence[RpmPushItem]: ...
+    def _module_filtered(self, file_path: str) -> bool: ...
+    def _get_module_name(self, nvr: str, file_path: str) -> str: ...
+    # TODO: maybe JsonObject is appropriate for the meta parameter type
+    def _push_items_from_module_build(
+        self, nvr: str, meta: Mapping[str, Any]
+    ) -> Sequence[PushItem_co]: ...
+    def _push_items_from_container_build(
+        self, nvr: str, meta: Mapping[str, Any]
+    ) -> Sequence[PushItem_co]: ...
+    def _get_operator_item(
+        self, nvr: str, meta: Mapping[str, Any], archives: Iterable[Mapping[str, Any]]
+    ) -> OperatorManifestPushItem: ...
+    def _rpm_futures(self) -> Sequence[Future[PushItem_co]]: ...
+    def _modulemd_futures(self) -> Sequence[Future[PushItem_co]]: ...
+    def _container_futures(self) -> Sequence[Future[PushItem_co]]: ...
+    def _do_fetch(self, koji_queue: Queue, exceptions: MutableSequence) -> None: ...

src/pushsource/_impl/backend/modulemd.pyi

@@ -0,0 +1,9 @@
+class Module(object):
+    name: str
+    stream: str
+    version: str
+    context: str
+    arch: str
+    nsvca: str
+    @classmethod
+    def from_file(cls, fname: str) -> "Module": ...

src/pushsource/_impl/backend/registry_source.pyi

@@ -0,0 +1,27 @@
+from collections import Iterator
+from re import Pattern
+from types import TracebackType
+from typing import Optional, Type
+
+from pushsource import Source, PushItem
+from pushsource._impl.model.base import PushItem_co
+from pushsource._impl.type_aliases import MaybeString
+
+IMAGE_URI_REGEX: Pattern
+
+class RegistrySource(Source):
+    def __init__(
+        self,
+        image: MaybeString,
+        dest: Optional[MaybeString] = ...,
+        dest_signing_key: Optional[MaybeString] = ...,
+    ) -> None: ...
+    def __enter__(self) -> "RegistrySource": ...
+    def __exit__(
+        self,
+        exc_type: Type[BaseException],
+        exc_val: BaseException,
+        exc_tb: TracebackType,
+    ) -> None: ...
+    def __iter__(self) -> Iterator[PushItem_co]: ...
+    def _push_item_from_registry_uri(self, uri: str, signing_key: str) -> PushItem: ...

src/pushsource/_impl/backend/staged/staged_ami.pyi

@@ -0,0 +1,10 @@
+from os import DirEntry
+
+from pushsource import AmiPushItem
+from pushsource._impl.backend.staged.staged_base import StagedBaseMixin
+from pushsource._impl.backend.staged.staged_utils import StagingLeafDir, StagingMetadata
+
+class StagedAmiMixin(StagedBaseMixin):
+    def __push_item(
+        self, leafdir: StagingLeafDir, metadata: StagingMetadata, entry: DirEntry
+    ) -> AmiPushItem: ...

src/pushsource/_impl/backend/staged/staged_base.pyi

@@ -0,0 +1,39 @@
+from collections import Callable, Sequence
+from os import DirEntry
+from typing import Final, ClassVar, MutableMapping, Any
+
+from pushsource import PushItem
+from pushsource._impl.backend.staged.staged_utils import StagingLeafDir, StagingMetadata
+from pushsource._impl.model.base import PushItem_co
+
+# Typically "self" isn't annotated, but in this case
+# "unbound" methods are being stashed in a dict and
+# the "self" parameter is not fulfilled until later via
+# partial application. This type annotation accounts for
+# the type of unbound "self" parameter in the callable signature.
+UnboundTypeHandlerDelegate = Callable[
+    ["StagedBaseMixin", StagingLeafDir, StagingMetadata, DirEntry], PushItem
+]
+# A type for an UnboundTypeHandlerDelegate where "self" has
+# been bound to an instance of StagedBaseMixin
+BoundTypeHandlerDelegate = Callable[
+    [StagingLeafDir, StagingMetadata, DirEntry], PushItem
+]
+# A partially applied type handler callable
+PartialTypeHandler = Callable[[StagingLeafDir, StagingMetadata], Sequence[PushItem_co]]
+
+class TypeHandler(object):
+    HANDLERS: Final[ClassVar[MutableMapping[str, UnboundTypeHandlerDelegate]]]
+    type_name: str
+    def __init__(self, type_name: str) -> None: ...
+    def __call__(self, fn: UnboundTypeHandlerDelegate) -> None: ...
+
+class StagedBaseMixin(object):
+    _FILE_TYPES = Final[ClassVar[MutableMapping[str, PartialTypeHandler]]]
+    def __init__(self, *args: Any, **kwargs: Any) -> None: ...
+    def __mixin_push_items(
+        self,
+        leafdir: StagingLeafDir,
+        metadata: StagingMetadata,
+        delegate: BoundTypeHandlerDelegate,
+    ) -> Sequence[PushItem_co]: ...

src/pushsource/_impl/backend/staged/staged_compsxml.pyi

@@ -0,0 +1,10 @@
+from os import DirEntry
+
+from pushsource import CompsXmlPushItem
+from pushsource._impl.backend.staged.staged_base import StagedBaseMixin
+from pushsource._impl.backend.staged.staged_utils import StagingLeafDir, StagingMetadata
+
+class StagedCompsXmlMixin(StagedBaseMixin):
+    def __push_item(
+        self, leafdir: StagingLeafDir, _: StagingMetadata, entry: DirEntry
+    ) -> CompsXmlPushItem: ...

src/pushsource/_impl/backend/staged/staged_errata.pyi

@@ -0,0 +1,10 @@
+from os import DirEntry
+
+from pushsource import ErratumPushItem
+from pushsource._impl.backend.staged.staged_base import StagedBaseMixin
+from pushsource._impl.backend.staged.staged_utils import StagingMetadata, StagingLeafDir
+
+class StagedErrataMixin(StagedBaseMixin):
+    def __make_push_item(
+        self, leafdir: StagingLeafDir, metadata: StagingMetadata, entry: DirEntry
+    ) -> ErratumPushItem: ...

src/pushsource/_impl/backend/staged/staged_files.pyi

@@ -0,0 +1,10 @@
+from os import DirEntry
+
+from pushsource import FilePushItem
+from pushsource._impl.backend.staged.staged_base import StagedBaseMixin
+from pushsource._impl.backend.staged.staged_utils import StagingLeafDir, StagingMetadata
+
+class StagedFilesMixin(StagedBaseMixin):
+    def __file_push_item(
+        self, leafdir: StagingLeafDir, metadata: StagingMetadata, entry: DirEntry
+    ) -> FilePushItem: ...

src/pushsource/_impl/backend/staged/staged_modulemd.pyi

@@ -0,0 +1,10 @@
+from os import DirEntry
+
+from pushsource import ModuleMdPushItem
+from pushsource._impl.backend.staged.staged_base import StagedBaseMixin
+from pushsource._impl.backend.staged.staged_utils import StagingLeafDir, StagingMetadata
+
+class StagedModuleMdMixin(StagedBaseMixin):
+    def __push_item(
+        self, leafdir: StagingLeafDir, metadata: StagingMetadata, entry: DirEntry
+    ) -> ModuleMdPushItem: ...

src/pushsource/_impl/backend/staged/staged_productid.pyi

@@ -0,0 +1,10 @@
+from os import DirEntry
+
+from .staged_utils import StagingLeafDir, StagingMetadata
+from ...model import ProductIdPushItem
+from .staged_base import StagedBaseMixin
+
+class StagedProductIdMixin(StagedBaseMixin):
+    def __push_item(
+        self, leafdir: StagingLeafDir, _: StagingMetadata, entry: DirEntry
+    ) -> ProductIdPushItem: ...

src/pushsource/_impl/backend/staged/staged_rpm.pyi

@@ -0,0 +1,10 @@
+from os import DirEntry
+
+from .staged_base import StagedBaseMixin
+from .staged_utils import StagingLeafDir, StagingMetadata
+from ...model import RpmPushItem
+
+class StagedRpmMixin(StagedBaseMixin):
+    def __push_item(
+        self, leafdir: StagingLeafDir, _: StagingMetadata, entry: DirEntry
+    ) -> RpmPushItem: ...