18643806651

Committed 20 Oct 2025 06:11AM UTC coverage: 89.969% (-4.0%) from 93.927%

Build # 18643806651

Build Type

Pull #2056

github

Committed by

web-flow

Commit Message

Merge 77f5b46bd into 8eac6367d

Pull Request Pull Request #2056: Bump coverage from 7.10.6 to 7.11.0

Run Details

9229 of 10258 relevant lines covered (89.97%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

27.33

/safe_eth/safe/api/transaction_service_api/transaction_service_api.py

import logging
import os
from typing import Any, Dict, List, Optional, Tuple, Union
from urllib.parse import urlencode

from eth_typing import ChecksumAddress, Hash32, HexStr
from eth_utils import to_checksum_address
from hexbytes import HexBytes

from safe_eth.eth import EthereumClient, EthereumNetwork
from safe_eth.eth.eip712 import eip712_encode_hash
from safe_eth.safe import SafeTx
from safe_eth.util.util import to_0x_hex_str

from ..base_api import SafeAPIException, SafeBaseAPI
from .entities import Balance, DataDecoded, DelegateUser, Message, Transaction
from .transaction_service_messages import get_delegate_message
from .transaction_service_tx import TransactionServiceTx

logger = logging.getLogger(__name__)


class ApiSafeTxHashNotMatchingException(SafeAPIException):
    pass


class TransactionServiceApi(SafeBaseAPI):
    NETWORK_SHORTNAME = {
        EthereumNetwork.ARBITRUM_ONE: "arb1",
        EthereumNetwork.AURORA_MAINNET: "aurora",
        EthereumNetwork.AVALANCHE_C_CHAIN: "avax",
        EthereumNetwork.BASE: "base",
        EthereumNetwork.BASE_SEPOLIA_TESTNET: "basesep",
        EthereumNetwork.BERACHAIN: "berachain",
        EthereumNetwork.BLAST: "blastmainnet",
        EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "bnb",
        EthereumNetwork.CELO_MAINNET: "celo",
        EthereumNetwork.GNOSIS: "gno",
        EthereumNetwork.GNOSIS_CHIADO_TESTNET: "chi",
        EthereumNetwork.HEMI_NETWORK: "hemi",
        EthereumNetwork.INK: "ink",
        EthereumNetwork.KATANA_MAINNET: "katana",
        EthereumNetwork.LENS: "lens",
        EthereumNetwork.LINEA: "linea",
        EthereumNetwork.MAINNET: "eth",
        EthereumNetwork.MANTLE: "mantle",
        EthereumNetwork.OPTIMISM: "oeth",
        EthereumNetwork.POLYGON: "pol",
        EthereumNetwork.POLYGON_ZKEVM: "zkevm",
        EthereumNetwork.SCROLL: "scr",
        EthereumNetwork.SEPOLIA: "sep",
        EthereumNetwork.SONIC_MAINNET: "sonic",
        EthereumNetwork.UNICHAIN: "unichain",
        EthereumNetwork.WORLD_CHAIN: "wc",
        EthereumNetwork.X_LAYER_MAINNET: "okb",
        EthereumNetwork.ZKSYNC_MAINNET: "zksync",
    }
    TRANSACTION_SERVICE_BASE_URL = "https://api.safe.global/tx-service"

    def __init__(
        self,
        network: EthereumNetwork,
        ethereum_client: Optional[EthereumClient] = None,
        base_url: Optional[str] = None,
        api_key: Optional[str] = os.environ.get("SAFE_TRANSACTION_SERVICE_API_KEY"),
        request_timeout: int = int(
            os.environ.get("SAFE_TRANSACTION_SERVICE_REQUEST_TIMEOUT", 10)
        ),
    ):
        super().__init__(network, ethereum_client, base_url, api_key, request_timeout)

    def _get_url_by_network(self, network: EthereumNetwork) -> Optional[str]:
        network_short_name = self.NETWORK_SHORTNAME.get(network)
        if not network_short_name:
            return None
        return f"{self.TRANSACTION_SERVICE_BASE_URL}/{network_short_name}"

    @classmethod
    def data_decoded_to_text(cls, data_decoded: Dict[str, Any]) -> Optional[str]:
        """
        Decoded data decoded to text
        :param data_decoded:
        :return:
        """
        if not data_decoded:
            return None

        method = data_decoded["method"]
        parameters = data_decoded.get("parameters", [])
        text = ""
        for (
            parameter
        ) in parameters:  # Multisend or executeTransaction from another Safe
            if "decodedValue" in parameter:
                text += (
                    method
                    + ":\n - "
                    + "\n - ".join(
                        [
                            decoded_text
                            for decoded_value in parameter.get("decodedValue", {})
                            if (
                                decoded_text := cls.data_decoded_to_text(
                                    decoded_value.get("decodedData", {})
                                )
                            )
                            is not None
                        ]
                    )
                    + "\n"
                )
        if text:
            return text.strip()
        else:
            return (
                method
                + ": "
                + ",".join([str(parameter["value"]) for parameter in parameters])
            )

    @classmethod
    def parse_signatures(cls, raw_tx: Transaction) -> Optional[bytes]:
        """
        Parse signatures in `confirmations` list to build a valid signature (owners must be sorted lexicographically)

        :param raw_tx:
        :return: Valid signature with signatures sorted lexicographically
        """
        if raw_tx["signatures"]:
            # Tx was executed and signatures field is populated
            return HexBytes(raw_tx["signatures"])
        elif raw_tx["confirmations"]:
            # Parse offchain transactions
            return b"".join(
                [
                    HexBytes(confirmation["signature"])
                    for confirmation in sorted(
                        raw_tx["confirmations"], key=lambda x: int(x["owner"], 16)
                    )
                    if confirmation["signatureType"] == "EOA"
                ]
            )
        return None

    def create_delegate_message_hash(self, delegate_address: ChecksumAddress) -> Hash32:
        return eip712_encode_hash(
            get_delegate_message(delegate_address, self.network.value)
        )

    def _build_transaction_service_tx(
        self, safe_tx_hash: Union[bytes, HexStr], tx_raw: Transaction
    ) -> TransactionServiceTx:
        signatures = self.parse_signatures(tx_raw)
        safe_tx = TransactionServiceTx(
            to_checksum_address(tx_raw["proposer"]) if tx_raw["proposer"] else None,
            self.ethereum_client,
            tx_raw["safe"],
            tx_raw["to"],
            int(tx_raw["value"]),
            HexBytes(tx_raw["data"]) if tx_raw["data"] else b"",
            int(tx_raw["operation"]),
            int(tx_raw["safeTxGas"]),
            int(tx_raw["baseGas"]),
            int(tx_raw["gasPrice"]),
            tx_raw["gasToken"],
            tx_raw["refundReceiver"],
            signatures=signatures if signatures else b"",
            safe_nonce=int(tx_raw["nonce"]),
            chain_id=self.network.value,
        )
        safe_tx.tx_hash = (
            HexBytes(tx_raw["transactionHash"]) if tx_raw["transactionHash"] else None
        )

        if safe_tx.safe_tx_hash != HexBytes(safe_tx_hash):
            raise ApiSafeTxHashNotMatchingException(
                f"API safe-tx-hash: {to_0x_hex_str(HexBytes(safe_tx_hash))} doesn't match the calculated safe-tx-hash: {to_0x_hex_str(HexBytes(safe_tx.safe_tx_hash))}"
            )

        return safe_tx

    def get_balances(self, safe_address: ChecksumAddress) -> List[Balance]:
        """

        :param safe_address:
        :return: a list of balances for provided Safe
        """
        response = self._get_request(f"/api/v1/safes/{safe_address}/balances/")
        if not response.ok:
            raise SafeAPIException(f"Cannot get balances: {response.content!r}")
        return response.json()

    def get_safe_transaction(
        self, safe_tx_hash: Union[bytes, HexStr]
    ) -> Tuple[TransactionServiceTx, Optional[HexBytes]]:
        """
        :param safe_tx_hash:
        :return: SafeTx and `tx-hash` if transaction was executed
        """
        safe_tx_hash_str = to_0x_hex_str(HexBytes(safe_tx_hash))
        response = self._get_request(
            f"/api/v2/multisig-transactions/{safe_tx_hash_str}/"
        )
        if not response.ok:
            raise SafeAPIException(
                f"Cannot get transaction with safe-tx-hash={safe_tx_hash_str}: {response.content!r}"
            )

        if not self.ethereum_client:
            logger.warning(
                "EthereumClient should be defined to get a executable SafeTx"
            )

        result = response.json()
        safe_tx = self._build_transaction_service_tx(safe_tx_hash, result)

        return safe_tx, HexBytes(safe_tx.tx_hash) if safe_tx.tx_hash else None

    def get_transactions(
        self, safe_address: ChecksumAddress, **kwargs: Dict[str, Union[str, int, bool]]
    ) -> List[Transaction]:
        """

        :param safe_address:
        :return: a list of transactions for provided Safe
        """
        url = f"/api/v2/safes/{safe_address}/multisig-transactions/"

        if kwargs:
            query_string = urlencode(
                {key: str(value) for key, value in kwargs.items() if value is not None}
            )
            url += "?" + query_string

        response = self._get_request(url)
        if not response.ok:
            raise SafeAPIException(f"Cannot get transactions: {response.content!r}")

        transactions = response.json().get("results", [])

        if safe_tx_hash_arg := kwargs.get("safe_tx_hash", None):
            # Validation that the calculated safe_tx_hash is the same as the safe_tx_hash provided for filter.
            safe_tx_hash = HexBytes(str(safe_tx_hash_arg))
            [
                self._build_transaction_service_tx(safe_tx_hash, tx)
                for tx in transactions
            ]

        return transactions

    def get_delegates(self, safe_address: ChecksumAddress) -> List[DelegateUser]:
        """

        :param safe_address:
        :return: a list of delegates for provided Safe
        """
        response = self._get_request(f"/api/v2/delegates/?safe={safe_address}")
        if not response.ok:
            raise SafeAPIException(f"Cannot get delegates: {response.content!r}")
        return response.json().get("results", [])

    def get_safes_for_owner(
        self, owner_address: ChecksumAddress
    ) -> List[ChecksumAddress]:
        """

        :param owner_address:
        :return: a List of Safe addresses which the owner_address is an owner
        """
        response = self._get_request(f"/api/v1/owners/{owner_address}/safes/")
        if not response.ok:
            raise SafeAPIException(f"Cannot get delegates: {response.content!r}")
        return response.json().get("safes", [])

    def post_signatures(self, safe_tx_hash: bytes, signatures: bytes) -> bool:
        """
        Create a new confirmation with provided signature for the given safe_tx_hash
        :param safe_tx_hash:
        :param signatures:
        :return: True if new confirmation was created
        """
        safe_tx_hash_str = to_0x_hex_str(safe_tx_hash)
        response = self._post_request(
            f"/api/v1/multisig-transactions/{safe_tx_hash_str}/confirmations/",
            payload={"signature": to_0x_hex_str(signatures)},
        )
        if not response.ok:
            raise SafeAPIException(
                f"Cannot post signatures for tx with safe-tx-hash={safe_tx_hash_str}: {response.content!r}"
            )
        return True

    def add_delegate(
        self,
        delegate_address: ChecksumAddress,
        delegator_address: ChecksumAddress,
        label: str,
        signature: bytes,
        safe_address: Optional[ChecksumAddress] = None,
    ) -> bool:
        add_payload = {
            "delegate": delegate_address,
            "delegator": delegator_address,
            "signature": to_0x_hex_str(HexBytes(signature)),
            "label": label,
        }
        if safe_address:
            add_payload["safe"] = safe_address
        response = self._post_request("/api/v2/delegates/", add_payload)
        if not response.ok:
            raise SafeAPIException(f"Cannot add delegate: {response.content!r}")
        return True

    def remove_delegate(
        self,
        delegate_address: ChecksumAddress,
        delegator_address: ChecksumAddress,
        signature: bytes,
        safe_address: Optional[ChecksumAddress] = None,
    ) -> bool:
        """
        Deletes a delegated user

        :param delegator_address:
        :param delegate_address:
        :param signature: Signature of a hash of an eip712 message.
        :param safe_address: If specified, a delegate is removed for a delegator for the specific safe.
            Otherwise, the delegate is deleted in a global form.
        :return:
        """
        remove_payload = {
            "delegator": delegator_address,
            "signature": to_0x_hex_str(HexBytes(signature)),
        }
        if safe_address:
            remove_payload["safe"] = safe_address
        response = self._delete_request(
            f"/api/v2/delegates/{delegate_address}/",
            remove_payload,
        )
        if not response.ok:
            raise SafeAPIException(f"Cannot remove delegate: {response.content!r}")
        return True

    def post_transaction(self, safe_tx: SafeTx) -> bool:
        random_sender = "0x0000000000000000000000000000000000000002"
        sender = safe_tx.sorted_signers[0] if safe_tx.sorted_signers else random_sender
        data = {
            "to": safe_tx.to,
            "value": safe_tx.value,
            "data": to_0x_hex_str(safe_tx.data) if safe_tx.data else None,
            "operation": safe_tx.operation,
            "gasToken": safe_tx.gas_token,
            "safeTxGas": safe_tx.safe_tx_gas,
            "baseGas": safe_tx.base_gas,
            "gasPrice": safe_tx.gas_price,
            "refundReceiver": safe_tx.refund_receiver,
            "nonce": safe_tx.safe_nonce,
            "contractTransactionHash": to_0x_hex_str(safe_tx.safe_tx_hash),
            "sender": sender,
            "signature": safe_tx.signatures.hex() if safe_tx.signatures else None,
            "origin": "Safe-CLI",
        }
        response = self._post_request(
            f"/api/v2/safes/{safe_tx.safe_address}/multisig-transactions/", data
        )
        if not response.ok:
            raise SafeAPIException(f"Error posting transaction: {response.content!r}")
        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/v2/multisig-transactions/{safe_tx_hash}/", payload
        )
        if not response.ok:
            raise SafeAPIException(f"Error deleting transaction: {response.content!r}")
        return True

    def post_message(
        self,
        safe_address: ChecksumAddress,
        message: Union[str, Dict],
        signature: bytes,
        safe_app_id: Optional[int] = 0,
    ) -> bool:
        """
        Create safe message on transaction service for provided Safe address

        :param safe_address:
        :param message: If str it will be encoded using EIP191, and if it's a dictionary it will be encoded using EIP712
        :param signature:
        :return:
        """
        payload = {
            "message": message,
            "safeAppId": safe_app_id,
            "signature": to_0x_hex_str(signature),
        }
        response = self._post_request(
            f"/api/v1/safes/{safe_address}/messages/", payload
        )
        if not response.ok:
            raise SafeAPIException(f"Error posting message: {response.content!r}")
        return True

    def get_message(self, safe_message_hash: bytes) -> Message:
        """

        :param safe_message_hash:
        :return: Safe message for provided Safe message hash
        """
        response = self._get_request(
            f"/api/v1/messages/{to_0x_hex_str(safe_message_hash)}/"
        )
        if not response.ok:
            raise SafeAPIException(f"Cannot get messages: {response.content!r}")
        return response.json()

    def get_messages(self, safe_address: ChecksumAddress) -> List[Message]:
        """

        :param safe_address:
        :return: list of messages for provided Safe address
        """
        response = self._get_request(f"/api/v1/safes/{safe_address}/messages/")
        if not response.ok:
            raise SafeAPIException(f"Cannot get messages: {response.content!r}")
        return response.json().get("results", [])

    def post_message_signature(
        self, safe_message_hash: bytes, signature: bytes
    ) -> bool:
        """
        Add a new confirmation for provided Safe message hash

        :param safe_message_hash:
        :param signature:
        :return:
        """
        payload = {"signature": to_0x_hex_str(signature)}
        response = self._post_request(
            f"/api/v1/messages/{to_0x_hex_str(safe_message_hash)}/signatures/",
            payload,
        )
        if not response.ok:
            raise SafeAPIException(
                f"Error posting message signature: {response.content!r}"
            )
        return True

    def decode_data(
        self, data: Union[bytes, HexStr], to_address: Optional[ChecksumAddress] = None
    ) -> DataDecoded:
        """
        Retrieve decoded information using tx service internal ABI information given the tx data.

        :param data: tx data as a 0x prefixed hexadecimal string.
        :param to_address: address of the contract. This will be used in case of more than one function identifiers matching.
        :return:
        """
        payload = {"data": to_0x_hex_str(HexBytes(data))}
        if to_address:
            payload["to"] = to_address
        response = self._post_request("/api/v1/data-decoder/", payload)
        if not response.ok:
            raise SafeAPIException(f"Cannot decode tx data: {response.content!r}")
        return response.json()

