Drop support for APIv1

docs/model/models.rst

@@ -24,12 +24,6 @@ APIv2
    :show-inheritance:
 
 
-APIv1 (Deprecated)
-------------------
-
-.. autoclass:: starmap_client.models.QueryResponse()
-   :members:
-
 Common
 ------
 

docs/provider/provider.rst

@@ -20,10 +20,3 @@ APIv2
 .. autoclass:: starmap_client.providers.InMemoryMapProviderV2
    :members:
    :special-members: __init__
-
-APIv1 (deprecated)
-~~~~~~~~~~~~~~~~~~
-
-.. autoclass:: starmap_client.providers.InMemoryMapProviderV1
-   :members:
-   :special-members: __init__

starmap_client/client.py

@@ -1,13 +1,12 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
 import logging
-from typing import Any, Dict, Iterator, List, Optional, Type, Union
+from typing import Any, Dict, Iterator, List, Optional
 
 from starmap_client.models import (
     Destination,
     Mapping,
     PaginatedRawData,
     Policy,
-    QueryResponse,
     QueryResponseContainer,
 )
 from starmap_client.providers import StarmapProvider
@@ -16,15 +15,6 @@
 log = logging.getLogger(__name__)
 
 
-Q = Union[QueryResponse, QueryResponseContainer]
-
-API_QUERY_RESPONSE: Dict[str, Type[Q]] = {
-    "v1": QueryResponse,
-    "v2": QueryResponseContainer,
-    "default": QueryResponseContainer,
-}
-
-
 class StarmapClient(object):
     """Implement the StArMap client."""
 
@@ -71,12 +61,12 @@ def __init__(
         self._provider = provider
         self._policies: List[Policy] = []
 
-    def _query(self, params: Dict[str, Any]) -> Optional[Q]:
+    def _query(self, params: Dict[str, Any]) -> Optional[QueryResponseContainer]:
         qr = None
         if self._provider:
             qr = self._provider.query(params)
         rsp = qr or self.session.get("/query", params=params)
-        if isinstance(rsp, QueryResponse) or isinstance(rsp, QueryResponseContainer):
+        if isinstance(rsp, QueryResponseContainer):
             log.debug(
                 "Returning response from the local provider %s", self._provider.__class__.__name__
             )
@@ -85,10 +75,9 @@ def _query(self, params: Dict[str, Any]) -> Optional[Q]:
             log.error(f"Marketplace mappings not defined for {params}")
             return None
         rsp.raise_for_status()
-        converter = API_QUERY_RESPONSE.get(self.api_version, API_QUERY_RESPONSE["default"])
-        return converter.from_json(json=rsp.json())
+        return QueryResponseContainer.from_json(json=rsp.json())
 
-    def query_image(self, nvr: str, **kwargs) -> Optional[Q]:
+    def query_image(self, nvr: str, **kwargs) -> Optional[QueryResponseContainer]:
         """
         Query StArMap using an image NVR.
 
@@ -97,7 +86,7 @@ def query_image(self, nvr: str, **kwargs) -> Optional[Q]:
             workflow(Workflow, optional): The desired workflow to retrieve the mappings (APIv1 Only)
 
         Returns:
-            Q: The query result when found or None.
+            QueryResponseContainer: The query result when found or None.
         """
         return self._query(params={"image": nvr, **kwargs})
 
@@ -106,17 +95,16 @@ def query_image_by_name(
         name: str,
         version: Optional[str] = None,
         **kwargs,
-    ) -> Optional[Q]:
+    ) -> Optional[QueryResponseContainer]:
         """
         Query StArMap using an image NVR.
 
         Args:
             name (str): The image name from NVR.
             version (str, optional): The version from NVR.
-            workflow(Workflow, optional): The desired workflow to retrieve the mappings (APIv1 Only)
 
         Returns:
-            Q: The query result when found or None.
+            QueryResponseContainer: The query result when found or None.
         """
         params = {"name": name, **kwargs}
         if version:

starmap_client/models.py

