Add app data repo functions (#14)

* refactor repo configs/readme

* add codegen module

* add common module

* remove stale subgraphs/core modules

* add contracts module

* add order_book module

* add order posting example

* add generated subgraph files

* add contract abis

* add generated codegen

* add generated order_book

* add downloaded schemas

* add utils of app data module

* wip: start convertion from hex to cid

* add generated api model

* add python-dotenv package

* refactor importing sort

* refactor .env usage and variables requested

* add appDataHex object and its convertion to cid

* wip: add converstions between cid, hex and doc

* rename files to snakecase

* enable env modify ipfs urls

* add app data doc tests

* wip: start app data doc to cid

* add app data and remove legacy code

* fix digest util functions

* fix import

* chore: relock poetry.lock

* chore: add multiformats

* chore(app_data): remove unused app_data schemas

---------

Co-authored-by: José Ribeiro <[email protected]>

README.md

@@ -164,6 +164,34 @@ Generate the SDK from the CoW Protocol smart contracts, Subgraph, and Orderbook
 make codegen
 ```
 
+## 🐄 Development
+
+### 🐄 Tests
+
+Run tests to ensure everything's working:
+
+```bash
+make test # or poetry run pytest
+```
+
+### 🐄 Formatting/Linting
+
+Run the formatter and linter:
+
+```bash
+make format # or ruff check . --fix
+make lint # or ruff format
+```
+
+### 🐄 Codegen
+
+Generate the SDK from the CoW Protocol smart contracts, Subgraph, and Orderbook API:
+
+```bash
+make codegen
+```
+
+
 ## 🐄 Contributing to the Herd
 
 Interested in contributing? Here's how you can help:

cow_py/app_data/app_data_cid.py

@@ -0,0 +1,15 @@
+from typing import Any, Dict
+
+from cow_py.app_data.consts import DEFAULT_IPFS_READ_URI
+from cow_py.app_data.utils import extract_digest, fetch_doc_from_cid
+
+
+class AppDataCid:
+    def __init__(self, app_data_cid: str):
+        self.app_data_cid = app_data_cid
+
+    async def to_doc(self, ipfs_uri: str = DEFAULT_IPFS_READ_URI) -> Dict[str, Any]:
+        return await fetch_doc_from_cid(self.app_data_cid, ipfs_uri)
+
+    def to_hex(self) -> str:
+        return extract_digest(self.app_data_cid)

cow_py/app_data/app_data_doc.py

@@ -0,0 +1,30 @@
+from typing import Dict, Any
+
+from eth_utils.crypto import keccak
+
+from cow_py.app_data.app_data_hex import AppDataHex
+from cow_py.app_data.consts import DEFAULT_APP_DATA_DOC
+from cow_py.app_data.utils import stringify_deterministic
+
+
+class AppDataDoc:
+    def __init__(
+        self, app_data_doc: Dict[str, Any] = {}, app_data_doc_string: str = ""
+    ):
+        self.app_data_doc = {**DEFAULT_APP_DATA_DOC, **app_data_doc}
+        self.app_data_doc_string = app_data_doc_string
+
+    def to_string(self) -> str:
+        if self.app_data_doc_string:
+            return self.app_data_doc_string
+        return stringify_deterministic(self.app_data_doc)
+
+    def to_hex(self) -> str:
+        # TODO: add validation of app data
+        full_app_data_json = self.to_string()
+        data_bytes = full_app_data_json.encode("utf-8")
+        return "0x" + keccak(data_bytes).hex()
+
+    def to_cid(self) -> str:
+        appDataHex = AppDataHex(self.to_hex()[2:])
+        return appDataHex.to_cid()

cow_py/app_data/app_data_hex.py

@@ -0,0 +1,63 @@
+from typing import Dict, Any
+from web3 import Web3
+from multiformats import multibase
+
+from cow_py.app_data.consts import DEFAULT_IPFS_READ_URI, MetaDataError
+from cow_py.app_data.utils import fetch_doc_from_cid
+
+CID_V1_PREFIX = 0x01
+CID_RAW_MULTICODEC = 0x55
+KECCAK_HASHING_ALGORITHM = 0x1B
+KECCAK_HASHING_LENGTH = 32
+CID_DAG_PB_MULTICODEC = 0x70
+SHA2_256_HASHING_ALGORITHM = 0x12
+SHA2_256_HASHING_LENGTH = 32
+
+
+class AppDataHex:
+    def __init__(self, app_data_hex: str):
+        self.app_data_hex = app_data_hex
+
+    def to_cid(self) -> str:
+        cid = self._app_data_hex_to_cid()
+        self._assert_cid(cid)
+        return cid
+
+    async def to_doc(self, ipfs_uri: str = DEFAULT_IPFS_READ_URI) -> Dict[str, Any]:
+        try:
+            cid = self.to_cid()
+            return await fetch_doc_from_cid(cid, ipfs_uri)
+        except Exception as e:
+            raise MetaDataError(
+                f"Unexpected error decoding AppData: appDataHex={self.app_data_hex}, message={e}"
+            )
+
+    def _assert_cid(self, cid: str):
+        if not cid:
+            raise MetaDataError(
+                f"Error getting CID from appDataHex: {self.app_data_hex}"
+            )
+
+    def _app_data_hex_to_cid(self) -> str:
+        cid_bytes = self._to_cid_bytes(
+            {
+                "version": CID_V1_PREFIX,
+                "multicodec": CID_RAW_MULTICODEC,
+                "hashing_algorithm": KECCAK_HASHING_ALGORITHM,
+                "hashing_length": KECCAK_HASHING_LENGTH,
+                "multihash_hex": self.app_data_hex,
+            }
+        )
+        return multibase.encode(cid_bytes, "base16")
+
+    def _to_cid_bytes(self, params: Dict[str, Any]) -> bytes:
+        hash_bytes = Web3.to_bytes(hexstr=params["multihash_hex"])
+        cid_prefix = bytes(
+            [
+                params["version"],
+                params["multicodec"],
+                params["hashing_algorithm"],
+                params["hashing_length"],
+            ]
+        )
+        return cid_prefix + hash_bytes

cow_py/app_data/consts.py

@@ -0,0 +1,15 @@
+import os
+
+DEFAULT_IPFS_READ_URI = os.getenv("IPFS_READ_URI", "https://cloudflare-ipfs.com/ipfs")
+
+LATEST_APP_DATA_VERSION = "1.1.0"
+DEFAULT_APP_CODE = "CoW Swap"
+DEFAULT_APP_DATA_DOC = {
+    "appCode": DEFAULT_APP_CODE,
+    "metadata": {},
+    "version": LATEST_APP_DATA_VERSION,
+}
+
+
+class MetaDataError(Exception):
+    pass

cow_py/app_data/utils.py

@@ -0,0 +1,39 @@
+from typing import Any, Dict
+import httpx
+from multiformats import CID
+from collections.abc import Mapping
+
+
+import json
+
+from cow_py.app_data.consts import DEFAULT_IPFS_READ_URI
+
+# CID uses multibase to self-describe the encoding used (See https://github.com/multiformats/multibase)
+#   - Most reference implementations (multiformats/cid or Pinata, etc) use base58btc encoding
+#   - However, the backend uses base16 encoding (See https://github.com/cowprotocol/services/blob/main/crates/app-data-hash/src/lib.rs#L64)
+MULTIBASE_BASE16 = "f"
+
+
+def extract_digest(cid_str: str) -> str:
+    cid = CID.decode(cid_str)
+    return "0x" + cid.raw_digest.hex()
+
+
+def sort_nested_dict(d):
+    return {
+        k: sort_nested_dict(v) if isinstance(v, Mapping) else v
+        for k, v in sorted(d.items())
+    }
+
+
+def stringify_deterministic(obj):
+    sorted_dict = sort_nested_dict(obj)
+    return json.dumps(sorted_dict, sort_keys=True, separators=(",", ":"))
+
+
+async def fetch_doc_from_cid(
+    cid: str, ipfs_uri: str = DEFAULT_IPFS_READ_URI
+) -> Dict[str, Any]:
+    async with httpx.AsyncClient() as client:
+        response = await client.get(f"{ipfs_uri}/{cid}")
+        return response.json()

pyproject.toml

@@ -14,6 +14,7 @@ pydantic = "^2.7.0"
 pytest-mock = "^3.14.0"
 backoff = "^2.2.1"
 aiolimiter = "^1.1.0"
+multiformats = "^0.3.1.post4"
 
 
 [tool.poetry.group.dev.dependencies]

tests/app_data/app_data_cid.py

@@ -0,0 +1,44 @@
+import httpx
+import pytest
+from unittest.mock import patch
+from cow_py.app_data.app_data_cid import AppDataCid
+from cow_py.app_data.consts import DEFAULT_IPFS_READ_URI
+from .mocks import APP_DATA_HEX, CID, APP_DATA_HEX_2, CID_2
+
+
+@pytest.mark.asyncio
+async def test_fetch_doc_from_cid():
+    valid_serialized_cid = "QmZZhNnqMF1gRywNKnTPuZksX7rVjQgTT3TJAZ7R6VE3b2"
+    expected = {
+        "appCode": "CowSwap",
+        "metadata": {
+            "referrer": {
+                "address": "0x1f5B740436Fc5935622e92aa3b46818906F416E9",
+                "version": "0.1.0",
+            }
+        },
+        "version": "0.1.0",
+    }
+
+    with patch("httpx.AsyncClient.get") as mock_get:
+        mock_get.return_value = httpx.Response(200, json=expected)
+
+        app_data_hex = AppDataCid(valid_serialized_cid)
+        app_data_document = await app_data_hex.to_doc()
+
+    assert app_data_document == expected
+    mock_get.assert_called_once_with(f"{DEFAULT_IPFS_READ_URI}/{valid_serialized_cid}")
+
+
+def test_app_data_cid_to_hex():
+    decoded_app_data_hex = CID.to_hex()
+    assert decoded_app_data_hex == APP_DATA_HEX.app_data_hex
+
+    decoded_app_data_hex_2 = CID_2.to_hex()
+    assert decoded_app_data_hex_2 == APP_DATA_HEX_2.app_data_hex
+
+
+def test_app_data_cid_to_hex_invalid_hash():
+    app_data_cid = AppDataCid("invalidCid")
+    with pytest.raises(Exception):
+        app_data_cid.to_hex()