#6 breakdown pt4 - add order_book module

cow_py/order_book/api.py

@@ -0,0 +1,203 @@
+import json
+from typing import Any, Dict, List
+
+from cow_py.common.api.api_base import ApiBase, Context
+from cow_py.common.config import SupportedChainId
+from cow_py.order_book.config import OrderBookAPIConfigFactory
+from typing import Union
+from cow_py.order_book.generated.model import OrderQuoteSide2, OrderQuoteValidity2
+
+from .generated.model import (
+    UID,
+    Address,
+    AppDataHash,
+    AppDataObject,
+    NativePriceResponse,
+    Order,
+    OrderCancellation,
+    OrderCreation,
+    OrderQuoteRequest,
+    OrderQuoteResponse,
+    OrderQuoteSide,
+    OrderQuoteSide1,
+    OrderQuoteSide3,
+    OrderQuoteValidity,
+    OrderQuoteValidity1,
+    SolverCompetitionResponse,
+    TotalSurplus,
+    Trade,
+    TransactionHash,
+)
+
+
+class OrderBookApi(ApiBase):
+    def __init__(
+        self,
+        config=OrderBookAPIConfigFactory.get_config("prod", SupportedChainId.MAINNET),
+    ):
+        self.config = config
+
+    async def get_version(self, context_override: Context = {}) -> str:
+        return await self._fetch(
+            path="/api/v1/version", context_override=context_override
+        )
+
+    async def get_trades_by_owner(
+        self, owner: Address, context_override: Context = {}
+    ) -> List[Trade]:
+        response = await self._fetch(
+            path="/api/v1/trades",
+            params={"owner": owner},
+            context_override=context_override,
+        )
+        return [Trade(**trade) for trade in response]
+
+    async def get_trades_by_order_uid(
+        self, order_uid: UID, context_override: Context = {}
+    ) -> List[Trade]:
+        response = await self._fetch(
+            path="/api/v1/trades",
+            params={"order_uid": order_uid},
+            context_override=context_override,
+        )
+        return [Trade(**trade) for trade in response]
+
+    async def get_orders_by_owner(
+        self,
+        owner: Address,
+        limit: int = 1000,
+        offset: int = 0,
+        context_override: Context = {},
+    ) -> List[Order]:
+        return [
+            Order(**order)
+            for order in await self._fetch(
+                path=f"/api/v1/account/{owner}/orders",
+                params={"limit": limit, "offset": offset},
+                context_override=context_override,
+            )
+        ]
+
+    async def get_order_by_uid(
+        self, order_uid: UID, context_override: Context = {}
+    ) -> Order:
+        response = await self._fetch(
+            path=f"/api/v1/orders/{order_uid}",
+            context_override=context_override,
+        )
+        return Order(**response)
+
+    def get_order_link(self, order_uid: UID) -> str:
+        return self.config.get_base_url() + f"/api/v1/orders/{order_uid.root}"
+
+    async def get_tx_orders(
+        self, tx_hash: TransactionHash, context_override: Context = {}
+    ) -> List[Order]:
+        response = await self._fetch(
+            path=f"/api/v1/transactions/{tx_hash}/orders",
+            context_override=context_override,
+        )
+        return [Order(**order) for order in response]
+
+    async def get_native_price(
+        self, tokenAddress: Address, context_override: Context = {}
+    ) -> NativePriceResponse:
+        response = await self._fetch(
+            path=f"/api/v1/token/{tokenAddress}/native_price",
+            context_override=context_override,
+        )
+        return NativePriceResponse(**response)
+
+    async def get_total_surplus(
+        self, user: Address, context_override: Context = {}
+    ) -> TotalSurplus:
+        response = await self._fetch(
+            path=f"/api/v1/users/{user}/total_surplus",
+            context_override=context_override,
+        )
+        return TotalSurplus(**response)
+
+    async def get_app_data(
+        self, app_data_hash: AppDataHash, context_override: Context = {}
+    ) -> Dict[str, Any]:
+        return await self._fetch(
+            path=f"/api/v1/app_data/{app_data_hash}",
+            context_override=context_override,
+        )
+
+    async def get_solver_competition(
+        self, action_id: Union[int, str] = "latest", context_override: Context = {}
+    ) -> SolverCompetitionResponse:
+        response = await self._fetch(
+            path=f"/api/v1/solver_competition/{action_id}",
+            context_override=context_override,
+        )
+        return SolverCompetitionResponse(**response)
+
+    async def get_solver_competition_by_tx_hash(
+        self, tx_hash: TransactionHash, context_override: Context = {}
+    ) -> SolverCompetitionResponse:
+        response = await self._fetch(
+            path=f"/api/v1/solver_competition/by_tx_hash/{tx_hash}",
+            context_override=context_override,
+        )
+        return SolverCompetitionResponse(**response)
+
+    async def post_quote(
+        self,
+        request: OrderQuoteRequest,
+        side: Union[OrderQuoteSide, OrderQuoteSide1, OrderQuoteSide2, OrderQuoteSide3],
+        validity: Union[
+            OrderQuoteValidity, OrderQuoteValidity1, OrderQuoteValidity2
+        ] = OrderQuoteValidity1(validTo=None),
+        context_override: Context = {},
+    ) -> OrderQuoteResponse:
+        response = await self._fetch(
+            path="/api/v1/quote",
+            json={
+                **request.model_dump(by_alias=True),
+                # side object need to be converted to json first to avoid on kind type
+                **json.loads(side.model_dump_json()),
+                **validity.model_dump(),
+            },
+            context_override=context_override,
+            method="POST",
+        )
+        return OrderQuoteResponse(**response)
+
+    async def post_order(self, order: OrderCreation, context_override: Context = {}):
+        response = await self._fetch(
+            path="/api/v1/orders",
+            json=json.loads(order.model_dump_json(by_alias=True)),
+            context_override=context_override,
+            method="POST",
+        )
+        return UID(response)
+
+    async def delete_order(
+        self,
+        orders_cancelation: OrderCancellation,
+        context_override: Context = {},
+    ):
+        response = await self._fetch(
+            path="/api/v1/orders",
+            json=orders_cancelation.model_dump_json(),
+            context_override=context_override,
+            method="DELETE",
+        )
+        return UID(response)
+
+    async def put_app_data(
+        self,
+        app_data: AppDataObject,
+        app_data_hash: str = "",
+        context_override: Context = {},
+    ) -> AppDataHash:
+        app_data_hash_url = app_data_hash if app_data_hash else ""
+        response = await self._fetch(
+            path=f"/api/v1/app_data/{app_data_hash_url}",
+            json=app_data.model_dump_json(),
+            context_override=context_override,
+            method="PUT",
+        )
+        return AppDataHash(response)

