#6 breakdown pt1 - add codegen

Makefile

@@ -2,8 +2,8 @@
 
 codegen: web3_codegen orderbook_codegen subgraph_codegen
 
-# web3_codegen:
-# 	poetry run web3_codegen
+web3_codegen:
+	poetry run python -m cow_py.codegen.main
 
 orderbook_codegen:
 	poetry run datamodel-codegen --url="https://raw.githubusercontent.com/cowprotocol/services/v2.245.1/crates/orderbook/openapi.yml" --output cow_py/order_book/generated/model.py --target-python-version 3.12 --output-model-type pydantic_v2.BaseModel --input-file-type openapi

README.md

@@ -122,14 +122,62 @@ make format # or ruff check . --fix
 make lint # or ruff format
 ```
 
-### 🐄 Codegen
+### 🐄 Code Generation
 
-Generate the SDK from the CoW Protocol smart contracts, Subgraph, and Orderbook API:
+The SDK uses various code generation tools for different components. Here's how to work with them:
+
+#### Full Code Generation
+
+To run all code generation processes:
 
 ```bash
 make codegen
 ```
 
+This command runs three separate code generation tasks:
+
+1. Web3 Codegen
+2. Orderbook Codegen
+3. Subgraph Codegen
+
+#### Individual Code Generation Tasks
+
+You can also run these tasks individually:
+
+1. Web3 Codegen:
+
+   ```bash
+   make web3_codegen
+   ```
+
+   This runs `python -m cow_py.codegen.main`, which processes the ABIs in the `cow_py/contracts/abi` directory and generates corresponding Python classes.
+
+2. Orderbook Codegen:
+
+   ```bash
+   make orderbook_codegen
+   ```
+
+   This uses `datamodel-codegen` to generate models from the CoW Protocol Orderbook OpenAPI specification.
+
+3. Subgraph Codegen:
+
+   ```bash
+   make subgraph_codegen
+   ```
+
+   This uses `ariadne-codegen` to generate code for interacting with the CoW Protocol subgraph.
+
+#### When to Update Generated Code
+
+You should run the appropriate code generation task when:
+
+1. There are changes to the smart contract ABIs (use `web3_codegen`).
+2. The Orderbook API specification is updated (use `orderbook_codegen`).
+3. The subgraph schema changes (use `subgraph_codegen`).
+4. You modify any of the code generation templates or logic.
+
+It's a good practice to run `make codegen` as part of your development process, especially before committing changes that might affect these generated components.
 
 ## 🐄 Contributing to the Herd
 
@@ -141,10 +189,11 @@ cd cow-py
 poetry install
 ```
 
-Run tests to ensure everything's working:
+After making changes, make sure to run the appropriate code generation tasks and tests:
 
 ```bash
-poetry run pytest
+make codegen
+make test
 ```
 
 ## 🐄 Need Help?

cow_py/common/api/api_base.py

@@ -0,0 +1,85 @@
+from abc import ABC
+from typing import Any, Optional
+
+import httpx
+
+from cow_py.common.api.decorators import rate_limitted, with_backoff
+from cow_py.common.config import SupportedChainId
+
+Context = dict[str, Any]
+
+
+class APIConfig(ABC):
+    """Base class for API configuration with common functionality."""
+
+    config_map = {}
+
+    def __init__(
+        self, chain_id: SupportedChainId, base_context: Optional[Context] = None
+    ):
+        self.chain_id = chain_id
+        self.context = base_context or {}
+
+    def get_base_url(self) -> str:
+        return self.config_map.get(
+            self.chain_id, "default URL if chain_id is not found"
+        )
+
+    def get_context(self) -> Context:
+        return {"base_url": self.get_base_url(), **self.context}
+
+
+class RequestStrategy:
+    async def make_request(self, client, url, method, **request_kwargs):
+        headers = {
+            "accept": "application/json",
+            "content-type": "application/json",
+        }
+
+        return await client.request(
+            url=url, headers=headers, method=method, **request_kwargs
+        )
+
+
+class ResponseAdapter:
+    async def adapt_response(self, _response):
+        raise NotImplementedError()
+
+
+class RequestBuilder:
+    def __init__(self, strategy, response_adapter):
+        self.strategy = strategy
+        self.response_adapter = response_adapter
+
+    async def execute(self, client, url, method, **kwargs):
+        response = await self.strategy.make_request(client, url, method, **kwargs)
+        return self.response_adapter.adapt_response(response)
+
+
+class JsonResponseAdapter(ResponseAdapter):
+    def adapt_response(self, response):
+        if response.headers.get("content-type") == "application/json":
+            return response.json()
+        else:
+            return response.text
+
+
+class ApiBase:
+    """Base class for APIs utilizing configuration and request execution."""
+
+    def __init__(self, config: APIConfig):
+        self.config = config
+
+    @with_backoff()
+    @rate_limitted()
+    async def _fetch(self, path, method="GET", **kwargs):
+        url = self.config.get_base_url() + path
+
+        del kwargs["context_override"]
+
+        async with httpx.AsyncClient() as client:
+            builder = RequestBuilder(
+                RequestStrategy(),
+                JsonResponseAdapter(),
+            )
+            return await builder.execute(client, url, method, **kwargs)

cow_py/common/api/decorators.py

