Implemented data serialization/deserialization with marshmallow

requirements.txt

@@ -15,3 +15,4 @@ psycopg2-binary>=2.9.3
 SQLAlchemy==1.4.41
 pandas==1.5.0
 click==8.1.3
+marshmallow==3.18.0

src/missing_prices.py

@@ -9,17 +9,20 @@
 from dune_client.client import DuneClient
 from dune_client.query import Query
 from dune_client.types import DuneRecord
+from marshmallow import fields
 
 from duneapi.api import DuneAPI
 from duneapi.types import Address, DuneQuery, Network
 from duneapi.util import open_query
 
+from src.utils import TokenSchema, CoinSchema, CoinsSchema
+
 DuneTokenPriceRow = tuple[str, str, str, str, int]
 
 
 # TODO - remove the Anys here: https://github.com/cowprotocol/data-misc/issues/20
 def load_coins() -> dict[str, dict[str, Any]]:
-    """ "
+    """
     Loads and returns coin dictionaries from Coin Paprika via their API.
     Excludes, inactive, new and non "token" types
     """
@@ -48,7 +51,10 @@ def load_coins() -> dict[str, dict[str, Any]]:
                 # print(f"Error with {err}, excluding entry {entry}")
 
     print(f"Excluded address for {missed} entries out of {len(entries)}")
-    return coin_dict
+    # return coin_dict
+    return CoinsSchema(keys=fields.Str(), values=fields.Nested(CoinSchema)).load(
+        coin_dict
+    )
 
 
 def write_results(results: list[DuneTokenPriceRow], path: str, filename: str) -> None:
@@ -103,7 +109,8 @@ def as_dune_repr(self, coin_id: str) -> dict[str, Any]:
 
 def load_tokens(dune: DuneClient) -> list[DuneRecord]:
     """Loads Tokens with missing prices from Dune"""
-    return dune.refresh(Query(query_id=1317238, name="Tokens with Missing Prices"))
+    results = dune.refresh(Query(query_id=1317238, name="Tokens with Missing Prices"))
+    return [TokenSchema().load(r) for r in results]
 
 
 def fetch_tokens_without_prices(dune: DuneAPI) -> list[CoinPaprikaToken]:
@@ -128,14 +135,14 @@ def run_missing_prices() -> None:
     print(f"Fetched {len(tokens)} traded tokens from Dune without prices")
     found, res = 0, []
     for token in tokens:
-        if token["address"].lower() in coins:
-            paprika_data = coins[token["address"].lower()]
+        if token["address"] in coins:
+            paprika_data = coins[token["address"]]
             dune_row = (
-                str(paprika_data["id"]),
+                paprika_data["id"],
                 "ethereum",
-                str(paprika_data["symbol"]),
-                str(paprika_data["address"].lower()),
-                int(token["decimals"]),
+                paprika_data["symbol"],
+                paprika_data["address"],
+                token["decimals"],
             )
             res.append(dune_row)
             found += 1

src/utils.py

@@ -3,9 +3,10 @@
 import os
 from datetime import datetime
 from enum import Enum
-from typing import Any
+from typing import Any, Mapping
 
 from duneapi.types import Network as LegacyDuneNetwork
+from marshmallow import fields, Schema
 
 
 def partition_array(arr: list[Any], size: int) -> list[list[Any]]:
@@ -79,3 +80,56 @@ def chain_id(self) -> int:
         Aligned with https://chainlist.org/
         """
         return {Network.MAINNET: 1, Network.GNOSIS: 100}[self]
+
+
+class LoweredString(fields.String):
+    """Custom marshmallow String field for lowered string"""
+
+    def _deserialize(self, value, *args, **kwargs):
+        if hasattr(value, "lower"):
+            value = value.lower()
+        return super()._deserialize(value, *args, **kwargs)
+
+    def _serialize(self, value, attr, obj, **kwargs):
+        if value is None:
+            return ""
+        return str(value).lower()
+
+
+class TokenSchema(Schema):
+    """TokenSchema CoinSchema for serializing/deserializing token data"""
+
+    address = LoweredString(required=True)
+    decimals = fields.Int(required=True)
+    popularity = fields.Int()
+    symbol = fields.String()
+
+
+class CoinSchema(Schema):
+    """CoinSchema for serializing/deserializing coin data"""
+
+    id = fields.String(required=True)
+    name = fields.String()
+    symbol = fields.String(required=True)
+    rank = fields.Int()
+    is_new = fields.Bool()
+    is_active = fields.Bool()
+    type = fields.String()
+    address = LoweredString(required=True)
+
+
+class CoinsSchema(fields.Dict):
+    """CoinsSchema for containing multiple coinschema-s"""
+
+    @staticmethod
+    def _get_obj(obj, _attr, _default):
+        """Accessor for the dump method"""
+        return obj
+
+    def dump(self, obj: Any):
+        """Serializes data"""
+        return self.serialize("", obj, accessor=self._get_obj)
+
+    def load(self, data: Mapping[str, Any]):
+        """Loads data into mapping"""
+        return self.deserialize(data)