cow_py/order_book/config.py

@@ -0,0 +1,39 @@
+from typing import Dict, Literal, Type
+
+from cow_py.common.api.api_base import APIConfig
+from cow_py.common.config import SupportedChainId
+
+
+class ProdAPIConfig(APIConfig):
+    config_map = {
+        SupportedChainId.MAINNET: "https://api.cow.fi/mainnet",
+        SupportedChainId.GNOSIS_CHAIN: "https://api.cow.fi/xdai",
+        SupportedChainId.SEPOLIA: "https://api.cow.fi/sepolia",
+    }
+
+
+class StagingAPIConfig(APIConfig):
+    config_map = {
+        SupportedChainId.MAINNET: "https://barn.api.cow.fi/mainnet",
+        SupportedChainId.GNOSIS_CHAIN: "https://barn.api.cow.fi/xdai",
+        SupportedChainId.SEPOLIA: "https://barn.api.cow.fi/sepolia",
+    }
+
+
+Envs = Literal["prod", "staging"]
+
+
+class OrderBookAPIConfigFactory:
+    config_classes: Dict[Envs, Type[APIConfig]] = {
+        "prod": ProdAPIConfig,
+        "staging": StagingAPIConfig,
+    }
+
+    @staticmethod
+    def get_config(env: Envs, chain_id: SupportedChainId) -> APIConfig:
+        config_class = OrderBookAPIConfigFactory.config_classes.get(env)
+
+        if config_class:
+            return config_class(chain_id)
+        else:
+            raise ValueError("Unknown environment")

examples/order_posting_e2e.py

