Add an helper for the Pricing API

boto3 has support for the Pricing API that can be used to
dynamically fetch the price information of EC2 instances.
However this is not straightforward to use, hence the
helper class. This is heavily inspired by the answers in
https://stackoverflow.com/questions/51673667/use-boto3-to-get-current-price-for-given-ec2-instance-type/51685222#51685222

src/e3/aws/pricing/__init__.py

@@ -0,0 +1,133 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+import json
+
+from e3.aws.util import get_region_name
+
+if TYPE_CHECKING:
+    from typing import Any
+    import botocore
+
+    _CacheKey = tuple[str | None, str | None, str | None]
+
+    # This is only to avoid repeating the type everywhere
+    PriceInformation = dict[str, Any]
+
+
+class Pricing:
+    """Pricing abstraction."""
+
+    def __init__(self, client: botocore.client.BaseClient) -> None:
+        """Initialize Pricing.
+
+        :param client: a client for the Pricing API
+        """
+        self.client = client
+        # Cache results of client.get_products requests
+        self._cache: dict[_CacheKey, list[PriceInformation]] = {}
+
+    def _cache_key(
+        self,
+        instance_type: str | None = None,
+        os: str | None = None,
+        region: str | None = None,
+    ) -> _CacheKey:
+        """Get the key for cache.
+
+        :param instance_type: EC2 instance type
+        :param os: operating system
+        :param region: region code
+        :return: key for cache
+        """
+        return (instance_type, os, region)
+
+    def ec2_price_information(
+        self,
+        instance_type: str | None = None,
+        os: str | None = None,
+        region: str | None = None,
+        filters: list[dict[str, Any]] | None = None,
+    ) -> list[PriceInformation]:
+        """Get pricing informations for EC2 instances.
+
+        :param instance_type: filter by EC2 instance type
+        :param os: filter by operating system
+        :param region: filter by region code
+        :param filters: additional filters for client.get_products
+        :return: pricing information as returned by client.get_products
+        """
+        # Check if the price information is already cached
+        key = self._cache_key(instance_type=instance_type, os=os, region=region)
+        if key in self._cache:
+            return self._cache[key]
+
+        # Even though the API data contains regionCode field, it will not return
+        # accurate data. However using the location field will, but then we need to
+        # translate the region code into a region name. You could skip this by using
+        # the region names in your code directly, but most other APIs are using the
+        # region code.
+        filters = filters if filters is not None else []
+        for field, value in (
+            ("operatingSystem", os),
+            ("instanceType", instance_type),
+            ("location", None if region is None else get_region_name(region)),
+        ):
+            if value is not None:
+                filters.append(
+                    {
+                        "Field": field,
+                        "Value": str(value),
+                        "Type": "TERM_MATCH",
+                    }
+                )
+
+        result: list[PriceInformation] = []
+        paginator = self.client.get_paginator("get_products")
+        for data in paginator.paginate(ServiceCode="AmazonEC2", Filters=filters):
+            for price in data["PriceList"]:
+                price = json.loads(price)
+
+                # Cache the individual response
+                attributes = price["product"]["attributes"]
+                self._cache[
+                    self._cache_key(
+                        instance_type=attributes["instanceType"],
+                        os=attributes["operatingSystem"],
+                        region=attributes["regionCode"],
+                    )
+                ] = [price]
+
+                result.append(price)
+
+        # Cache the whole response
+        self._cache[key] = result
+        return result
+
+    def ec2_on_demand_price(
+        self, instance_type: str, os: str, region: str
+    ) -> float | None:
+        """Get the on-demand hourly price of an EC2 instance.
+
+        :param instance_type: EC2 instance type
+        :param os: operating system
+        :param region: region code
+        :return: hourly price or None if no price information is found
+        """
+        prices = self.ec2_price_information(
+            instance_type,
+            os,
+            region,
+            # Filters for on-demand information only
+            filters=[
+                {"Type": "TERM_MATCH", "Field": "capacitystatus", "Value": "Used"},
+                {"Type": "TERM_MATCH", "Field": "preInstalledSw", "Value": "NA"},
+                {"Type": "TERM_MATCH", "Field": "tenancy", "Value": "shared"},
+            ],
+        )
+
+        if not prices:
+            return None
+
+        price_data = list(prices[0]["terms"]["OnDemand"].values())[0]
+        price_per_unit_data = list(price_data["priceDimensions"].values())[0]
+        return float(price_per_unit_data["pricePerUnit"]["USD"])

