Add classes to interact with Safe APIs (#292)

- Support Tx Service
- Support Relay Service (deprecated)
- Closes #284

gnosis/safe/api/__init__.py

@@ -0,0 +1,12 @@
+# flake8: noqa F401
+from .base_api import SafeAPIException
+from .relay_service_api import RelayEstimation, RelaySentTransaction, RelayServiceApi
+from .transaction_service_api import TransactionServiceApi
+
+__all__ = [
+    "SafeAPIException",
+    "RelayServiceApi",
+    "RelayEstimation",
+    "RelaySentTransaction",
+    "TransactionServiceApi",
+]

gnosis/safe/api/base_api.py

@@ -0,0 +1,58 @@
+from abc import ABC
+from typing import Dict, Optional
+from urllib.parse import urljoin
+
+import requests
+
+from gnosis.eth.ethereum_client import (
+    EthereumClient,
+    EthereumNetwork,
+    EthereumNetworkNotSupported,
+)
+
+
+class SafeAPIException(Exception):
+    pass
+
+
+class SafeBaseAPI(ABC):
+    URL_BY_NETWORK: Dict[EthereumNetwork, str] = {}
+
+    def __init__(
+        self,
+        network: EthereumNetwork,
+        ethereum_client: Optional[EthereumClient] = None,
+        base_url: Optional[str] = None,
+    ):
+        """
+        :param network: Network for the transaction service
+        :param ethereum_client:
+        :param base_url: If a custom transaction service is used
+        :raises: EthereumNetworkNotSupported
+        """
+        self.network = network
+        self.ethereum_client = ethereum_client
+        self.base_url = base_url or self.URL_BY_NETWORK.get(network)
+        if not self.base_url:
+            raise EthereumNetworkNotSupported(network)
+
+    @classmethod
+    def from_ethereum_client(cls, ethereum_client: EthereumClient) -> "SafeBaseAPI":
+        ethereum_network = ethereum_client.get_network()
+        return cls(ethereum_network, ethereum_client=ethereum_client)
+
+    def _get_request(self, url: str) -> requests.Response:
+        full_url = urljoin(self.base_url, url)
+        return requests.get(full_url)
+
+    def _post_request(self, url: str, payload: Dict) -> requests.Response:
+        full_url = urljoin(self.base_url, url)
+        return requests.post(
+            full_url, json=payload, headers={"Content-type": "application/json"}
+        )
+
+    def _delete_request(self, url: str, payload: Dict) -> requests.Response:
+        full_url = urljoin(self.base_url, url)
+        return requests.delete(
+            full_url, json=payload, headers={"Content-type": "application/json"}
+        )

gnosis/safe/api/relay_service_api.py

@@ -0,0 +1,97 @@
+from urllib.parse import urljoin
+
+import requests
+from eth_typing import ChecksumAddress, HexStr
+
+from gnosis.eth.ethereum_client import EthereumNetwork
+
+from .. import SafeTx
+from ..signatures import signature_split
+from .base_api import SafeAPIException, SafeBaseAPI
+
+try:
+    from typing import TypedDict
+except ImportError:
+    from typing_extensions import TypedDict
+
+
+class RelayEstimation(TypedDict):
+    safeTxGas: int
+    baseGas: int
+    gasPrice: int
+    lastUsedNonce: int
+    gasToken: ChecksumAddress
+    refundReceiver: ChecksumAddress
+
+
+class RelaySentTransaction(TypedDict):
+    safeTxHash: HexStr
+    txHash: HexStr
+
+
+class RelayServiceApi(SafeBaseAPI):
+    URL_BY_NETWORK = {
+        EthereumNetwork.GOERLI: "https://safe-relay.goerli.gnosis.io/",
+        EthereumNetwork.MAINNET: "https://safe-relay.gnosis.io",
+        EthereumNetwork.RINKEBY: "https://safe-relay.rinkeby.gnosis.io",
+    }
+
+    def send_transaction(
+        self, safe_address: str, safe_tx: SafeTx
+    ) -> RelaySentTransaction:
+        url = urljoin(self.base_url, f"/api/v1/safes/{safe_address}/transactions/")
+        signatures = []
+        for i in range(len(safe_tx.signatures) // 65):
+            v, r, s = signature_split(safe_tx.signatures, i)
+            signatures.append(
+                {
+                    "v": v,
+                    "r": r,
+                    "s": s,
+                }
+            )
+
+        data = {
+            "to": safe_tx.to,
+            "value": safe_tx.value,
+            "data": safe_tx.data.hex() if safe_tx.data else None,
+            "operation": safe_tx.operation,
+            "gasToken": safe_tx.gas_token,
+            "safeTxGas": safe_tx.safe_tx_gas,
+            "dataGas": safe_tx.base_gas,
+            "gasPrice": safe_tx.gas_price,
+            "refundReceiver": safe_tx.refund_receiver,
+            "nonce": safe_tx.safe_nonce,
+            "signatures": signatures,
+        }
+        response = requests.post(url, json=data)
+        if not response.ok:
+            raise SafeAPIException(f"Error posting transaction: {response.content}")
+        else:
+            return RelaySentTransaction(response.json())
+
+    def get_estimation(self, safe_address: str, safe_tx: SafeTx) -> RelayEstimation:
+        """
+        :param safe_address:
+        :param safe_tx:
+        :return: RelayEstimation
+        """
+        url = urljoin(
+            self.base_url, f"/api/v2/safes/{safe_address}/transactions/estimate/"
+        )
+        data = {
+            "to": safe_tx.to,
+            "value": safe_tx.value,
+            "data": safe_tx.data.hex() if safe_tx.data else None,
+            "operation": safe_tx.operation,
+            "gasToken": safe_tx.gas_token,
+        }
+        response = requests.post(url, json=data)
+        if not response.ok:
+            raise SafeAPIException(f"Error posting transaction: {response.content}")
+        else:
+            response_json = response.json()
+            # Convert values to int
+            for key in ("safeTxGas", "baseGas", "gasPrice"):
+                response_json[key] = int(response_json[key])
+            return RelayEstimation(response_json)