@@ -1,15 +1,8 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
-import sys
-from copy import deepcopy
 from enum import Enum
-from typing import Any, Dict, List, Optional, Type
+from typing import Any, Dict, List, Optional, Type, TypedDict
 
-if sys.version_info >= (3, 8):
-    from typing import TypedDict  # pragma: no cover
-else:
-    from typing_extensions import TypedDict  # pragma: no cover
-
-from attrs import Attribute, Factory, asdict, evolve, field, frozen
+from attrs import Attribute, field, frozen
 from attrs.validators import deep_iterable, deep_mapping, instance_of, min_len, optional
 
 from starmap_client.utils import assert_is_dict, dict_merge
@@ -20,7 +13,6 @@
     'Destination',
     'Mapping',
     'Policy',
-    'QueryResponse',
     'QueryResponseEntity',
     'QueryResponseContainer',
     'PaginatedRawData',
@@ -142,7 +134,7 @@ class StarmapBaseData(MetaMixin, StarmapJSONDecodeMixin):
     id: Optional[str] = field(validator=optional(instance_of(str)))
     """
     The unique ID for a StArMap model.
-    This field is never set on :class:`~starmap_client.models.QueryResponse`.
+    This field is never set on :class:`~starmap_client.models.QueryResponseEntity`.
     """
 
 
@@ -239,51 +231,6 @@ class Policy(StarmapBaseData):
     """The policy workflow name."""
 
 
-# ============================================ APIv1 ===============================================
-
-
-@frozen
-class QueryResponse(StarmapJSONDecodeMixin):
-    """Represent a query response from StArMap."""
-
-    name: str = field(validator=instance_of(str))
-    """The :class:`~Policy` name."""
-
-    workflow: Workflow = field(converter=lambda x: Workflow(x))
-    """The :class:`~Policy` workflow."""
-
-    clouds: Dict[str, List[Destination]] = field(
-        default=Factory(dict),
-        validator=deep_mapping(
-            key_validator=instance_of(str),
-            value_validator=deep_iterable(
-                member_validator=instance_of(Destination), iterable_validator=instance_of(list)
-            ),
-            mapping_validator=instance_of(dict),
-        ),
-    )
-    """Dictionary with the cloud marketplaces aliases and their respective Destinations."""
-
-    @classmethod
-    def _preprocess_json(cls, json: Any) -> Dict[str, Any]:
-        """
-        Convert the JSON format to the expected by QueryResponse.
-
-        Params:
-            json (dict): A JSON containing a StArMap Query response.
-        Returns:
-            dict: The modified JSON.
-        """
-        mappings = json.pop("mappings", {})
-        for c in mappings.keys():
-            if not isinstance(mappings[c], list):
-                raise ValueError(f"Expected mappings to be a list, got \"{type(mappings[c])}\".")
-            dst = [Destination.from_json(d) for d in mappings[c]]
-            mappings[c] = dst
-        json["clouds"] = mappings
-        return json
-
-
 # ============================================ APIv2 ===============================================
 
 
@@ -429,27 +376,6 @@ def get_mapping_for_account(self, account: str) -> MappingResponseObject:
             raise KeyError(f"No mappings found for account name {account}")
         return obj
 
-    def to_classic_query_response(self) -> QueryResponse:
-        """Return the representation of this object as a :class:`~QueryResponse` from APIv1."""
-
-        def add_bc_to_dst_meta(clouds: Dict[str, List[Destination]]):
-            if self.billing_code_config:
-                bc_data = {k: asdict(v) for k, v in self.billing_code_config.items()}
-                for dst_list in clouds.values():
-                    for d in dst_list:
-                        meta = d.meta or {}
-                        meta["billing-code-config"] = bc_data
-                        d = evolve(d, meta=meta)
-
-        clouds: Dict[str, List[Destination]] = {}
-        for k, v in self.mappings.items():
-            clouds[k] = [deepcopy(d) for d in v.destinations]
-
-        if self.billing_code_config:
-            add_bc_to_dst_meta(clouds)
-
-        return QueryResponse(name=self.name, workflow=self.workflow, clouds=clouds)
-
     @staticmethod
     def _unify_meta_with_mappings(json: Dict[str, Any]) -> None:
         """Merge the ``meta`` data from package into the mappings."""

