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

92.5

/safe_eth/eth/account_abstraction/bundler_client.py

import logging
from functools import cache, lru_cache
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union

from eth_typing import ChecksumAddress, HexAddress, HexStr
from hexbytes import HexBytes

from safe_eth.util.http import prepare_http_session

from .exceptions import BundlerClientConnectionException, BundlerClientResponseException
from .user_operation import UserOperation, UserOperationV07
from .user_operation_receipt import UserOperationReceipt

logger = logging.getLogger(__name__)


class BundlerClient:
    """
    Account Abstraction client for EIP4337 bundlers
    """

    def __init__(
        self,
        url: str,
        retry_count: int = 0,
    ):
        self.url = url
        self.retry_count = retry_count
        self.http_session = prepare_http_session(1, 100, retry_count=retry_count)

    def __str__(self):
        return f"Bundler client at {self.url}"

    def _do_request(
        self, payload: Union[Dict[Any, Any], Sequence[Dict[Any, Any]]]
    ) -> Union[
        Optional[Union[Dict[str, Any]]],
        Optional[List[Dict[str, Any]]],
        Optional[List[str]],
    ]:
        """
        :param payload: Allows simple request or a batch request
        :return: Result of the request
        :raises BundlerClientConnectionException: If there's a problem connecting to the bundler
        :raises BundlerClientResponseException: If the request from the bundler contains an error
        """
        try:
            response = self.http_session.post(self.url, json=payload)
        except IOError as exception:
            raise BundlerClientConnectionException(
                f"Error connecting to bundler {self.url} : {exception}"
            ) from exception

        if not response.ok:
            raise BundlerClientConnectionException(
                f"Error connecting to bundler {self.url} : {response.status_code} {response.content!r}"
            )

        bundler_responses = response.json()
        if not isinstance(bundler_responses, list):
            bundler_responses = [bundler_responses]

        results = []
        for rpc_response in bundler_responses:
            result = rpc_response.get("result")
            if not result and "error" in rpc_response:
                error_str = f'Bundler returned error for payload {payload} : {rpc_response["error"]}'
                logger.warning(error_str)
                raise BundlerClientResponseException(error_str)
            results.append(result)

        if not isinstance(payload, list):
            return results[0]
        return results

    @staticmethod
    def _parse_user_operation_receipt(
        user_operation_receipt: Dict[str, Any]
    ) -> UserOperationReceipt:
        return UserOperationReceipt(
            HexBytes(user_operation_receipt["userOpHash"]),
            user_operation_receipt["entryPoint"],
            user_operation_receipt["sender"],
            int(user_operation_receipt["nonce"], 16),
            user_operation_receipt["paymaster"],
            int(user_operation_receipt["actualGasCost"], 16),
            int(user_operation_receipt["actualGasUsed"], 16),
            user_operation_receipt["success"],
            user_operation_receipt["reason"],
            user_operation_receipt["logs"],
        )

    @staticmethod
    def _get_user_operation_by_hash_payload(
        user_operation_hash: HexStr, request_id: int = 1
    ) -> Dict[str, Any]:
        return {
            "jsonrpc": "2.0",
            "method": "eth_getUserOperationByHash",
            "params": [user_operation_hash],
            "id": request_id,
        }

    @staticmethod
    def _get_user_operation_receipt_payload(
        user_operation_hash: HexStr, request_id: int = 1
    ) -> Dict[str, Any]:
        return {
            "jsonrpc": "2.0",
            "method": "eth_getUserOperationReceipt",
            "params": [user_operation_hash],
            "id": request_id,
        }

    @cache
    def get_chain_id(self):
        payload = {
            "jsonrpc": "2.0",
            "method": "eth_chainId",
            "params": [],
            "id": 1,
        }
        result = self._do_request(payload)
        return int(result, 16)

    @lru_cache(maxsize=1024)
    def get_user_operation_by_hash(
        self, user_operation_hash: HexStr
    ) -> Optional[Union[UserOperation, UserOperationV07]]:
        """
        https://docs.alchemy.com/reference/eth-getuseroperationbyhash

        :param user_operation_hash:
        :return: ``UserOperation`` or ``None`` if not found
        :raises BundlerClientConnectionException:
        :raises BundlerClientResponseException:
        """
        payload = self._get_user_operation_by_hash_payload(user_operation_hash)
        result = self._do_request(payload)
        if result and isinstance(result, dict):
            return UserOperation.from_bundler_response(user_operation_hash, result)
        else:
            return None

    @lru_cache(maxsize=1024)
    def get_user_operation_receipt(
        self, user_operation_hash: HexStr
    ) -> Optional[UserOperationReceipt]:
        """
        https://docs.alchemy.com/reference/eth-getuseroperationreceipt

        :param user_operation_hash:
        :return: ``UserOperationReceipt`` or ``None`` if not found
        :raises BundlerClientConnectionException:
        :raises BundlerClientResponseException:
        """
        payload = self._get_user_operation_receipt_payload(user_operation_hash)
        result = self._do_request(payload)
        if result and isinstance(result, dict):
            return UserOperationReceipt.from_bundler_response(result)
        else:
            return None

    @lru_cache(maxsize=1024)
    def get_user_operation_and_receipt(
        self, user_operation_hash: HexStr
    ) -> Optional[Tuple[Union[UserOperation, UserOperationV07], UserOperationReceipt]]:
        """
        Get UserOperation and UserOperationReceipt in the same request using a batch query.
        NOTE: Batch requests are not supported by Pimlico

        :param user_operation_hash:
        :return: Tuple with ``UserOperation`` and ``UserOperationReceipt``, or ``None`` if not found
        :raises BundlerClientConnectionException:
        :raises BundlerClientResponseException:
        """
        payload = [
            self._get_user_operation_by_hash_payload(user_operation_hash, request_id=1),
            self._get_user_operation_receipt_payload(user_operation_hash, request_id=2),
        ]
        results = self._do_request(payload)
        if (
            results
            and isinstance(results, list)
            and len(results) >= 2
            and isinstance(results[0], dict)
            and isinstance(results[1], dict)
        ):
            return UserOperation.from_bundler_response(
                user_operation_hash, results[0]
            ), self._parse_user_operation_receipt(results[1])

        return None

    @lru_cache(maxsize=None)
    def supported_entry_points(self) -> List[ChecksumAddress]:
        """
        https://docs.alchemy.com/reference/eth-supportedentrypoints

        :return: List of supported entrypoints
        :raises BundlerClientConnectionException:
        :raises BundlerClientResponseException:
        """
        payload = {
            "jsonrpc": "2.0",
            "method": "eth_supportedEntryPoints",
            "params": [],
            "id": 1,
        }
        result = self._do_request(payload)
        if result and isinstance(result, list):
            return [
                ChecksumAddress(HexAddress(HexStr(address)))
                for address in result
                if isinstance(address, str)
            ]

        return []

