Add models for StArMap APIv2 Query

This commit introduces the following new models for the StArMap APIv2:

- `QueryResponseContainer`: holds the whole Query APIv2 response within
  its property `responses` which is a list of `QueryResponseEntity`.

- `QueryResponseEntity`: Represents a single mapping for a given
  workflow, name and cloud. It can be converted to the `QueryResponse`
  from APIv1 model which works in an analogue way (without support for
  cloud name as a property).

- `MappingResponseObject`: Represent a single mapping entity which is
  part of `QueryResponseEntity`. It holds a list of `Destination`
  object, similar to what `QueryResponse` does on its `clouds`
  attribute.

- `BillingCodeRule`: A dedicated model to represent a Billing Code Rule
  for the comnmunity workflow.

Refers to SPSTRAT-382

starmap_client/models.py

@@ -1,26 +1,35 @@
 # 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
+from typing import Any, Dict, List, Optional, Type
 
 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, field, frozen
+from attrs import Attribute, Factory, asdict, evolve, field, frozen
 from attrs.validators import deep_iterable, deep_mapping, instance_of, min_len, optional
 
+from starmap_client.utils import assert_is_dict, dict_union
+
 __all__ = [
+    'BillingCodeRule',
     'Destination',
     'Mapping',
     'Policy',
     'QueryResponse',
+    'QueryResponseEntity',
+    'QueryResponseContainer',
     'PaginatedRawData',
     'PaginationMetadata',
 ]
 
 
+# ============================================ Common ==============================================
+
+
 class PaginationMetadata(TypedDict):
     """Datastructure of the metadata about the paginated query."""
 
@@ -222,6 +231,9 @@ class Policy(StarmapBaseData):
     """The policy workflow name."""
 
 
+# ============================================ APIv1 ===============================================
+
+
 @frozen
 class QueryResponse(StarmapJSONDecodeMixin):
     """Represent a query response from StArMap."""
@@ -262,3 +274,260 @@ def _preprocess_json(cls, json: Any) -> Dict[str, Any]:
             mappings[c] = dst
         json["clouds"] = mappings
         return json