@@ -0,0 +1,58 @@
+import backoff
+import httpx
+from aiolimiter import AsyncLimiter
+
+DEFAULT_LIMITER_OPTIONS = {"rate": 5, "per": 1.0}
+
+DEFAULT_BACKOFF_OPTIONS = {
+    "max_tries": 10,
+    "max_time": None,
+    "jitter": None,
+}
+
+
+def dig(self, *keys):
+    try:
+        for key in keys:
+            self = self[key]
+        return self
+    except KeyError:
+        return None
+
+
+def with_backoff():
+    def decorator(func):
+        async def wrapper(*args, **kwargs):
+            backoff_opts = dig(kwargs, "context_override", "backoff_opts")
+
+            if backoff_opts is None:
+                internal_backoff_opts = DEFAULT_BACKOFF_OPTIONS
+            else:
+                internal_backoff_opts = backoff_opts
+
+            @backoff.on_exception(
+                backoff.expo, httpx.HTTPStatusError, **internal_backoff_opts
+            )
+            async def closure():
+                return await func(*args, **kwargs)
+
+            return await closure()
+
+        return wrapper
+
+    return decorator
+
+
+def rate_limitted(
+    rate=DEFAULT_LIMITER_OPTIONS["rate"], per=DEFAULT_LIMITER_OPTIONS["per"]
+):
+    limiter = AsyncLimiter(rate, per)
+
+    def decorator(func):
+        async def wrapper(*args, **kwargs):
+            async with limiter:
+                return await func(*args, **kwargs)
+
+        return wrapper
+
+    return decorator

cow_py/common/chains/__init__.py

@@ -6,24 +6,26 @@ class Chain(Enum):
     Supported chains and their respective `chainId` for the SDK.
     """
 
-    MAINNET = 1
-    GNOSIS = 100
-    SEPOLIA = 11155111
+    MAINNET = (1, "ethereum", "https://etherscan.io")
+    GNOSIS = (100, "gnosis", "https://gnosisscan.io")
+    SEPOLIA = (11155111, "sepolia", "https://sepolia.etherscan.io")
 
-    def __init__(self, id) -> None:
+    def __init__(self, id: int, network_name: str, explorer_url: str) -> None:
         self.id = id
+        self.network_name = network_name
+        self.explorer_url = explorer_url
 
+    @property
+    def name(self) -> str:
+        return self.network_name
 
-SUPPORTED_CHAINS = {Chain.MAINNET, Chain.GNOSIS, Chain.SEPOLIA}
+    @property
+    def explorer(self) -> str:
+        return self.explorer_url
 
-CHAIN_NAMES = {
-    Chain.MAINNET: "ethereum",
-    Chain.GNOSIS: "gnosis",
-    Chain.SEPOLIA: "sepolia",
-}
+    @property
+    def chain_id(self) -> int:
+        return self.id
 
-CHAIN_SCANNER_MAP = {
-    Chain.MAINNET: "https://etherscan.io",
-    Chain.GNOSIS: "https://gnosisscan.io",
-    Chain.SEPOLIA: "https://sepolia.etherscan.io/",
-}
+
+SUPPORTED_CHAINS = {chain for chain in Chain}

cow_py/common/chains/utils.py

@@ -1,6 +1,6 @@
-from cow_py.common.chains import CHAIN_SCANNER_MAP, Chain
+from cow_py.common.chains import Chain
 
 
 def get_explorer_link(chain: Chain, tx_hash: str) -> str:
     """Return the scan link for the provided transaction hash."""
-    return f"{CHAIN_SCANNER_MAP[chain]}/tx/{tx_hash}"
+    return f"{chain.explorer_url}/tx/{tx_hash}"

cow_py/common/config.py

@@ -1,4 +1,35 @@
+from dataclasses import dataclass
 from enum import Enum
+from typing import Dict, Optional
+
+
+class SupportedChainId(Enum):
+    MAINNET = 1
+    GNOSIS_CHAIN = 100
+    SEPOLIA = 11155111
+
+
+class CowEnv(Enum):
+    PROD = "prod"
+    STAGING = "staging"
+
+
+ApiBaseUrls = Dict[SupportedChainId, str]
+
+
+@dataclass
+class ApiContext:
+    chain_id: SupportedChainId
+    env: CowEnv
+    base_urls: Optional[ApiBaseUrls] = None
+    max_tries: Optional[int] = 5
+
+
+# Define the list of available environments.
+ENVS_LIST = [CowEnv.PROD, CowEnv.STAGING]
+
+# Define the default CoW Protocol API context.
+DEFAULT_COW_API_CONTEXT = ApiContext(env=CowEnv.PROD, chain_id=SupportedChainId.MAINNET)
 
 
 class IPFSConfig(Enum):

cow_py/common/constants.py

@@ -1,5 +1,6 @@
 from enum import Enum
 from typing import Dict
+
 from .chains import Chain
 
 """
@@ -17,14 +18,14 @@ class CowContractAddress(Enum):
     EXTENSIBLE_FALLBACK_HANDLER = "0x2f55e8b20D0B9FEFA187AA7d00B6Cbe563605bF5"
 
 
-def map_address_to_supported_networks(address) -> Dict[Chain, str]:
+def map_address_to_supported_networks(address) -> Dict[int, str]:
     """
     Maps a given address to all supported networks.
 
     :param address: The address to be mapped.
     :return: A dictionary mapping the address to each supported chain.
     """
-    return {chain_id: address for chain_id in Chain}
+    return {chain.chain_id: address for chain in Chain}
 
 
 COW_PROTOCOL_SETTLEMENT_CONTRACT_CHAIN_ADDRESS_MAP = map_address_to_supported_networks(