18643788944

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

Build # 18643788944

Build Type

Pull #2055

github

Committed by

web-flow

Commit Message

Merge bca00d803 into 8eac6367d

Pull Request Pull Request #2055: Bump psycopg[binary] from 3.2.9 to 3.2.11

Run Details

9227 of 10258 relevant lines covered (89.95%)

1.8 hits per line

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

41.86

/safe_eth/safe/api/base_api.py

from abc import ABC, abstractmethod
from typing import Dict, Optional

import requests

from safe_eth.eth.ethereum_client import (
    EthereumClient,
    EthereumNetwork,
    EthereumNetworkNotSupported,
)
from safe_eth.util.http import build_full_url, prepare_http_session


class SafeAPIException(Exception):
    pass


class SafeBaseAPI(ABC):
    def __init__(
        self,
        network: EthereumNetwork,
        ethereum_client: Optional[EthereumClient] = None,
        base_url: Optional[str] = None,
        api_key: Optional[str] = None,
        request_timeout: int = 10,
    ):
        """
        :param network: Network for the transaction service
        :param ethereum_client:
        :param base_url: If a custom transaction service is used
        :param api_key: Api key of authenticated Safe services
        :raises: EthereumNetworkNotSupported
        """
        self.network = network
        self.ethereum_client = ethereum_client
        self.base_url = self._get_api_base_url(network, base_url)
        self.api_key = api_key
        self.http_session = prepare_http_session(10, 100)
        self.request_timeout = request_timeout

    @abstractmethod
    def _get_url_by_network(self, network: EthereumNetwork) -> Optional[str]:
        """
        Should return the base URL for the given network.
        :param network: EthereumNetwork to get the base URL for.
        :return: base URL for the given network. None if not found.
        """
        pass

    def _get_api_base_url(
        self, network: EthereumNetwork, custom_base_url: Optional[str] = None
    ) -> str:
        """
        Returns the base API URL for the specified Ethereum network.
        If a custom_base_url is provided, it will be used instead of the default.
        Otherwise, the method attempts to retrieve the URL associated with the given network.

        :param network: The EthereumNetwork to get the base URL for.
        :param custom_base_url: Optional custom base URL to override the default.
        :return: The base URL corresponding to the specified network.
        :raises EthereumNetworkNotSupported: If the network is not supported and no URL is found.
        """
        base_url = custom_base_url or self._get_url_by_network(network)
        if not base_url:
            raise EthereumNetworkNotSupported(network)
        return base_url

    @classmethod
    def from_ethereum_client(
        cls,
        ethereum_client: EthereumClient,
        api_key: Optional[str] = None,
    ) -> "SafeBaseAPI":
        ethereum_network = ethereum_client.get_network()
        return cls(ethereum_network, ethereum_client=ethereum_client, api_key=api_key)

    def _get_request_headers(self, include_json_body: bool = False) -> Dict[str, str]:
        """
        Build the default HTTP headers for a request.

        :param include_json_body: If it includes the JSON Content-Type header.
        :return: Dictionary of headers to include in the request.
        """
        headers = {}
        if include_json_body:
            headers["Content-Type"] = "application/json"
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        return headers

    def _get_request(self, url: str) -> requests.Response:
        full_url = build_full_url(self.base_url, url)
        return self.http_session.get(
            full_url, headers=self._get_request_headers(), timeout=self.request_timeout
        )

    def _post_request(self, url: str, payload: Dict) -> requests.Response:
        full_url = build_full_url(self.base_url, url)
        return self.http_session.post(
            full_url,
            json=payload,
            headers=self._get_request_headers(include_json_body=True),
            timeout=self.request_timeout,
        )

    def _delete_request(self, url: str, payload: Dict) -> requests.Response:
        full_url = build_full_url(self.base_url, url)
        return self.http_session.delete(
            full_url,
            json=payload,
            headers=self._get_request_headers(include_json_body=True),
            timeout=self.request_timeout,
        )