starmap_client/providers/__init__.py

@@ -1,6 +1,3 @@
 # SPDX-License-Identifier: GPL-3.0-or-later
 from starmap_client.providers.base import StarmapProvider  # noqa: F401
-from starmap_client.providers.memory import (  # noqa: F401
-    InMemoryMapProviderV1,
-    InMemoryMapProviderV2,
-)
+from starmap_client.providers.memory import InMemoryMapProviderV2  # noqa: F401

starmap_client/providers/base.py

@@ -1,9 +1,9 @@
 from abc import ABC, abstractmethod
 from typing import Any, Dict, Generic, List, Optional, TypeVar
 
-from starmap_client.models import QueryResponse, QueryResponseContainer, QueryResponseEntity
+from starmap_client.models import QueryResponseContainer, QueryResponseEntity
 
-T = TypeVar("T", QueryResponse, QueryResponseContainer, QueryResponseEntity)
+T = TypeVar("T", QueryResponseContainer, QueryResponseEntity)
 
 
 class StarmapProvider(ABC, Generic[T]):

starmap_client/providers/memory.py

@@ -1,75 +1,10 @@
 from typing import Any, Dict, List, Optional
 
-from starmap_client.models import QueryResponse, QueryResponseContainer, QueryResponseEntity
+from starmap_client.models import QueryResponseContainer, QueryResponseEntity
 from starmap_client.providers.base import StarmapProvider
 from starmap_client.providers.utils import get_image_name
 
 
-class InMemoryMapProviderV1(StarmapProvider):
-    """Provide in memory (RAM) QueryResponse mapping objects for APIv1."""
-
-    api = "v1"
-
-    def __init__(
-        self, map_responses: Optional[List[QueryResponse]] = None, *args, **kwargs
-    ) -> None:
-        """Crete a new InMemoryMapProvider object.
-
-        Args:
-            map_responses (list, optional)
-                List of QueryResponse objects to load into memory. They will be
-                used by query to fetch the correct response based on name
-                and workflow.
-        """
-        self._separator = str(kwargs.pop("separator", "+"))
-        self._content: Dict[str, QueryResponse] = {}
-        super(StarmapProvider, self).__init__()
-        self._boostrap(map_responses)
-
-    def _boostrap(self, map_responses: Optional[List[QueryResponse]]) -> None:
-        """Initialize the internal content dictionary.
-
-        Args:
-            map_responses (list, optional)
-                List of QueryResponse objects to load into memory.
-        """
-        if not map_responses:
-            return None
-
-        # The in memory content is made of a combination of name and workflow
-        for map in map_responses:
-            key = f"{map.name}{self._separator}{map.workflow.value}"
-            self._content[key] = map
-
-    def list_content(self) -> List[QueryResponse]:
-        """Return a list of stored content."""
-        return list(self._content.values())
-
-    def store(self, response: QueryResponse) -> None:
-        """Store/replace a single QueryResponse object.
-
-        Args:
-            response (QueryResponse):
-                The object to store.
-        """
-        key = f"{response.name}{self._separator}{response.workflow.value}"
-        self._content[key] = response
-
-    def query(self, params: Dict[str, Any]) -> Optional[QueryResponse]:
-        """Return the mapping from memory according to the received params.
-
-        Args:
-            params (dict):
-                The request params to retrieve the mapping.
-        Returns:
-            The requested mapping when found.
-        """
-        name = params.get("name") or get_image_name(params.get("image"))
-        workflow = str(params.get("workflow", ""))
-        search_key = f"{name}{self._separator}{workflow}"
-        return self._content.get(search_key)
-
-
 class InMemoryMapProviderV2(StarmapProvider):
     """Provide in memory (RAM) QueryResponseContainer objects for APIv2."""