1	import logging	1✔
2	from functools import cache, lru_cache	1✔
3	from typing import Any, Dict, List, Optional, Sequence, Tuple, Union	1✔
4
5	from eth_typing import ChecksumAddress, HexAddress, HexStr	1✔
6	from hexbytes import HexBytes	1✔
7
8	from safe_eth.util.http import prepare_http_session	1✔
9
10	from .exceptions import BundlerClientConnectionException, BundlerClientResponseException	1✔
11	from .user_operation import UserOperation, UserOperationV07	1✔
12	from .user_operation_receipt import UserOperationReceipt	1✔
13
14	logger = logging.getLogger(__name__)	1✔
15
16
17	class BundlerClient:	1✔
18	"""
19	Account Abstraction client for EIP4337 bundlers
20	"""
21
22	def __init__(	1✔
23	self,
24	url: str,
25	retry_count: int = 0,
26	):
27	self.url = url	1✔
28	self.retry_count = retry_count	1✔
29	self.http_session = prepare_http_session(1, 100, retry_count=retry_count)	1✔
30
31	def __str__(self):	1✔
32	return f"Bundler client at {self.url}"	×
33
34	def _do_request(	1✔
35	self, payload: Union[Dict[Any, Any], Sequence[Dict[Any, Any]]]
36	) -> Union[
37	Optional[Union[Dict[str, Any]]],
38	Optional[List[Dict[str, Any]]],
39	Optional[List[str]],
40	]:
41	"""
42	:param payload: Allows simple request or a batch request
43	:return: Result of the request
44	:raises BundlerClientConnectionException: If there's a problem connecting to the bundler
45	:raises BundlerClientResponseException: If the request from the bundler contains an error
46	"""
47	try:	1✔
48	response = self.http_session.post(self.url, json=payload)	1✔
49	except IOError as exception:	1✔
50	raise BundlerClientConnectionException(	1✔
51	f"Error connecting to bundler {self.url} : {exception}"
52	) from exception
53
54	if not response.ok:	1✔
55	raise BundlerClientConnectionException(	×
56	f"Error connecting to bundler {self.url} : {response.status_code} {response.content!r}"
57	)
58
59	bundler_responses = response.json()	1✔
60	if not isinstance(bundler_responses, list):	1✔
61	bundler_responses = [bundler_responses]	1✔
62
63	results = []	1✔
64	for rpc_response in bundler_responses:	1✔
65	result = rpc_response.get("result")	1✔
66	if not result and "error" in rpc_response:	1✔
67	error_str = f'Bundler returned error for payload {payload} : {rpc_response["error"]}'	1✔
68	logger.warning(error_str)	1✔
69	raise BundlerClientResponseException(error_str)	1✔
70	results.append(result)	1✔
71
72	if not isinstance(payload, list):	1✔
73	return results[0]	1✔
74	return results	1✔
75
76	@staticmethod	1✔
77	def _parse_user_operation_receipt(	1✔
78	user_operation_receipt: Dict[str, Any]
79	) -> UserOperationReceipt:
80	return UserOperationReceipt(	1✔
81	HexBytes(user_operation_receipt["userOpHash"]),
82	user_operation_receipt["entryPoint"],
83	user_operation_receipt["sender"],
84	int(user_operation_receipt["nonce"], 16),
85	user_operation_receipt["paymaster"],
86	int(user_operation_receipt["actualGasCost"], 16),
87	int(user_operation_receipt["actualGasUsed"], 16),
88	user_operation_receipt["success"],
89	user_operation_receipt["reason"],
90	user_operation_receipt["logs"],
91	)
92
93	@staticmethod	1✔
94	def _get_user_operation_by_hash_payload(	1✔
95	user_operation_hash: HexStr, request_id: int = 1
96	) -> Dict[str, Any]:
97	return {	1✔
98	"jsonrpc": "2.0",
99	"method": "eth_getUserOperationByHash",
100	"params": [user_operation_hash],
101	"id": request_id,
102	}
103
104	@staticmethod	1✔
105	def _get_user_operation_receipt_payload(	1✔
106	user_operation_hash: HexStr, request_id: int = 1
107	) -> Dict[str, Any]:
108	return {	1✔
109	"jsonrpc": "2.0",
110	"method": "eth_getUserOperationReceipt",
111	"params": [user_operation_hash],
112	"id": request_id,
113	}
114
115	@cache	1✔
116	def get_chain_id(self):	1✔
117	payload = {	×
118	"jsonrpc": "2.0",
119	"method": "eth_chainId",
120	"params": [],
121	"id": 1,
122	}
123	result = self._do_request(payload)	×
124	return int(result, 16)	×
125
126	@lru_cache(maxsize=1024)	1✔
127	def get_user_operation_by_hash(	1✔
128	self, user_operation_hash: HexStr
129	) -> Optional[Union[UserOperation, UserOperationV07]]:
130	"""
131	https://docs.alchemy.com/reference/eth-getuseroperationbyhash
132
133	:param user_operation_hash:
134	:return: ``UserOperation`` or ``None`` if not found
135	:raises BundlerClientConnectionException:
136	:raises BundlerClientResponseException:
137	"""
138	payload = self._get_user_operation_by_hash_payload(user_operation_hash)	1✔
139	result = self._do_request(payload)	1✔
140	if result and isinstance(result, dict):	1✔
141	return UserOperation.from_bundler_response(user_operation_hash, result)	1✔
142	else:
143	return None	1✔
144
145	@lru_cache(maxsize=1024)	1✔
146	def get_user_operation_receipt(	1✔
147	self, user_operation_hash: HexStr
148	) -> Optional[UserOperationReceipt]:
149	"""
150	https://docs.alchemy.com/reference/eth-getuseroperationreceipt
151
152	:param user_operation_hash:
153	:return: ``UserOperationReceipt`` or ``None`` if not found
154	:raises BundlerClientConnectionException:
155	:raises BundlerClientResponseException:
156	"""
157	payload = self._get_user_operation_receipt_payload(user_operation_hash)	1✔
158	result = self._do_request(payload)	1✔
159	if result and isinstance(result, dict):	1✔
160	return UserOperationReceipt.from_bundler_response(result)	1✔
161	else:
162	return None	1✔
163
164	@lru_cache(maxsize=1024)	1✔
165	def get_user_operation_and_receipt(	1✔
166	self, user_operation_hash: HexStr
167	) -> Optional[Tuple[Union[UserOperation, UserOperationV07], UserOperationReceipt]]:
168	"""
169	Get UserOperation and UserOperationReceipt in the same request using a batch query.
170	NOTE: Batch requests are not supported by Pimlico
171
172	:param user_operation_hash:
173	:return: Tuple with ``UserOperation`` and ``UserOperationReceipt``, or ``None`` if not found
174	:raises BundlerClientConnectionException:
175	:raises BundlerClientResponseException:
176	"""
177	payload = [	1✔
178	self._get_user_operation_by_hash_payload(user_operation_hash, request_id=1),
179	self._get_user_operation_receipt_payload(user_operation_hash, request_id=2),
180	]
181	results = self._do_request(payload)	1✔
182	if (	1✔
183	results
184	and isinstance(results, list)
185	and len(results) >= 2
186	and isinstance(results[0], dict)
187	and isinstance(results[1], dict)
188	):
189	return UserOperation.from_bundler_response(	1✔
190	user_operation_hash, results[0]
191	), self._parse_user_operation_receipt(results[1])
192
193	return None	1✔
194
195	@lru_cache(maxsize=None)	1✔
196	def supported_entry_points(self) -> List[ChecksumAddress]:	1✔
197	"""
198	https://docs.alchemy.com/reference/eth-supportedentrypoints
199
200	:return: List of supported entrypoints
201	:raises BundlerClientConnectionException:
202	:raises BundlerClientResponseException:
203	"""
204	payload = {	1✔
205	"jsonrpc": "2.0",
206	"method": "eth_supportedEntryPoints",
207	"params": [],
208	"id": 1,
209	}
210	result = self._do_request(payload)	1✔
211	if result and isinstance(result, list):	1✔
212	return [	1✔
213	ChecksumAddress(HexAddress(HexStr(address)))
214	for address in result
215	if isinstance(address, str)
216	]
217
218	return []	×

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