1	import logging	1✔
2	import os	1✔
3	from typing import Any, Dict, List, Optional, Tuple, Union	1✔
4	from urllib.parse import urlencode	1✔
5
6	from eth_typing import ChecksumAddress, Hash32, HexStr	1✔
7	from eth_utils import to_checksum_address	1✔
8	from hexbytes import HexBytes	1✔
9
10	from safe_eth.eth import EthereumClient, EthereumNetwork	1✔
11	from safe_eth.eth.eip712 import eip712_encode_hash	1✔
12	from safe_eth.safe import SafeTx	1✔
13	from safe_eth.util.util import to_0x_hex_str	1✔
14
15	from ..base_api import SafeAPIException, SafeBaseAPI	1✔
16	from .entities import Balance, DataDecoded, DelegateUser, Message, Transaction	1✔
17	from .transaction_service_messages import get_delegate_message	1✔
18	from .transaction_service_tx import TransactionServiceTx	1✔
19
20	logger = logging.getLogger(__name__)	1✔
21
22
23	class ApiSafeTxHashNotMatchingException(SafeAPIException):	1✔
24	pass	1✔
25
26
27	class TransactionServiceApi(SafeBaseAPI):	1✔
28	NETWORK_SHORTNAME = {	1✔
29	EthereumNetwork.ARBITRUM_ONE: "arb1",
30	EthereumNetwork.AURORA_MAINNET: "aurora",
31	EthereumNetwork.AVALANCHE_C_CHAIN: "avax",
32	EthereumNetwork.BASE: "base",
33	EthereumNetwork.BASE_SEPOLIA_TESTNET: "basesep",
34	EthereumNetwork.BERACHAIN: "berachain",
35	EthereumNetwork.BLAST: "blastmainnet",
36	EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "bnb",
37	EthereumNetwork.CELO_MAINNET: "celo",
38	EthereumNetwork.GNOSIS: "gno",
39	EthereumNetwork.GNOSIS_CHIADO_TESTNET: "chi",
40	EthereumNetwork.HEMI_NETWORK: "hemi",
41	EthereumNetwork.INK: "ink",
42	EthereumNetwork.KATANA_MAINNET: "katana",
43	EthereumNetwork.LENS: "lens",
44	EthereumNetwork.LINEA: "linea",
45	EthereumNetwork.MAINNET: "eth",
46	EthereumNetwork.MANTLE: "mantle",
47	EthereumNetwork.OPTIMISM: "oeth",
48	EthereumNetwork.POLYGON: "pol",
49	EthereumNetwork.POLYGON_ZKEVM: "zkevm",
50	EthereumNetwork.SCROLL: "scr",
51	EthereumNetwork.SEPOLIA: "sep",
52	EthereumNetwork.SONIC_MAINNET: "sonic",
53	EthereumNetwork.UNICHAIN: "unichain",
54	EthereumNetwork.WORLD_CHAIN: "wc",
55	EthereumNetwork.X_LAYER_MAINNET: "okb",
56	EthereumNetwork.ZKSYNC_MAINNET: "zksync",
57	}
58	TRANSACTION_SERVICE_BASE_URL = "https://api.safe.global/tx-service"	1✔
59
60	def __init__(	1✔
61	self,
62	network: EthereumNetwork,
63	ethereum_client: Optional[EthereumClient] = None,
64	base_url: Optional[str] = None,
65	api_key: Optional[str] = os.environ.get("SAFE_TRANSACTION_SERVICE_API_KEY"),
66	request_timeout: int = int(
67	os.environ.get("SAFE_TRANSACTION_SERVICE_REQUEST_TIMEOUT", 10)
68	),
69	):
70	super().__init__(network, ethereum_client, base_url, api_key, request_timeout)	×
71
72	def _get_url_by_network(self, network: EthereumNetwork) -> Optional[str]:	1✔
73	network_short_name = self.NETWORK_SHORTNAME.get(network)	×
74	if not network_short_name:	×
75	return None	×
76	return f"{self.TRANSACTION_SERVICE_BASE_URL}/{network_short_name}"	×
77
78	@classmethod	1✔
79	def data_decoded_to_text(cls, data_decoded: Dict[str, Any]) -> Optional[str]:	1✔
80	"""
81	Decoded data decoded to text
82	:param data_decoded:
83	:return:
84	"""
85	if not data_decoded:	×
86	return None	×
87
88	method = data_decoded["method"]	×
89	parameters = data_decoded.get("parameters", [])	×
90	text = ""	×
91	for (	×
92	parameter
93	) in parameters: # Multisend or executeTransaction from another Safe
94	if "decodedValue" in parameter:	×
95	text += (	×
96	method
97	+ ":\n - "
98	+ "\n - ".join(
99	[
100	decoded_text
101	for decoded_value in parameter.get("decodedValue", {})
102	if (
103	decoded_text := cls.data_decoded_to_text(
104	decoded_value.get("decodedData", {})
105	)
106	)
107	is not None
108	]
109	)
110	+ "\n"
111	)
112	if text:	×
113	return text.strip()	×
114	else:
115	return (	×
116	method
117	+ ": "
118	+ ",".join([str(parameter["value"]) for parameter in parameters])
119	)
120
121	@classmethod	1✔
122	def parse_signatures(cls, raw_tx: Transaction) -> Optional[bytes]:	1✔
123	"""
124	Parse signatures in `confirmations` list to build a valid signature (owners must be sorted lexicographically)
125
126	:param raw_tx:
127	:return: Valid signature with signatures sorted lexicographically
128	"""
129	if raw_tx["signatures"]:	×
130	# Tx was executed and signatures field is populated
131	return HexBytes(raw_tx["signatures"])	×
132	elif raw_tx["confirmations"]:	×
133	# Parse offchain transactions
134	return b"".join(	×
135	[
136	HexBytes(confirmation["signature"])
137	for confirmation in sorted(
138	raw_tx["confirmations"], key=lambda x: int(x["owner"], 16)
139	)
140	if confirmation["signatureType"] == "EOA"
141	]
142	)
143	return None	×
144
145	def create_delegate_message_hash(self, delegate_address: ChecksumAddress) -> Hash32:	1✔
146	return eip712_encode_hash(	×
147	get_delegate_message(delegate_address, self.network.value)
148	)
149
150	def _build_transaction_service_tx(	1✔
151	self, safe_tx_hash: Union[bytes, HexStr], tx_raw: Transaction
152	) -> TransactionServiceTx:
153	signatures = self.parse_signatures(tx_raw)	×
154	safe_tx = TransactionServiceTx(	×
155	to_checksum_address(tx_raw["proposer"]) if tx_raw["proposer"] else None,
156	self.ethereum_client,
157	tx_raw["safe"],
158	tx_raw["to"],
159	int(tx_raw["value"]),
160	HexBytes(tx_raw["data"]) if tx_raw["data"] else b"",
161	int(tx_raw["operation"]),
162	int(tx_raw["safeTxGas"]),
163	int(tx_raw["baseGas"]),
164	int(tx_raw["gasPrice"]),
165	tx_raw["gasToken"],
166	tx_raw["refundReceiver"],
167	signatures=signatures if signatures else b"",
168	safe_nonce=int(tx_raw["nonce"]),
169	chain_id=self.network.value,
170	)
171	safe_tx.tx_hash = (	×
172	HexBytes(tx_raw["transactionHash"]) if tx_raw["transactionHash"] else None
173	)
174
175	if safe_tx.safe_tx_hash != HexBytes(safe_tx_hash):	×
176	raise ApiSafeTxHashNotMatchingException(	×
177	f"API safe-tx-hash: {to_0x_hex_str(HexBytes(safe_tx_hash))} doesn't match the calculated safe-tx-hash: {to_0x_hex_str(HexBytes(safe_tx.safe_tx_hash))}"
178	)
179
180	return safe_tx	×
181
182	def get_balances(self, safe_address: ChecksumAddress) -> List[Balance]:	1✔
183	"""
184
185	:param safe_address:
186	:return: a list of balances for provided Safe
187	"""
188	response = self._get_request(f"/api/v1/safes/{safe_address}/balances/")	×
189	if not response.ok:	×
190	raise SafeAPIException(f"Cannot get balances: {response.content!r}")	×
191	return response.json()	×
192
193	def get_safe_transaction(	1✔
194	self, safe_tx_hash: Union[bytes, HexStr]
195	) -> Tuple[TransactionServiceTx, Optional[HexBytes]]:
196	"""
197	:param safe_tx_hash:
198	:return: SafeTx and `tx-hash` if transaction was executed
199	"""
200	safe_tx_hash_str = to_0x_hex_str(HexBytes(safe_tx_hash))	×
201	response = self._get_request(	×
202	f"/api/v2/multisig-transactions/{safe_tx_hash_str}/"
203	)
204	if not response.ok:	×
205	raise SafeAPIException(	×
206	f"Cannot get transaction with safe-tx-hash={safe_tx_hash_str}: {response.content!r}"
207	)
208
209	if not self.ethereum_client:	×
210	logger.warning(	×
211	"EthereumClient should be defined to get a executable SafeTx"
212	)
213
214	result = response.json()	×
215	safe_tx = self._build_transaction_service_tx(safe_tx_hash, result)	×
216
217	return safe_tx, HexBytes(safe_tx.tx_hash) if safe_tx.tx_hash else None	×
218
219	def get_transactions(	1✔
220	self, safe_address: ChecksumAddress, **kwargs: Dict[str, Union[str, int, bool]]
221	) -> List[Transaction]:
222	"""
223
224	:param safe_address:
225	:return: a list of transactions for provided Safe
226	"""
227	url = f"/api/v2/safes/{safe_address}/multisig-transactions/"	×
228
229	if kwargs:	×
230	query_string = urlencode(	×
231	{key: str(value) for key, value in kwargs.items() if value is not None}
232	)
233	url += "?" + query_string	×
234
235	response = self._get_request(url)	×
236	if not response.ok:	×
237	raise SafeAPIException(f"Cannot get transactions: {response.content!r}")	×
238
239	transactions = response.json().get("results", [])	×
240
241	if safe_tx_hash_arg := kwargs.get("safe_tx_hash", None):	×
242	# Validation that the calculated safe_tx_hash is the same as the safe_tx_hash provided for filter.
243	safe_tx_hash = HexBytes(str(safe_tx_hash_arg))	×
244	[	×
245	self._build_transaction_service_tx(safe_tx_hash, tx)
246	for tx in transactions
247	]
248
249	return transactions	×
250
251	def get_delegates(self, safe_address: ChecksumAddress) -> List[DelegateUser]:	1✔
252	"""
253
254	:param safe_address:
255	:return: a list of delegates for provided Safe
256	"""
257	response = self._get_request(f"/api/v2/delegates/?safe={safe_address}")	×
258	if not response.ok:	×
259	raise SafeAPIException(f"Cannot get delegates: {response.content!r}")	×
260	return response.json().get("results", [])	×
261
262	def get_safes_for_owner(	1✔
263	self, owner_address: ChecksumAddress
264	) -> List[ChecksumAddress]:
265	"""
266
267	:param owner_address:
268	:return: a List of Safe addresses which the owner_address is an owner
269	"""
270	response = self._get_request(f"/api/v1/owners/{owner_address}/safes/")	×
271	if not response.ok:	×
272	raise SafeAPIException(f"Cannot get delegates: {response.content!r}")	×
273	return response.json().get("safes", [])	×
274
275	def post_signatures(self, safe_tx_hash: bytes, signatures: bytes) -> bool:	1✔
276	"""
277	Create a new confirmation with provided signature for the given safe_tx_hash
278	:param safe_tx_hash:
279	:param signatures:
280	:return: True if new confirmation was created
281	"""
282	safe_tx_hash_str = to_0x_hex_str(safe_tx_hash)	×
283	response = self._post_request(	×
284	f"/api/v1/multisig-transactions/{safe_tx_hash_str}/confirmations/",
285	payload={"signature": to_0x_hex_str(signatures)},
286	)
287	if not response.ok:	×
288	raise SafeAPIException(	×
289	f"Cannot post signatures for tx with safe-tx-hash={safe_tx_hash_str}: {response.content!r}"
290	)
291	return True	×
292
293	def add_delegate(	1✔
294	self,
295	delegate_address: ChecksumAddress,
296	delegator_address: ChecksumAddress,
297	label: str,
298	signature: bytes,
299	safe_address: Optional[ChecksumAddress] = None,
300	) -> bool:
301	add_payload = {	×
302	"delegate": delegate_address,
303	"delegator": delegator_address,
304	"signature": to_0x_hex_str(HexBytes(signature)),
305	"label": label,
306	}
307	if safe_address:	×
308	add_payload["safe"] = safe_address	×
309	response = self._post_request("/api/v2/delegates/", add_payload)	×
310	if not response.ok:	×
311	raise SafeAPIException(f"Cannot add delegate: {response.content!r}")	×
312	return True	×
313
314	def remove_delegate(	1✔
315	self,
316	delegate_address: ChecksumAddress,
317	delegator_address: ChecksumAddress,
318	signature: bytes,
319	safe_address: Optional[ChecksumAddress] = None,
320	) -> bool:
321	"""
322	Deletes a delegated user
323
324	:param delegator_address:
325	:param delegate_address:
326	:param signature: Signature of a hash of an eip712 message.
327	:param safe_address: If specified, a delegate is removed for a delegator for the specific safe.
328	Otherwise, the delegate is deleted in a global form.
329	:return:
330	"""
331	remove_payload = {	×
332	"delegator": delegator_address,
333	"signature": to_0x_hex_str(HexBytes(signature)),
334	}
335	if safe_address:	×
336	remove_payload["safe"] = safe_address	×
337	response = self._delete_request(	×
338	f"/api/v2/delegates/{delegate_address}/",
339	remove_payload,
340	)
341	if not response.ok:	×
342	raise SafeAPIException(f"Cannot remove delegate: {response.content!r}")	×
343	return True	×
344
345	def post_transaction(self, safe_tx: SafeTx) -> bool:	1✔
346	random_sender = "0x0000000000000000000000000000000000000002"	×
347	sender = safe_tx.sorted_signers[0] if safe_tx.sorted_signers else random_sender	×
348	data = {	×
349	"to": safe_tx.to,
350	"value": safe_tx.value,
351	"data": to_0x_hex_str(safe_tx.data) if safe_tx.data else None,
352	"operation": safe_tx.operation,
353	"gasToken": safe_tx.gas_token,
354	"safeTxGas": safe_tx.safe_tx_gas,
355	"baseGas": safe_tx.base_gas,
356	"gasPrice": safe_tx.gas_price,
357	"refundReceiver": safe_tx.refund_receiver,
358	"nonce": safe_tx.safe_nonce,
359	"contractTransactionHash": to_0x_hex_str(safe_tx.safe_tx_hash),
360	"sender": sender,
361	"signature": safe_tx.signatures.hex() if safe_tx.signatures else None,
362	"origin": "Safe-CLI",
363	}
364	response = self._post_request(	×
365	f"/api/v2/safes/{safe_tx.safe_address}/multisig-transactions/", data
366	)
367	if not response.ok:	×
368	raise SafeAPIException(f"Error posting transaction: {response.content!r}")	×
369	return True	×
370
371	def delete_transaction(self, safe_tx_hash: str, signature: str) -> bool:	1✔
372	"""
373
374	:param safe_tx_hash: hash of eip712 see in transaction_service_messages.py generate_remove_transaction_message function
375	:param signature: signature of safe_tx_hash by transaction proposer
376	:return:
377	"""
378	payload = {"safeTxHash": safe_tx_hash, "signature": signature}	×
379	response = self._delete_request(	×
380	f"/api/v2/multisig-transactions/{safe_tx_hash}/", payload
381	)
382	if not response.ok:	×
383	raise SafeAPIException(f"Error deleting transaction: {response.content!r}")	×
384	return True	×
385
386	def post_message(	1✔
387	self,
388	safe_address: ChecksumAddress,
389	message: Union[str, Dict],
390	signature: bytes,
391	safe_app_id: Optional[int] = 0,
392	) -> bool:
393	"""
394	Create safe message on transaction service for provided Safe address
395
396	:param safe_address:
397	:param message: If str it will be encoded using EIP191, and if it's a dictionary it will be encoded using EIP712
398	:param signature:
399	:return:
400	"""
401	payload = {	×
402	"message": message,
403	"safeAppId": safe_app_id,
404	"signature": to_0x_hex_str(signature),
405	}
406	response = self._post_request(	×
407	f"/api/v1/safes/{safe_address}/messages/", payload
408	)
409	if not response.ok:	×
410	raise SafeAPIException(f"Error posting message: {response.content!r}")	×
411	return True	×
412
413	def get_message(self, safe_message_hash: bytes) -> Message:	1✔
414	"""
415
416	:param safe_message_hash:
417	:return: Safe message for provided Safe message hash
418	"""
419	response = self._get_request(	×
420	f"/api/v1/messages/{to_0x_hex_str(safe_message_hash)}/"
421	)
422	if not response.ok:	×
423	raise SafeAPIException(f"Cannot get messages: {response.content!r}")	×
424	return response.json()	×
425
426	def get_messages(self, safe_address: ChecksumAddress) -> List[Message]:	1✔
427	"""
428
429	:param safe_address:
430	:return: list of messages for provided Safe address
431	"""
432	response = self._get_request(f"/api/v1/safes/{safe_address}/messages/")	×
433	if not response.ok:	×
434	raise SafeAPIException(f"Cannot get messages: {response.content!r}")	×
435	return response.json().get("results", [])	×
436
437	def post_message_signature(	1✔
438	self, safe_message_hash: bytes, signature: bytes
439	) -> bool:
440	"""
441	Add a new confirmation for provided Safe message hash
442
443	:param safe_message_hash:
444	:param signature:
445	:return:
446	"""
447	payload = {"signature": to_0x_hex_str(signature)}	×
448	response = self._post_request(	×
449	f"/api/v1/messages/{to_0x_hex_str(safe_message_hash)}/signatures/",
450	payload,
451	)
452	if not response.ok:	×
453	raise SafeAPIException(	×
454	f"Error posting message signature: {response.content!r}"
455	)
456	return True	×
457
458	def decode_data(	1✔
459	self, data: Union[bytes, HexStr], to_address: Optional[ChecksumAddress] = None
460	) -> DataDecoded:
461	"""
462	Retrieve decoded information using tx service internal ABI information given the tx data.
463
464	:param data: tx data as a 0x prefixed hexadecimal string.
465	:param to_address: address of the contract. This will be used in case of more than one function identifiers matching.
466	:return:
467	"""
468	payload = {"data": to_0x_hex_str(HexBytes(data))}	×
469	if to_address:	×
470	payload["to"] = to_address	×
471	response = self._post_request("/api/v1/data-decoder/", payload)	×
472	if not response.ok:	×
473	raise SafeAPIException(f"Cannot decode tx data: {response.content!r}")	×
474	return response.json()	×

safe-global / safe-eth-py / 18643806651

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous