Merge pull request #1 from yftacherzog/add-protocols

chore(RHTAPWATCH-636): add compose protocols

compose_generator/compose_generator.py

@@ -0,0 +1,33 @@
+from dataclasses import dataclass
+
+from .protocols import (
+    ComposeConfigurations,
+    ComposeConfigurationsGenerator,
+    ComposeFetcher,
+    ComposeReference,
+    ComposeRequester,
+)
+
+
+@dataclass(frozen=True)
+class ComposeGenerator:
+    """
+    Given implementations to all building blocks if a compose generator, generate a
+    compose and return its reference.
+
+    :param configurations_generator: an object to generate the configurations used for
+        requesting a new compose
+    :param requestor: an object to request a new composed
+    :param fetcher: an object to fetch a compose once it's ready
+    """
+    configurations_generator: ComposeConfigurationsGenerator
+    requestor: ComposeRequester
+    fetcher: ComposeFetcher
+
+    def __call__(self) -> ComposeReference:
+        configs: ComposeConfigurations = self.configurations_generator()
+        request_reference: ComposeReference = self.requestor(configs=configs)
+        result_reference: ComposeReference = self.fetcher(
+            request_reference=request_reference
+        )
+        return result_reference

compose_generator/odcs_compose_generator.py

@@ -0,0 +1,30 @@
+#!/usr/bin/env python3
+
+from .odcs_fetcher import ODCSFetcher
+from .odcs_requester import ODCSRequester
+from .odcs_configurations_generator import ODCSConfigurationsGenerator
+from .compose_generator import ComposeGenerator
+
+
+def main():
+    """
+    Get inputs from container and content_sets yamls and relay them to an ODCS
+    compose generator that will request a compose and store it in a TBD location.
+    """
+    # Get inputs from container and content_sets yamls
+    container_data = {}
+    content_sets_data = {}
+
+    compose_generator = ComposeGenerator(
+        configurations_generator=ODCSConfigurationsGenerator(
+            container_data=container_data,
+            content_sets_data=content_sets_data,
+        ),
+        requestor=ODCSRequester(),
+        fetcher=ODCSFetcher(),
+    )
+    compose_generator()
+
+
+if __name__ == "__main__":
+    main()

compose_generator/odcs_configurations_generator.py

@@ -0,0 +1,25 @@
+from dataclasses import dataclass
+
+from .protocols import ComposeConfigurations, ComposeConfigurationsGenerator
+
+
+@dataclass(frozen=True)
+class ODCSConfigurations(ComposeConfigurations):
+    """
+    Configurations to be used for requesting an ODCS compose
+    """
+
+
+@dataclass(frozen=True)
+class ODCSConfigurationsGenerator(ComposeConfigurationsGenerator):
+    """
+    Generate odcs configurations based on container and content_sets yamls.
+
+    :param container_data: data loaded from container.yaml
+    :param content_sets_data: data loaded from content_sets.yaml
+    """
+    container_data: dict
+    content_sets_data: dict
+
+    def __call__(self) -> ODCSConfigurations:
+        raise NotImplementedError()

compose_generator/odcs_fetcher.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+from pathlib import Path
+
+from .protocols import ComposeFetcher, ComposeReference
+
+
+@dataclass(frozen=True)
+class ODCSResultReference(ComposeReference):
+    """
+    Reference to a locally-stored compose result
+    """
+    compose_path: Path
+
+
+@dataclass(frozen=True)
+class ODCSFetcher(ComposeFetcher):
+    """
+    Fetch ODCS compose based on a remote compose-reference and store it locally
+    """
+    def __call__(self, request_reference: ComposeReference) -> ODCSResultReference:
+        raise NotImplementedError()

compose_generator/odcs_requester.py

@@ -0,0 +1,21 @@
+from dataclasses import dataclass
+
+from .protocols import ComposeConfigurations, ComposeReference, ComposeRequester
+
+
+@dataclass(frozen=True)
+class ODCSRequestReference(ComposeReference):
+    """
+    Reference to a remotely-stored compose data
+    """
+    compose_url: str
+
+
+@dataclass(frozen=True)
+class ODCSRequester(ComposeRequester):
+    """
+    Request a new ODCS compose based on compose configurations and return a reference
+    to the remote compose location.
+    """
+    def __call__(self, configs: ComposeConfigurations) -> ODCSRequestReference:
+        raise NotImplementedError()

compose_generator/protocols.py

@@ -0,0 +1,38 @@
+from typing import Protocol
+
+
+class ComposeConfigurations(Protocol):
+    """
+    Data class for storing compose configurations
+    """
+
+
+class ComposeConfigurationsGenerator(Protocol):
+    """
+    Generate compose configurations
+    """
+    def __call__(self) -> ComposeConfigurations:
+        pass
+
+
+class ComposeReference(Protocol):
+    """
+    Data class for storing compose reference
+    """
+
+
+class ComposeRequester(Protocol):
+    """
+    Given compose configurations, return a remote compose reference
+    """
+    def __call__(self, configs: ComposeConfigurations) -> ComposeReference:
+        pass
+
+
+class ComposeFetcher(Protocol):
+    """
+    Given remote compose reference, return a local compose reference
+    """
+
+    def __call__(self, request_reference: ComposeReference) -> ComposeReference:
+        pass