src/e3/aws/util/__init__.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+import json
+from pkg_resources import resource_filename
+
+
+def get_region_name(region_code: str) -> str | None:
+    """Translate region code to region name.
+
+    This makes use of data/endpoints.json from botocore to map
+    from one to the other.
+
+    :param region_code: region code
+    :return: region name or None if the region code is not found
+    """
+    endpoint_file = resource_filename("botocore", "data/endpoints.json")
+    with open(endpoint_file) as f:
+        data = json.load(f)
+
+    return data["partitions"][0]["regions"].get(region_code, {}).get("description")

tests/tests_e3_aws/pricing/main_test.py

@@ -0,0 +1,113 @@
+from __future__ import annotations
+from typing import TYPE_CHECKING
+import pytest
+import boto3
+import json
+from botocore.stub import Stubber
+
+from e3.aws.pricing import Pricing
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+
+
+# EC2 instance price
+INSTANCE_PRICE = 0.177
+
+# EC2 instance price information
+INSTANCE_PRICE_INFORMATION = {
+    "product": {
+        "attributes": {
+            "operatingSystem": "Ubuntu Pro",
+            "regionCode": "us-east-1",
+            "instanceType": "c6i.xlarge",
+        },
+    },
+    "terms": {
+        "OnDemand": {
+            "YD4JEF3ADAGG84PN.JRTCKXETXF": {
+                "priceDimensions": {
+                    "YD4JEF3ADAGG84PN.JRTCKXETXF.6YS6EN2CT7": {
+                        "unit": "Hrs",
+                        "endRange": "Inf",
+                        "pricePerUnit": {"USD": str(INSTANCE_PRICE)},
+                    }
+                },
+            }
+        }
+    },
+}
+
+# Response returned by client.get_products
+GET_PRODUCTS_RESPONSE = {"PriceList": [json.dumps(INSTANCE_PRICE_INFORMATION)]}
+
+# Parameters for calling client.get_products
+GET_PRODUCTS_PARAMS = {
+    "Filters": [
+        {
+            "Field": "operatingSystem",
+            "Type": "TERM_MATCH",
+            "Value": "Ubuntu Pro",
+        },
+        {"Field": "instanceType", "Type": "TERM_MATCH", "Value": "c6.2xlarge"},
+        {
+            "Field": "location",
+            "Type": "TERM_MATCH",
+            "Value": "US East (N. Virginia)",
+        },
+    ],
+    "ServiceCode": "AmazonEC2",
+}
+
+# Parameters with on-demand filters
+ON_DEMAND_GET_PRODUCTS_PARAMS = {
+    **GET_PRODUCTS_PARAMS,
+    "Filters": [
+        {"Field": "capacitystatus", "Type": "TERM_MATCH", "Value": "Used"},
+        {"Field": "preInstalledSw", "Type": "TERM_MATCH", "Value": "NA"},
+        {"Field": "tenancy", "Type": "TERM_MATCH", "Value": "shared"},
+    ]
+    + GET_PRODUCTS_PARAMS["Filters"],
+}
+
+
+@pytest.fixture
+def client() -> Iterable[Pricing]:
+    """Return a client for Pricing."""
+    client = boto3.client("pricing", region_name="us-east-1")
+
+    yield Pricing(client=client)
+
+
+def test_ec2_price_information(client: Pricing) -> None:
+    """Test ec2_price_information."""
+    stubber = Stubber(client.client)
+    stubber.add_response(
+        "get_products",
+        GET_PRODUCTS_RESPONSE,
+        GET_PRODUCTS_PARAMS,
+    )
+    with stubber:
+        # The first time the response should be cached so we need only one stub
+        for _ in range(2):
+            price_information = client.ec2_price_information(
+                instance_type="c6.2xlarge", os="Ubuntu Pro", region="us-east-1"
+            )
+
+            assert price_information == [INSTANCE_PRICE_INFORMATION]
+
+
+def test_ec2_on_demand_price(client: Pricing) -> None:
+    """Test ec2_on_demand_price."""
+    stubber = Stubber(client.client)
+    stubber.add_response(
+        "get_products",
+        GET_PRODUCTS_RESPONSE,
+        ON_DEMAND_GET_PRODUCTS_PARAMS,
+    )
+    with stubber:
+        price = client.ec2_on_demand_price(
+            instance_type="c6.2xlarge", os="Ubuntu Pro", region="us-east-1"
+        )
+
+        assert price == INSTANCE_PRICE