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,264 @@ 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: Optional[Dict[str, BillingCodeRule]] = field(
+ validator=optional(
+ deep_mapping(
+ key_validator=instance_of(str),
+ value_validator=instance_of(BillingCodeRule),
+ mapping_validator=instance_of(dict),
+ )
+ )
+ )
+ """The Billing Code Configuration for the community workflow."""
+
+ 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 :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."""
+ 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."
+}