+
+
+# ============================================ APIv2 ===============================================
+
+
+class BillingImageType(str, Enum):
+    """Define the image type for :class:`~BillingCodeRule` for APIv2."""
+
+    access = "access"
+    hourly = "hourly"
+    marketplace = "marketplace"
+
+
+@frozen
+class BillingCodeRule(StarmapJSONDecodeMixin):
+    """Define a single Billing Code Configuration rule for APIv2."""
+
+    codes: List[str] = field(
+        validator=deep_iterable(
+            member_validator=instance_of(str), iterable_validator=instance_of(list)
+        )
+    )
+    """The billing codes to insert when this rule is matched."""
+
+    image_name: str = field(validator=instance_of(str))
+    """The image name to match the rule."""
+
+    image_types: List[BillingImageType] = field(
+        converter=lambda x: [BillingImageType[d] for d in x],
+        validator=deep_iterable(
+            member_validator=instance_of(BillingImageType), iterable_validator=instance_of(list)
+        ),
+    )
+    """Image types list. Supported values are ``access`` and ``hourly``."""
+
+    name: Optional[str]
+    """The billing code rule name."""
+
+
+@frozen
+class MappingResponseObject(StarmapJSONDecodeMixin, MetaMixin):
+    """Represent a single mapping response from :class:`~QueryResponseObject` for APIv2."""
+
+    destinations: List[Destination] = field(
+        validator=deep_iterable(
+            iterable_validator=instance_of(list), member_validator=instance_of(Destination)
+        )
+    )
+    """List of destinations for the mapping response object."""
+
+    provider: Optional[str] = field(validator=optional(instance_of(str)))
+    """The provider name for the community workflow."""
+
+    @staticmethod
+    def _unify_meta_with_destinations(json: Dict[str, Any]) -> None:
+        """Merge the ``meta`` data from mappings into the destinations."""
+        destinations = json.get("destinations", [])
+        if not isinstance(destinations, list):
+            raise ValueError(f"Expected destinations to be a list, got \"{type(destinations)}\"")
+        meta = json.get("meta", {})
+        for d in destinations:
+            d["meta"] = dict_union(meta, d.get("meta", {}))
+
+    @classmethod
+    def _preprocess_json(cls, json: Any) -> Dict[str, Any]:
+        """
+        Properly adjust the Destinations list for building this object.
+
+        Params:
+            json (dict): A JSON containing a StArMap Query response.
+        Returns:
+            dict: The modified JSON.
+        """
+        cls._unify_meta_with_destinations(json)
+        provider = json.get("provider", None)
+        destinations = json.get("destinations", [])
+        # _unify_meta_lower_entity(json, "destinations")
+        converted_destinations = []
+        for d in destinations:
+            d["provider"] = provider
+            converted_destinations.append(Destination.from_json(d))
+        json["destinations"] = converted_destinations
+        return json
+
+
+@frozen
+class QueryResponseEntity(StarmapJSONDecodeMixin, MetaMixin):
+    """Represent a single query response entity from StArMap APIv2."""
+
+    name: str = field(validator=instance_of(str))
+    """The :class:`~Policy` name."""
+
+    billing_code_config: Dict[str, BillingCodeRule] = field(
+        validator=deep_mapping(
+            key_validator=instance_of(str),
+            value_validator=instance_of(BillingCodeRule),
+            mapping_validator=instance_of(dict),
+        )
+    )
+
+    cloud: str = field(validator=instance_of(str))
+    """The cloud name where the destinations are meant to."""
+
+    workflow: Workflow = field(converter=lambda x: Workflow(x))
+    """The :class:`~Policy` workflow."""
+
+    mappings: Dict[str, MappingResponseObject] = field(
+        validator=deep_mapping(
+            key_validator=instance_of(str),
+            value_validator=instance_of(MappingResponseObject),
+            mapping_validator=instance_of(dict),
+        ),
+    )
+    """Dictionary with the cloud account names and MappingResponseObjects."""
+
+    @property
+    def account_names(self) -> List[str]:
+        """Return the list of cloud account names declared on ``mappings``."""
+        return list(self.mappings.keys())
+
+    @property
+    def all_mappings(self) -> List[MappingResponseObject]:
+        """Return all ``MappingResponseObject`` stored in ``mappings``."""
+        return list(self.mappings.values())
+
+    def get_mapping_for_account(self, account: str) -> MappingResponseObject:
+        """Return a single ``MappingResponseObject`` for a given account name.
+
+        Args:
+            account (str):
+                The account name to retrieve the ``MappingResponseObject``
+        Returns:
+            MappingResponseObject: The required mapping when found
+        Raises: KeyError when not found
+        """
+        obj = self.mappings.get(account, None)
+        if not obj:
+            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 ``QueryResponse`` from APIv1."""
+
+        def add_bc_to_dst_meta(clouds: Dict[str, List[Destination]]):
+            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."""
+        mappings = json.get("mappings", {})
+        meta = json.get("meta", {})
+        for k, v in mappings.items():
+            mappings[k]["meta"] = dict_union(meta, v.get("meta", {}))
+
+    @classmethod
+    def _preprocess_json(cls, json: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Properly adjust the MappingResponseObject list and BillingCodeConfig dict for building this object.
+
+        Params:
+            json (dict): A JSON containing a StArMap Query response.
+        Returns:
+            dict: The modified JSON.
+        """  # noqa: D202 E501
+
+        def parse_entity_build_obj(
+            entity_name: str, converter_type: Type[StarmapJSONDecodeMixin]
+        ) -> None:
+            entity = json.pop(entity_name, {})
+            for k in entity.keys():
+                assert_is_dict(entity[k])
+                obj = converter_type.from_json(entity[k])
+                entity[k] = obj
+            json[entity_name] = entity
+
+        bcc = json.pop("billing-code-config", {})
+        json["billing_code_config"] = bcc
+        cls._unify_meta_with_mappings(json)
+        parse_entity_build_obj("mappings", MappingResponseObject)
+        parse_entity_build_obj("billing_code_config", BillingCodeRule)
+        return json
+
+
+@frozen
+class QueryResponseContainer:
+    """Represent a full query response from APIv2."""
+
+    responses: List[QueryResponseEntity] = field(
+        validator=deep_iterable(
+            member_validator=instance_of(QueryResponseEntity), iterable_validator=instance_of(list)
+        )
+    )
+    """List with all responses from a Query V2 mapping."""
+
+    @classmethod
+    def from_json(cls, json: Any):
+        """
+        Convert the APIv2 response JSON into this object.
+
+        Args:
+            json (list)
+                A JSON containing a StArMap APIv2 response.
+        Returns:
+            The converted object from JSON.
+        """
+        if not isinstance(json, list):
+            raise ValueError(f"Expected root to be a list, got \"{type(json)}\".")
+
+        responses = [QueryResponseEntity.from_json(qre) for qre in json]
+        return cls(responses)
+
+    def filter_by_workflow(self, workflow: Workflow) -> List[QueryResponseEntity]:
+        """Return a sublist of the responses with only the selected workflow.
+
+        Args:
+            workflow (Workflow):
+                The workflow to filter the list of responses
+        Returns:
+            list: The sublist with only the selected workflows
+        """
+        return [x for x in self.responses if x.workflow == workflow]
+
+    def filter_by_cloud(self, cloud: str) -> List[QueryResponseEntity]:
+        """Return a sublist of the responses with only the selected cloud name.
+
+        Args:
+            cloud (str):
+                The cloud name to filter the list of responses
+        Returns:
+            list: The sublist with only the selected cloud name.
+        """
+        return [x for x in self.responses if x.cloud == cloud]
+
+    def filter_by(self, **kwargs):
+        """Return a sublist of the responses with the selected filters."""
+        filters = {
+            "cloud": self.filter_by_cloud,
+            "workflow": self.filter_by_workflow,
+        }
+        res = []
+        for k, v in kwargs.items():
+            res.extend(filters[k](v))
+        return res

starmap_client/utils.py

@@ -0,0 +1,25 @@
+from typing import Any, Dict
+
+
+def assert_is_dict(data: Any) -> None:
+    """Ensure the incoming data is a dictionary, raises ``ValueError`` if not."""
+    if not isinstance(data, dict):
+        raise ValueError(f"Expected dictionary, got {type(data)}: {data}")
+
+
+def dict_union(a: Dict[str, Any], b: Dict[str, Any]) -> Dict[str, Any]:
+    """Return a new dictionary with the combination of A and B.
+
+    Args:
+        a (dict):
+            The left dictionary to be combined
+        b (dict):
+            The right dictionary to be combined. It will override the same keys from A.
+    Returns:
+        dict: A new dictionary with combination of A and B.
+    """
+    c = {}
+    for x in [a, b]:
+        assert_is_dict(x)
+        c.update(x)
+    return c

tests/data/query_v2/mapping_response_obj/invalid_mro1.json

@@ -0,0 +1,12 @@
+{
+    "destinations": {
+        "architecture": "x86_64",
+        "destination": "fake_destination",
+        "overwrite": false,
+        "restrict_version": true,
+        "meta": {
+            "description": "Description on destination level."
+        }
+    },
+    "error": "Expected destinations to be a list, got \"<class 'dict'>\""
+}

tests/data/query_v2/mapping_response_obj/invalid_mro2.json

@@ -0,0 +1,4 @@
+{
+    "destinations": true,
+    "error": "Expected destinations to be a list, got \"<class 'bool'>\""
+}

tests/data/query_v2/mapping_response_obj/invalid_mro3.json

@@ -0,0 +1,13 @@
+{
+    "destinations": [
+        {
+            "architecture": "x86_64",
+            "destination": "fake_destination",
+            "overwrite": false,
+            "restrict_version": true
+        }
+    ],
+    "meta": [],
+    "provider": "AWS",
+    "error": "Expected dictionary, got <class 'list'>: \\[\\]"
+}

tests/data/query_v2/mapping_response_obj/valid_mro1.json

@@ -0,0 +1,14 @@
+{
+    "destinations": [
+        {
+            "architecture": "x86_64",
+            "destination": "fake_destination",
+            "overwrite": false,
+            "restrict_version": true
+        }
+    ],
+    "meta": {
+        "description": "Description on map level."
+    },
+    "provider": "AWS"
+}

tests/data/query_v2/mapping_response_obj/valid_mro1_meta.json

@@ -0,0 +1,3 @@
+{
+    "description": "Description on map level."
+}