Refactor Safe Transaction API client

gnosis/safe/api/transaction_service_api/__init__.py

@@ -0,0 +1,5 @@
+from .transaction_service_api import TransactionServiceApi
+
+__all__ = [
+    "TransactionServiceApi",
+]

gnosis/safe/api/transaction_service_api.py → gnosis/safe/api/transaction_service_api/transaction_service_api.py

@@ -1,5 +1,4 @@
 import logging
-import time
 from typing import Any, Dict, List, Optional, Tuple, Union
 
 from eth_account.signers.local import LocalAccount
@@ -10,7 +9,9 @@
 from gnosis.eth import EthereumNetwork
 from gnosis.safe import SafeTx
 
-from .base_api import SafeAPIException, SafeBaseAPI
+from ..base_api import SafeAPIException, SafeBaseAPI
+from .transaction_service_messages import get_delegate_message
+from .transaction_service_tx import TransactionServiceTx
 
 logger = logging.getLogger(__name__)
 
@@ -36,9 +37,7 @@ class TransactionServiceApi(SafeBaseAPI):
 
     @classmethod
     def create_delegate_message_hash(cls, delegate_address: ChecksumAddress) -> str:
-        totp = int(time.time()) // 3600
-        hash_to_sign = Web3.keccak(text=delegate_address + str(totp))
-        return hash_to_sign
+        return Web3.keccak(text=get_delegate_message(delegate_address))
 
     @classmethod
     def data_decoded_to_text(cls, data_decoded: Dict[str, Any]) -> Optional[str]:
@@ -115,7 +114,7 @@ def get_balances(self, safe_address: str) -> List[Dict[str, Any]]:
 
     def get_safe_transaction(
         self, safe_tx_hash: Union[bytes, HexStr]
-    ) -> Tuple[SafeTx, Optional[HexBytes]]:
+    ) -> Tuple[TransactionServiceTx, Optional[HexBytes]]:
         """
         :param safe_tx_hash:
         :return: SafeTx and `tx-hash` if transaction was executed
@@ -133,7 +132,8 @@ def get_safe_transaction(
             logger.warning(
                 "EthereumClient should be defined to get a executable SafeTx"
             )
-        safe_tx = SafeTx(
+        safe_tx = TransactionServiceTx(
+            result["proposer"],
             self.ethereum_client,
             result["safe"],
             result["to"],
@@ -276,6 +276,21 @@ def post_transaction(self, safe_tx: SafeTx) -> bool:
             raise SafeAPIException(f"Error posting transaction: {response.content}")
         return True
 
+    def delete_transaction(self, safe_tx_hash: str, signature: str) -> bool:
+        """
+
+        :param safe_tx_hash: hash of eip712 see in transaction_service_messages.py generate_remove_transaction_message function
+        :param signature: signature of safe_tx_hash by transaction proposer
+        :return:
+        """
+        payload = {"safeTxHash": safe_tx_hash, "signature": signature}
+        response = self._delete_request(
+            f"/api/v1/multisig-transactions/{safe_tx_hash}/", payload
+        )
+        if not response.ok:
+            raise SafeAPIException(f"Error deleting transaction: {response.content}")
+        return True
+
     def post_message(
         self,
         safe_address: ChecksumAddress,

gnosis/safe/api/transaction_service_api/transaction_service_messages.py

@@ -0,0 +1,63 @@
+import time
+from typing import Dict
+
+from eth_typing import ChecksumAddress
+
+
+def get_totp() -> int:
+    """
+
+    :return: time-based one-time password
+    """
+    return int(time.time()) // 3600
+
+
+def get_delegate_message(cls, delegate_address: ChecksumAddress) -> str:
+    """
+    Retrieves the required message for creating or removing a delegate on Safe Transaction Service.
+
+    :param cls:
+    :param delegate_address:
+    :return: generated str message
+    """
+    totp = get_totp()
+    return delegate_address + str(totp)
+
+
+def get_remove_transaction_message(
+    safe_address: ChecksumAddress, safe_tx_hash: bytes, chain_id: int
+) -> Dict:
+    """
+    Retrieves the required message for removing a not executed transaction on Safe Transaction Service.
+
+    :param safe_address:
+    :param safe_tx_hash:
+    :param chain_id:
+    :return: generated EIP712 message
+    """
+    remove_transaction_message = {
+        "types": {
+            "EIP712Domain": [
+                {"name": "name", "type": "string"},
+                {"name": "version", "type": "string"},
+                {"name": "chainId", "type": "uint256"},
+                {"name": "verifyingContract", "type": "address"},
+            ],
+            "DeleteRequest": [
+                {"name": "safeTxHash", "type": "bytes32"},
+                {"name": "totp", "type": "uint256"},
+            ],
+        },
+        "primaryType": "DeleteRequest",
+        "domain": {
+            "name": "Safe Transaction Service",
+            "version": "1.0",
+            "chainId": chain_id,
+            "verifyingContract": safe_address,
+        },
+        "message": {
+            "safeTxHash": safe_tx_hash,
+            "totp": get_totp(),
+        },
+    }
+    return remove_transaction_message

gnosis/safe/api/transaction_service_api/transaction_service_tx.py

@@ -0,0 +1,9 @@
+from eth_typing import ChecksumAddress
+
+from gnosis.safe import SafeTx
+
+
+class TransactionServiceTx(SafeTx):
+    def __init__(self, proposer: ChecksumAddress, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.proposer = proposer

gnosis/safe/tests/api/test_transaction_service_api.py

@@ -4,8 +4,7 @@
 
 from gnosis.eth import EthereumClient, EthereumNetwork, EthereumNetworkNotSupported
 from gnosis.eth.tests.ethereum_test_case import EthereumTestCaseMixin
-
-from ...api.transaction_service_api import TransactionServiceApi
+from gnosis.safe.api.transaction_service_api import TransactionServiceApi
 
 
 class TestTransactionServiceAPI(EthereumTestCaseMixin, TestCase):