15611201093

Committed 12 Jun 2025 12:59PM UTC coverage: 93.976%. First build

Build # 15611201093

Build Type

push

github

Committed by

web-flow

Commit Message

Add api key to Safe services API and update base URLs (#1752)

* Add api key to safe services API and update base URLs

* Apply PR suggestions

Run Details

47 of 50 new or added lines in 4 files covered. (94.0%)

9609 of 10225 relevant lines covered (93.98%)

2.82 hits per line

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

86.05

/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) -> "SafeBaseAPI":
        ethereum_network = ethereum_client.get_network()
        return cls(ethereum_network, ethereum_client=ethereum_client)

    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	3✔
2	from typing import Dict, Optional	3✔
3
4	import requests	3✔
5
6	from safe_eth.eth.ethereum_client import (	3✔
7	EthereumClient,
8	EthereumNetwork,
9	EthereumNetworkNotSupported,
10	)
11	from safe_eth.util.http import build_full_url, prepare_http_session	3✔
12
13
14	class SafeAPIException(Exception):	3✔
15	pass	3✔
16
17
18	class SafeBaseAPI(ABC):	3✔
19	def __init__(	3✔
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	3✔
35	self.ethereum_client = ethereum_client	3✔
36	self.base_url = self._get_api_base_url(network, base_url)	3✔
37	self.api_key = api_key	3✔
38	self.http_session = prepare_http_session(10, 100)	3✔
39	self.request_timeout = request_timeout	3✔
40
41	@abstractmethod	3✔
42	def _get_url_by_network(self, network: EthereumNetwork) -> Optional[str]:	3✔
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	"""
NEW 48	pass	×
49
50	def _get_api_base_url(	3✔
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)	3✔
64	if not base_url:	3✔
65	raise EthereumNetworkNotSupported(network)	3✔
66	return base_url	3✔
67
68	@classmethod	3✔
69	def from_ethereum_client(cls, ethereum_client: EthereumClient) -> "SafeBaseAPI":	3✔
70	ethereum_network = ethereum_client.get_network()	3✔
71	return cls(ethereum_network, ethereum_client=ethereum_client)	3✔
72
73	def _get_request_headers(self, include_json_body: bool = False) -> Dict[str, str]:	3✔
74	"""
75	Build the default HTTP headers for a request.
76
77	:param include_json_body: If it includes the JSON Content-Type header.
78	:return: Dictionary of headers to include in the request.
79	"""
80	headers = {}	3✔
81	if include_json_body:	3✔
NEW 82	headers["Content-Type"] = "application/json"	×
83	if self.api_key:	3✔
84	headers["Authorization"] = f"Bearer {self.api_key}"	3✔
85	return headers	3✔
86
87	def _get_request(self, url: str) -> requests.Response:	3✔
88	full_url = build_full_url(self.base_url, url)	3✔
89	return self.http_session.get(	3✔
90	full_url, headers=self._get_request_headers(), timeout=self.request_timeout
91	)
92
93	def _post_request(self, url: str, payload: Dict) -> requests.Response:	3✔
94	full_url = build_full_url(self.base_url, url)	×
95	return self.http_session.post(	×
96	full_url,
97	json=payload,
98	headers=self._get_request_headers(include_json_body=True),
99	timeout=self.request_timeout,
100	)
101
102	def _delete_request(self, url: str, payload: Dict) -> requests.Response:	3✔
103	full_url = build_full_url(self.base_url, url)	×
104	return self.http_session.delete(	×
105	full_url,
106	json=payload,
107	headers=self._get_request_headers(include_json_body=True),
108	timeout=self.request_timeout,
109	)

safe-global / safe-eth-py / 15611201093

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