Merge pull request #3 from cowdao-grants/swap-on-cowswap

Swap on cowswap

.github/actions/python_prepare/action.yaml

@@ -0,0 +1,15 @@
+name: "Prepare Python environment"
+description: "Set up Python and install dependencies"
+runs:
+  using: "composite"
+  steps:
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v2
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install Poetry
+      shell: bash
+      run: curl -sSL https://install.python-poetry.org | python3 -
+    - name: Install dependencies
+      shell: bash
+      run: poetry install --all-extras

.github/workflows/python_cd.yaml

@@ -0,0 +1,23 @@
+name: Python CD
+
+on:
+  release:
+    types: [ published ]
+
+jobs:
+  publish-pypi-package:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: write
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@v3
+        with:
+          token: ${{ secrets.GITHUB_TOKEN }}
+      - name: Verify the tag version in the pyproject.toml
+        run: grep -q "version = \"${{ github.event.release.tag_name }}\"" pyproject.toml || exit 1
+        shell: bash
+      - uses: ./.github/actions/python_prepare
+      - name: Build and Publish
+        run: poetry publish -p ${{ secrets.PYPI_TOKEN }} -u "__token__" --build
+        shell: bash

.github/workflows/python_ci.yaml

@@ -0,0 +1,59 @@
+name: Python CI
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+  workflow_dispatch:
+
+env:
+  COWSWAP_TEST_PRIVATE_KEY: ${{ secrets.COWSWAP_TEST_PRIVATE_KEY }}
+
+jobs:
+  mypy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: ./.github/actions/python_prepare
+      - name: Run mypy
+        run: poetry run mypy
+
+  pytest:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [ '3.10.x', '3.11.x', '3.12.x' ]
+    name: pytest - Python ${{ matrix.python-version }} 
+    steps:
+      - uses: actions/checkout@v2
+      - uses: ./.github/actions/python_prepare
+      - name: Run pytest unit tests
+        run: poetry run python -m pytest tests/
+
+  black:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: ./.github/actions/python_prepare
+      - name: Check with black
+        run: poetry run black --check .
+
+  autoflake:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: ./.github/actions/python_prepare
+      - name: Check with autoflake
+        run: |
+          poetry run autoflake --in-place --remove-all-unused-imports --remove-unused-variables --recursive .
+          git diff --exit-code --quiet || exit 1
+
+  isort:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - uses: ./.github/actions/python_prepare
+      - name: Check with isort
+        run: |
+          poetry run isort --profile black .
+          git diff --exit-code --quiet || exit 1

cowswap_client/cow_client.py