1	from abc import ABC, abstractmethod	2✔
2	from typing import Dict, Optional	2✔
3
4	import requests	2✔
5
6	from safe_eth.eth.ethereum_client import (	2✔
7	EthereumClient,
8	EthereumNetwork,
9	EthereumNetworkNotSupported,
10	)
11	from safe_eth.util.http import build_full_url, prepare_http_session	2✔
12
13
14	class SafeAPIException(Exception):	2✔
15	pass	2✔
16
17
18	class SafeBaseAPI(ABC):	2✔
19	def __init__(	2✔
20	self,
21	network: EthereumNetwork,
22	ethereum_client: Optional[EthereumClient] = None,
23	base_url: Optional[str] = None,
24	api_key: Optional[str] = None,
25	request_timeout: int = 10,
26	):
27	"""
28	:param network: Network for the transaction service
29	:param ethereum_client:
30	:param base_url: If a custom transaction service is used
31	:param api_key: Api key of authenticated Safe services
32	:raises: EthereumNetworkNotSupported
33	"""
34	self.network = network	×
35	self.ethereum_client = ethereum_client	×
36	self.base_url = self._get_api_base_url(network, base_url)	×
37	self.api_key = api_key	×
38	self.http_session = prepare_http_session(10, 100)	×
39	self.request_timeout = request_timeout	×
40
41	@abstractmethod	2✔
42	def _get_url_by_network(self, network: EthereumNetwork) -> Optional[str]:	2✔
43	"""
44	Should return the base URL for the given network.
45	:param network: EthereumNetwork to get the base URL for.
46	:return: base URL for the given network. None if not found.
47	"""
48	pass	×
49
50	def _get_api_base_url(	2✔
51	self, network: EthereumNetwork, custom_base_url: Optional[str] = None
52	) -> str:
53	"""
54	Returns the base API URL for the specified Ethereum network.
55	If a custom_base_url is provided, it will be used instead of the default.
56	Otherwise, the method attempts to retrieve the URL associated with the given network.
57
58	:param network: The EthereumNetwork to get the base URL for.
59	:param custom_base_url: Optional custom base URL to override the default.
60	:return: The base URL corresponding to the specified network.
61	:raises EthereumNetworkNotSupported: If the network is not supported and no URL is found.
62	"""
63	base_url = custom_base_url or self._get_url_by_network(network)	×
64	if not base_url:	×
65	raise EthereumNetworkNotSupported(network)	×
66	return base_url	×
67
68	@classmethod	2✔
69	def from_ethereum_client(	2✔
70	cls,
71	ethereum_client: EthereumClient,
72	api_key: Optional[str] = None,
73	) -> "SafeBaseAPI":
74	ethereum_network = ethereum_client.get_network()	×
75	return cls(ethereum_network, ethereum_client=ethereum_client, api_key=api_key)	×
76
77	def _get_request_headers(self, include_json_body: bool = False) -> Dict[str, str]:	2✔
78	"""
79	Build the default HTTP headers for a request.
80
81	:param include_json_body: If it includes the JSON Content-Type header.
82	:return: Dictionary of headers to include in the request.
83	"""
84	headers = {}	×
85	if include_json_body:	×
86	headers["Content-Type"] = "application/json"	×
87	if self.api_key:	×
88	headers["Authorization"] = f"Bearer {self.api_key}"	×
89	return headers	×
90
91	def _get_request(self, url: str) -> requests.Response:	2✔
92	full_url = build_full_url(self.base_url, url)	×
93	return self.http_session.get(	×
94	full_url, headers=self._get_request_headers(), timeout=self.request_timeout
95	)
96
97	def _post_request(self, url: str, payload: Dict) -> requests.Response:	2✔
98	full_url = build_full_url(self.base_url, url)	×
99	return self.http_session.post(	×
100	full_url,
101	json=payload,
102	headers=self._get_request_headers(include_json_body=True),
103	timeout=self.request_timeout,
104	)
105
106	def _delete_request(self, url: str, payload: Dict) -> requests.Response:	2✔
107	full_url = build_full_url(self.base_url, url)	×
108	return self.http_session.delete(	×
109	full_url,
110	json=payload,
111	headers=self._get_request_headers(include_json_body=True),
112	timeout=self.request_timeout,
113	)

safe-global / safe-eth-py / 18643788944

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