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")

tests/order_book/test_api.py

@@ -0,0 +1,147 @@
+from unittest.mock import AsyncMock, Mock, patch
+
+import pytest
+
+from cow_py.order_book.api import OrderBookApi
+from cow_py.order_book.generated.model import OrderQuoteSide1
+from cow_py.order_book.generated.model import OrderQuoteSideKindSell
+from cow_py.order_book.generated.model import TokenAmount
+from cow_py.order_book.generated.model import (
+    OrderQuoteRequest,
+    OrderQuoteResponse,
+    Trade,
+    OrderCreation,
+)
+
+
+@pytest.fixture
+def order_book_api():
+    return OrderBookApi()
+
+
+@pytest.mark.asyncio
+async def test_get_version(order_book_api):
+    expected_version = "1.0.0"
+    with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request:
+        mock_request.return_value = AsyncMock(
+            status_code=200,
+            text=expected_version,
+        )
+        version = await order_book_api.get_version()
+
+        mock_request.assert_awaited_once()
+        assert version == expected_version
+
+
+@pytest.mark.asyncio
+async def test_get_trades_by_order_uid(order_book_api):
+    mock_trade_data = [
+        {
+            "blockNumber": 123456,
+            "logIndex": 789,
+            "orderUid": "mock_order_uid",
+            "owner": "mock_owner_address",
+            "sellToken": "mock_sell_token_address",
+            "buyToken": "mock_buy_token_address",
+            "sellAmount": "100",
+            "sellAmountBeforeFees": "120",
+            "buyAmount": "200",
+            "txHash": "mock_transaction_hash",
+        }
+    ]
+    mock_trade = Trade(**mock_trade_data[0])
+    with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request:
+        mock_request.return_value = AsyncMock(
+            status_code=200,
+            headers={"content-type": "application/json"},
+            json=Mock(return_value=mock_trade_data),
+        )
+        trades = await order_book_api.get_trades_by_order_uid("mock_order_uid")
+        mock_request.assert_awaited_once()
+        assert trades == [mock_trade]
+
+
+@pytest.mark.asyncio
+async def test_post_quote(order_book_api):
+    mock_order_quote_request = OrderQuoteRequest(
+        **{
+            "sellToken": "0x",
+            "buyToken": "0x",
+            "receiver": "0x",
+            "appData": "app_data_object",
+            "appDataHash": "0x",
+            "from": "0x",
+            "priceQuality": "verified",
+            "signingScheme": "eip712",
+            "onchainOrder": False,
+        }
+    )
+
+    mock_order_quote_side = OrderQuoteSide1(
+        sellAmountBeforeFee=TokenAmount("0"), kind=OrderQuoteSideKindSell.sell
+    )
+    mock_order_quote_response_data = {
+        "quote": {
+            "sellToken": "0x",
+            "buyToken": "0x",
+            "receiver": "0x",
+            "sellAmount": "0",
+            "buyAmount": "0",
+            "feeAmount": "0",
+            "validTo": 0,
+            "appData": "0x",
+            "partiallyFillable": True,
+            "sellTokenBalance": "erc20",
+            "buyTokenBalance": "erc20",
+            "kind": "buy",
+        },
+        "verified": True,
+        "from": "0x",
+        "expiration": "0",
+    }
+    mock_order_quote_response = OrderQuoteResponse(**mock_order_quote_response_data)
+    with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request:
+        mock_request.return_value = AsyncMock(
+            status_code=200,
+            headers={"content-type": "application/json"},
+            json=Mock(return_value=mock_order_quote_response_data),
+        )
+        response = await order_book_api.post_quote(
+            mock_order_quote_request, mock_order_quote_side
+        )
+        mock_request.assert_awaited_once()
+        assert response == mock_order_quote_response
+
+
+@pytest.mark.asyncio
+async def test_post_order(order_book_api):
+    mock_response = "mock_uid"
+    mock_order_creation = OrderCreation(
+        **{
+            "sellToken": "0x",
+            "buyToken": "0x",
+            "sellAmount": "0",
+            "buyAmount": "0",
+            "validTo": 0,
+            "feeAmount": "0",
+            "kind": "buy",
+            "partiallyFillable": True,
+            "appData": "0x",
+            "signingScheme": "eip712",
+            "signature": "0x",
+            "receiver": "0x",
+            "sellTokenBalance": "erc20",
+            "buyTokenBalance": "erc20",
+            "quoteId": 0,
+            "appDataHash": "0x",
+            "from_": "0x",
+        }
+    )
+    with patch("httpx.AsyncClient.request", new_callable=AsyncMock) as mock_request:
+        mock_request.return_value = AsyncMock(
+            status_code=200,
+            text=mock_response,
+        )
+        response = await order_book_api.post_order(mock_order_creation)
+        mock_request.assert_awaited_once()
+        assert response.root == mock_response