Skip to content

Adapter

safe_kit.adapter

EthAdapter

Bases: ABC

Abstract base class for Ethereum adapters.

Source code in safe_kit/adapter.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
class EthAdapter(ABC):
    """
    Abstract base class for Ethereum adapters.
    """

    @abstractmethod
    def get_balance(self, address: str) -> int:
        pass

    @abstractmethod
    def get_chain_id(self) -> int:
        pass

    @abstractmethod
    def get_safe_contract(self, safe_address: str) -> Any:
        pass

    @abstractmethod
    def get_contract(self, address: str, abi: list[Any]) -> Any:
        pass

    @abstractmethod
    def get_signer_address(self) -> str | None:
        pass

    @abstractmethod
    def sign_message(self, message: str) -> str:
        pass

    @abstractmethod
    def sign_typed_data(self, data: dict[str, Any]) -> str:
        pass

    @abstractmethod
    def get_storage_at(self, address: str, position: int) -> bytes:
        pass

    @abstractmethod
    def get_code(self, address: str) -> bytes:
        pass

    @abstractmethod
    def get_transaction_count(self, address: str) -> int:
        pass

    @abstractmethod
    def is_contract(self, address: str) -> bool:
        pass

    @abstractmethod
    def to_checksum_address(self, address: str) -> str:
        pass

    @abstractmethod
    def wait_for_transaction_receipt(self, tx_hash: str, timeout: int = 120) -> Any:
        pass

    @abstractmethod
    def send_transaction(self, tx: dict[str, Any]) -> str:
        """Send a raw transaction and return the transaction hash."""
        pass

    @abstractmethod
    def get_block(self, block_identifier: str | int = "latest") -> dict[str, Any]:
        """Get block information."""
        pass

get_block(block_identifier='latest') abstractmethod

Get block information.

Source code in safe_kit/adapter.py 
70
71
72
73
@abstractmethod
def get_block(self, block_identifier: str | int = "latest") -> dict[str, Any]:
    """Get block information."""
    pass

send_transaction(tx) abstractmethod

Send a raw transaction and return the transaction hash.

Source code in safe_kit/adapter.py 
65
66
67
68
@abstractmethod
def send_transaction(self, tx: dict[str, Any]) -> str:
    """Send a raw transaction and return the transaction hash."""
    pass

Web3Adapter

Bases: EthAdapter

Web3.py implementation of the EthAdapter.

Source code in safe_kit/adapter.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
class Web3Adapter(EthAdapter):
    """
    Web3.py implementation of the EthAdapter.
    """

    def __init__(self, web3: Web3, signer: LocalAccount | None = None):
        self.web3 = web3
        self.signer = signer

    def get_balance(self, address: str) -> int:
        return self.web3.eth.get_balance(self.web3.to_checksum_address(address))

    def get_chain_id(self) -> int:
        return self.web3.eth.chain_id

    def get_safe_contract(self, safe_address: str) -> Any:
        from safe_kit.abis import SAFE_ABI

        return self.web3.eth.contract(
            address=self.web3.to_checksum_address(safe_address), abi=SAFE_ABI
        )

    def get_contract(self, address: str, abi: list[Any]) -> Any:
        return self.web3.eth.contract(
            address=self.web3.to_checksum_address(address), abi=abi
        )

    def get_signer_address(self) -> str | None:
        if self.signer:
            return str(self.signer.address)
        return None

    def sign_message(self, message: str | bytes) -> str:
        if not self.signer:
            raise ValueError("No signer available")

        from eth_account.messages import encode_defunct

        if isinstance(message, bytes):
            signable_message = encode_defunct(primitive=message)
        elif message.startswith("0x"):
            signable_message = encode_defunct(hexstr=message)
        else:
            signable_message = encode_defunct(text=message)

        signed_message = self.signer.sign_message(signable_message)  # type: ignore[no-untyped-call]
        return str(signed_message.signature.hex())

    def sign_typed_data(self, data: dict[str, Any]) -> str:
        if not self.signer:
            raise ValueError("No signer available")
        from eth_account.messages import encode_typed_data

        signable_message = encode_typed_data(full_message=data)
        signed_message = self.signer.sign_message(signable_message)  # type: ignore[no-untyped-call]
        return str(signed_message.signature.hex())

    def get_storage_at(self, address: str, position: int) -> bytes:
        return self.web3.eth.get_storage_at(
            self.web3.to_checksum_address(address), position
        )

    def get_code(self, address: str) -> bytes:
        return self.web3.eth.get_code(self.web3.to_checksum_address(address))

    def get_transaction_count(self, address: str) -> int:
        return self.web3.eth.get_transaction_count(
            self.web3.to_checksum_address(address)
        )

    def is_contract(self, address: str) -> bool:
        code = self.get_code(address)
        return len(code) > 0

    def to_checksum_address(self, address: str) -> str:
        return self.web3.to_checksum_address(address)

    def wait_for_transaction_receipt(self, tx_hash: str, timeout: int = 120) -> Any:
        from hexbytes import HexBytes

        return self.web3.eth.wait_for_transaction_receipt(
            HexBytes(tx_hash), timeout=timeout
        )

    def send_transaction(self, tx: dict[str, Any]) -> str:
        """Send a raw transaction and return the transaction hash."""
        if not self.signer:
            raise ValueError("No signer available")
        tx_hash = self.web3.eth.send_transaction(tx)  # type: ignore[arg-type]
        return tx_hash.hex()

    def get_block(self, block_identifier: str | int = "latest") -> dict[str, Any]:
        """Get block information."""
        block = self.web3.eth.get_block(block_identifier)  # type: ignore[arg-type]
        return dict(block)

get_block(block_identifier='latest')

Get block information.

Source code in safe_kit/adapter.py 
167
168
169
170
def get_block(self, block_identifier: str | int = "latest") -> dict[str, Any]:
    """Get block information."""
    block = self.web3.eth.get_block(block_identifier)  # type: ignore[arg-type]
    return dict(block)

send_transaction(tx)

Send a raw transaction and return the transaction hash.

Source code in safe_kit/adapter.py 
160
161
162
163
164
165
def send_transaction(self, tx: dict[str, Any]) -> str:
    """Send a raw transaction and return the transaction hash."""
    if not self.signer:
        raise ValueError("No signer available")
    tx_hash = self.web3.eth.send_transaction(tx)  # type: ignore[arg-type]
    return tx_hash.hex()