@@ -0,0 +1,115 @@
+from typing import Any
+
+import requests
+from eth_account import Account
+from eth_account.signers.local import LocalAccount
+from eth_typing import ChecksumAddress
+from loguru import logger
+from web3 import Web3
+
+from cowswap_client.encoding import DOMAIN, MESSAGE_TYPES, MESSAGE_TYPES_CANCELLATION
+from cowswap_client.gtypes import Wei
+from cowswap_client.models import (
+    CowServer,
+    OrderKind,
+    OrderStatus,
+    QuoteInput,
+    QuoteOutput,
+)
+
+
+class CowClient:
+    def __init__(
+        self, account: LocalAccount, api_url: CowServer = CowServer.GNOSIS_STAGING
+    ):
+        self.api_url = api_url
+        self.account = account
+
+    def get_version(self) -> str:
+        r = requests.get(f"{self.api_url.value}/api/v1/version")
+        return r.text
+
+    def build_swap_params(
+        self, sell_token: ChecksumAddress, buy_token: ChecksumAddress, sell_amount: Wei
+    ) -> QuoteInput:
+        quote = QuoteInput(
+            from_=self.account.address,
+            sell_token=sell_token,
+            buy_token=buy_token,
+            receiver=self.account.address,
+            sell_amount_before_fee=str(sell_amount),
+            kind=OrderKind.SELL,
+            app_data="0x0000000000000000000000000000000000000000000000000000000000000000",
+            valid_for=1080,
+        )
+        return quote
+
+    @staticmethod
+    def _if_error_log_and_raise(r: requests.Response) -> None:
+        try:
+            r.raise_for_status()
+        except Exception as e:
+            logger.error(f"Error occured on response: {r.text}, Exception - {e}")
+            raise
+
+    def post_quote(self, quote: QuoteInput) -> QuoteOutput:
+        quote_dict = quote.model_dump(by_alias=True, exclude_none=True)
+        r = requests.post(f"{self.api_url.value}/api/v1/quote", json=quote_dict)
+
+        self._if_error_log_and_raise(r)
+        return QuoteOutput.model_validate(r.json()["quote"])
+
+    @staticmethod
+    def build_order_with_fee_and_sell_amounts(quote: QuoteOutput) -> dict[str, Any]:
+        quote.sell_amount = str(int(quote.sell_amount) + int(quote.fee_amount))
+        quote.fee_amount = "0"
+        quote_dict = quote.model_dump(by_alias=True, exclude_none=True)
+        return quote_dict
+
+    def post_order(self, quote: QuoteOutput, web3: Web3 | None = None) -> str:
+        # Note that allowance is not checked - please make sure you have enough allowance to make the order.
+        order_data = self.build_order_with_fee_and_sell_amounts(quote)
+
+        # sign
+        signed_message = Account.sign_typed_data(
+            self.account.key, DOMAIN, MESSAGE_TYPES, order_data
+        )
+        order_data["signature"] = signed_message.signature.hex()
+        # post
+        r = requests.post(f"{self.api_url.value}/api/v1/orders", json=order_data)
+        self._if_error_log_and_raise(r)
+        order_id = r.content.decode().replace('"', "")
+        return order_id
+
+    def cancel_order_if_not_already_cancelled(self, order_uids: list[str]) -> None:
+        signed_message_cancellation = Account.sign_typed_data(
+            self.account.key,
+            DOMAIN,
+            MESSAGE_TYPES_CANCELLATION,
+            {"orderUids": order_uids},
+        )
+        cancellation_request_obj = {
+            "orderUids": order_uids,
+            "signature": signed_message_cancellation.signature.hex(),
+            "signingScheme": "eip712",
+        }
+
+        order_status = self.get_order_status(order_uids[0])
+        if order_status == OrderStatus.CANCELLED:
+            return
+
+        r = requests.delete(
+            f"{self.api_url.value}/api/v1/orders", json=cancellation_request_obj
+        )
+        r.raise_for_status()
+
+    def get_order_status(self, order_uid: str) -> OrderStatus:
+        r = requests.get(f"{self.api_url.value}/api/v1/orders/{order_uid}/status")
+        r.raise_for_status()
+
+        order_type = r.json()["type"]
+        if order_type not in iter(OrderStatus):
+            raise ValueError(
+                f"order_type {order_type} from order_uid {order_uid} cannot be processed."
+            )
+        return OrderStatus(order_type)

cowswap_client/encoding.py

@@ -0,0 +1,43 @@
+from eth_typing import ChecksumAddress
+from web3 import Web3
+
+from cowswap_client.gtypes import ChainID
+
+MESSAGE_TYPES_CANCELLATION = {
+    "OrderCancellations": [
+        {"name": "orderUids", "type": "bytes[]"},
+    ]
+}
+
+# EIP-712 Types
+MESSAGE_TYPES = {
+    "Order": [
+        {"name": "sellToken", "type": "address"},
+        {"name": "buyToken", "type": "address"},
+        {"name": "receiver", "type": "address"},
+        {"name": "sellAmount", "type": "uint256"},
+        {"name": "buyAmount", "type": "uint256"},
+        {"name": "validTo", "type": "uint32"},
+        {"name": "appData", "type": "bytes32"},
+        {"name": "feeAmount", "type": "uint256"},
+        {"name": "kind", "type": "string"},
+        {"name": "partiallyFillable", "type": "bool"},
+        {"name": "sellTokenBalance", "type": "string"},
+        {"name": "buyTokenBalance", "type": "string"},
+    ]
+}
+
+# EIP-712 Domain
+DOMAIN = {
+    "name": "Gnosis Protocol",
+    "version": "v2",
+    "chainId": 100,  # Replace with actual chainId
+    "verifyingContract": "0x9008D19f58AAbD9eD0D60971565AA8510560ab41",  # Gnosis Mainnet, from
+}
+
+# Relayer address for allowance purposes
+RELAYER_ADDRESSES: dict[ChainID, ChecksumAddress] = {
+    ChainID(100): Web3.to_checksum_address(
+        "0xC92E8bdf79f0507f65a392b0ab4667716BFE0110"
+    ),  # Gnosis
+}

cowswap_client/gtypes.py

@@ -0,0 +1,12 @@
+from typing import NewType
+
+from pydantic.types import SecretStr
+from web3.types import (  # noqa: F401  # Import for the sake of easy importing with others from here.
+    Nonce,
+    TxParams,
+    TxReceipt,
+    Wei,
+)
+
+PrivateKey = NewType("PrivateKey", SecretStr)
+ChainID = NewType("ChainID", int)

cowswap_client/models.py

@@ -0,0 +1,75 @@
+from enum import Enum
+from typing import Optional
+
+from pydantic import BaseModel, ConfigDict, Field, model_validator
+from typing_extensions import Self
+
+
+class OrderKind(str, Enum):
+    BUY = "buy"
+    SELL = "sell"
+
+
+class CowServer(str, Enum):
+    GNOSIS_PROD = "https://api.cow.fi/xdai"
+    GNOSIS_STAGING = "https://barn.api.cow.fi/xdai"
+
+
+class BaseQuote(BaseModel):
+    sell_token: str = Field(alias="sellToken")
+    buy_token: str = Field(alias="buyToken")
+    receiver: str
+    app_data: Optional[str] = Field(default=None, alias="appData")
+    sell_token_balance: str = Field(default="erc20", alias="sellTokenBalance")
+    buy_token_balance: str = Field(default="erc20", alias="buyTokenBalance")
+    price_quality: str = Field(default="fast", alias="priceQuality")
+    signing_scheme: str = Field(default="eip712", alias="signingScheme")
+    partially_fillable: bool = Field(default=False, alias="partiallyFillable")
+    kind: OrderKind = Field(default=OrderKind.BUY)
+
+
+class QuoteOutput(BaseQuote):
+    fee_amount: str = Field(alias="feeAmount")
+    buy_amount: str = Field(alias="buyAmount")
+    sell_amount: str = Field(alias="sellAmount")
+    valid_to: int = Field(alias="validTo")
+
+    @model_validator(mode="after")
+    def check_either_buy_or_sell_amount_set(self) -> Self:
+        if self.sell_amount is None and self.buy_amount is None:
+            raise ValueError("neither buy nor sell amounts set")
+        if self.kind == "sell" and self.sell_amount is None:
+            raise ValueError("sellAmountBeforeFee not set")
+        elif self.kind == "buy" and self.buy_amount is None:
+            raise ValueError("buyAmountAfterFee not set")
+        return self
+
+
+class QuoteInput(BaseQuote):
+    from_: Optional[str] = Field(default=None, alias="from")
+    sell_amount_before_fee: Optional[str] = Field(
+        default=None, alias="sellAmountBeforeFee"
+    )
+    buy_amount_after_fee: Optional[str] = Field(default=None, alias="buyAmountAfterFee")
+    model_config = ConfigDict(populate_by_name=True)
+    valid_for: int = Field(alias="validFor")
+
+    @model_validator(mode="after")
+    def check_either_buy_or_sell_amount_set(self) -> Self:
+        if self.sell_amount_before_fee is None and self.buy_amount_after_fee is None:
+            raise ValueError("neither buy nor sell amounts set")
+        if self.kind == "sell" and self.sell_amount_before_fee is None:
+            raise ValueError("sellAmountBeforeFee not set")
+        elif self.kind == "buy" and self.buy_amount_after_fee is None:
+            raise ValueError("buyAmountAfterFee not set")
+        return self
+
+
+class OrderStatus(str, Enum):
+    OPEN = "open"
+    SCHEDULED = "scheduled"
+    ACTIVE = "active"
+    SOLVED = "solved"
+    EXECUTING = "executing"
+    TRADED = "traded"
+    CANCELLED = "cancelled"

cowswap_client/utils.py

@@ -0,0 +1,48 @@
+from typing import NoReturn, Optional, Type, TypeVar
+
+T = TypeVar("T")
+
+
+def should_not_happen(
+    msg: str = "Should not happen.", exp: Type[ValueError] = ValueError
+) -> NoReturn:
+    """
+    Utility function to raise an exception with a message.
+
+    Handy for cases like this:
+
+    ```
+    return (
+        1 if variable == X
+        else 2 if variable == Y
+        else 3 if variable == Z
+        else should_not_happen(f"Variable {variable} is uknown.")
+    )
+    ```
+
+    To prevent silent bugs with useful error message.
+    """
+    raise exp(msg)
+
+
+def check_not_none(
+    value: Optional[T],
+    msg: str = "Value shouldn't be None.",
+    exp: Type[ValueError] = ValueError,
+) -> T:
+    """
+    Utility to remove optionality from a variable.
+
+    Useful for cases like this:
+
+    ```
+    keys = pma.utils.get_keys()
+    pma.omen.omen_buy_outcome_tx(
+        from_addres=check_not_none(keys.bet_from_address),  # <-- No more Optional[HexAddress], so type checker will be happy.
+        ...,
+    )
+    ```
+    """
+    if value is None:
+        should_not_happen(msg=msg, exp=exp)
+    return value