@@ -0,0 +1,120 @@
+# To run this test you will need to fill the .env file with the necessary variables (see .env.example).
+# You will also need to have enough funds in you wallet of the sell token to create the order.
+# The funds have to already be approved to the CoW Swap Vault Relayer
+
+import asyncio
+import json
+import os
+from dataclasses import asdict
+
+from web3 import Account
+
+from cow_py.common.chains import Chain
+from cow_py.common.config import SupportedChainId
+from cow_py.common.constants import CowContractAddress
+from cow_py.contracts.domain import domain
+from cow_py.contracts.order import Order
+from cow_py.contracts.sign import EcdsaSignature, SigningScheme
+from cow_py.contracts.sign import sign_order as _sign_order
+from cow_py.order_book.api import OrderBookApi
+from cow_py.order_book.config import OrderBookAPIConfigFactory
+from cow_py.order_book.generated.model import OrderQuoteSide1, TokenAmount
+from cow_py.order_book.generated.model import OrderQuoteSideKindSell
+from cow_py.order_book.generated.model import (
+    UID,
+    OrderCreation,
+    OrderQuoteRequest,
+    OrderQuoteResponse,
+)
+
+BUY_TOKEN = "0xfFf9976782d46CC05630D1f6eBAb18b2324d6B14"  # WETH
+SELL_TOKEN = "0xbe72E441BF55620febc26715db68d3494213D8Cb"  # USDC
+SELL_AMOUNT_BEFORE_FEE = "10000000000000000000"  # 100 USDC with 18 decimals
+ORDER_KIND = "sell"
+CHAIN = Chain.SEPOLIA
+CHAIN_ID = SupportedChainId.SEPOLIA
+
+config = OrderBookAPIConfigFactory.get_config("prod", CHAIN_ID)
+ORDER_BOOK_API = OrderBookApi(config)
+
+ADDRESS = os.getenv("USER_ADDRESS")
+ACCOUNT = Account.from_key(os.getenv("PRIVATE_KEY"))
+
+
+async def get_order_quote(
+    order_quote_request: OrderQuoteRequest, order_side: OrderQuoteSide1
+) -> OrderQuoteResponse:
+    return await ORDER_BOOK_API.post_quote(order_quote_request, order_side)
+
+
+def sign_order(order: Order) -> EcdsaSignature:
+    order_domain = asdict(
+        domain(
+            chain=CHAIN, verifying_contract=CowContractAddress.SETTLEMENT_CONTRACT.value
+        )
+    )
+    del order_domain["salt"]  # TODO: improve interfaces
+
+    return _sign_order(order_domain, order, ACCOUNT, SigningScheme.EIP712)
+
+
+async def post_order(order: Order, signature: EcdsaSignature) -> UID:
+    order_creation = OrderCreation(
+        **{
+            "from": ADDRESS,
+            "sellToken": order.sellToken,
+            "buyToken": order.buyToken,
+            "sellAmount": order.sellAmount,
+            "feeAmount": order.feeAmount,
+            "buyAmount": order.buyAmount,
+            "validTo": order.validTo,
+            "kind": order.kind,
+            "partiallyFillable": order.partiallyFillable,
+            "appData": order.appData,
+            "signature": signature.data,
+            "signingScheme": "eip712",
+            "receiver": order.receiver,
+        },
+    )
+    return await ORDER_BOOK_API.post_order(order_creation)
+
+
+async def main():
+    order_quote_request = OrderQuoteRequest(
+        **{
+            "sellToken": SELL_TOKEN,
+            "buyToken": BUY_TOKEN,
+            "from": ADDRESS,
+        }
+    )
+    order_side = OrderQuoteSide1(
+        kind=OrderQuoteSideKindSell.sell,
+        sellAmountBeforeFee=TokenAmount(SELL_AMOUNT_BEFORE_FEE),
+    )
+
+    order_quote = await get_order_quote(order_quote_request, order_side)
+
+    order_quote_dict = json.loads(order_quote.quote.model_dump_json(by_alias=True))
+    order = Order(
+        **{
+            "sellToken": SELL_TOKEN,
+            "buyToken": BUY_TOKEN,
+            "receiver": ADDRESS,
+            "validTo": order_quote_dict["validTo"],
+            "appData": "0x0000000000000000000000000000000000000000000000000000000000000000",
+            "sellAmount": SELL_AMOUNT_BEFORE_FEE,  # Since it is a sell order, the sellAmountBeforeFee is the same as the sellAmount
+            "buyAmount": order_quote_dict["buyAmount"],
+            "feeAmount": "0",  # CoW Swap does not charge fees
+            "kind": ORDER_KIND,
+            "sellTokenBalance": "erc20",
+            "buyTokenBalance": "erc20",
+        }
+    )
+
+    signature = sign_order(order)
+    order_uid = await post_order(order, signature)
+    print(f"order posted on link: {ORDER_BOOK_API.get_order_link(order_uid)}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())