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

41.79

/safe_eth/eth/clients/ens_client.py

import os
from dataclasses import dataclass
from functools import cache
from typing import Any, Dict, List, Optional, Union

import requests
from eth_typing import HexStr
from hexbytes import HexBytes

from safe_eth.util.util import to_0x_hex_str


class EnsClient:
    """
    Resolves Ethereum Name Service domains using ``thegraph`` API
    """

    @dataclass
    class Config:
        base_url: str

        @property
        def url(self) -> str:
            return self.base_url

    @dataclass
    class SubgraphConfig(Config):
        api_key: str
        subgraph_id: str

        @property
        def url(self):
            return f"{self.base_url}/api/{self.api_key}/subgraphs/id/{self.subgraph_id}"

    def __init__(self, config: Config):
        self.config = config
        self.request_timeout = int(
            os.environ.get("ENS_CLIENT_REQUEST_TIMEOUT", 5)
        )  # Seconds
        self.request_session = requests.Session()

    def is_available(self) -> bool:
        """
        :return: True if service is available, False if it's down
        """
        query = {"query": "{ __schema { queryType { name } } }"}
        try:
            response = self.request_session.post(
                self.config.url, json=query, timeout=self.request_timeout
            )
            return response.ok
        except IOError:
            return False

    @staticmethod
    def domain_hash_to_hex_str(domain_hash: Union[HexStr, bytes, int]) -> HexStr:
        """
        :param domain_hash:
        :return: Domain hash as an hex string of 66 chars (counting with 0x), padding with zeros if needed
        """
        if not domain_hash:
            domain_hash = b""
        return HexStr(to_0x_hex_str(HexBytes(domain_hash)).rjust(66, "0"))

    @cache
    def _query_by_domain_hash(self, domain_hash_str: HexStr) -> Optional[str]:
        query = """
                {
                    domains(where: {labelhash: "domain_hash"}) {
                        labelName
                    }
                }
                """.replace(
            "domain_hash", domain_hash_str
        )
        try:
            response = self.request_session.post(
                self.config.url,
                json={"query": query},
                timeout=self.request_timeout,
            )
        except IOError:
            return None

        """
        Example:
        {
            "data": {
                "domains": [
                    {
                        "labelName": "safe-multisig"
                    }
                ]
            }
        }
        """
        if response.ok:
            data = response.json()
            if data:
                domains = data.get("data", {}).get("domains")
                if domains:
                    return domains[0].get("labelName")
        return None

    def query_by_domain_hash(
        self, domain_hash: Union[HexStr, bytes, int]
    ) -> Optional[str]:
        """
        Get domain label from domain_hash (keccak of domain name without the TLD, don't confuse with namehash)
        used for ENS ERC721 token_id. Use another method for caching purposes (use same parameter type)

        :param domain_hash: keccak of domain name without the TLD, don't confuse with namehash. E.g. For
            batman.eth it would be just keccak('batman')
        :return: domain label if found
        """
        domain_hash_str = self.domain_hash_to_hex_str(domain_hash)
        return self._query_by_domain_hash(domain_hash_str)

    def query_by_account(self, account: str) -> Optional[List[Dict[str, Any]]]:
        """
        :param account: ethereum account to search for ENS registered addresses
        :return: None if there's a problem or not found, otherwise example of dictionary returned:
        {
            "registrations": [
                {
                    "domain": {
                        "isMigrated": true,
                        "labelName": "gilfoyle",
                        "labelhash": "0xadfd886b420023026d5c0b1be0ffb5f18bb2f37143dff545aeaea0d23a4ba910",
                        "name": "gilfoyle.eth",
                        "parent": {
                            "name": "eth"
                        }
                    },
                    "expiryDate": "1905460880"
                }
            ]
        }
        """
        query = """query getRegistrations {
          account(id: "account_id") {
            registrations {
              expiryDate
              domain {
                labelName
                labelhash
                name
                isMigrated
                parent {
                  name
                }
              }
            }
          }
        }""".replace(
            "account_id", account.lower()
        )
        try:
            response = self.request_session.post(
                self.config.url,
                json={"query": query},
                timeout=self.request_timeout,
            )
        except IOError:
            return None

        if response.ok:
            data = response.json()
            if data:
                return data.get("data", {}).get("account")
        return None

1	import os	1✔
2	from dataclasses import dataclass	1✔
3	from functools import cache	1✔
4	from typing import Any, Dict, List, Optional, Union	1✔
5
6	import requests	1✔
7	from eth_typing import HexStr	1✔
8	from hexbytes import HexBytes	1✔
9
10	from safe_eth.util.util import to_0x_hex_str	1✔
11
12
13	class EnsClient:	1✔
14	"""
15	Resolves Ethereum Name Service domains using ``thegraph`` API
16	"""
17
18	@dataclass	1✔
19	class Config:	1✔
20	base_url: str	1✔
21
22	@property	1✔
23	def url(self) -> str:	1✔
24	return self.base_url	×
25
26	@dataclass	1✔
27	class SubgraphConfig(Config):	1✔
28	api_key: str	1✔
29	subgraph_id: str	1✔
30
31	@property	1✔
32	def url(self):	1✔
33	return f"{self.base_url}/api/{self.api_key}/subgraphs/id/{self.subgraph_id}"	×
34
35	def __init__(self, config: Config):	1✔
36	self.config = config	×
37	self.request_timeout = int(	×
38	os.environ.get("ENS_CLIENT_REQUEST_TIMEOUT", 5)
39	) # Seconds
40	self.request_session = requests.Session()	×
41
42	def is_available(self) -> bool:	1✔
43	"""
44	:return: True if service is available, False if it's down
45	"""
46	query = {"query": "{ __schema { queryType { name } } }"}	×
47	try:	×
48	response = self.request_session.post(	×
49	self.config.url, json=query, timeout=self.request_timeout
50	)
51	return response.ok	×
52	except IOError:	×
53	return False	×
54
55	@staticmethod	1✔
56	def domain_hash_to_hex_str(domain_hash: Union[HexStr, bytes, int]) -> HexStr:	1✔
57	"""
58	:param domain_hash:
59	:return: Domain hash as an hex string of 66 chars (counting with 0x), padding with zeros if needed
60	"""
61	if not domain_hash:	×
62	domain_hash = b""	×
63	return HexStr(to_0x_hex_str(HexBytes(domain_hash)).rjust(66, "0"))	×
64
65	@cache	1✔
66	def _query_by_domain_hash(self, domain_hash_str: HexStr) -> Optional[str]:	1✔
67	query = """	×
68	{
69	domains(where: {labelhash: "domain_hash"}) {
70	labelName
71	}
72	}
73	""".replace(
74	"domain_hash", domain_hash_str
75	)
76	try:	×
77	response = self.request_session.post(	×
78	self.config.url,
79	json={"query": query},
80	timeout=self.request_timeout,
81	)
82	except IOError:	×
83	return None	×
84
85	"""	×
86	Example:
87	{
88	"data": {
89	"domains": [
90	{
91	"labelName": "safe-multisig"
92	}
93	]
94	}
95	}
96	"""
97	if response.ok:	×
98	data = response.json()	×
99	if data:	×
100	domains = data.get("data", {}).get("domains")	×
101	if domains:	×
102	return domains[0].get("labelName")	×
103	return None	×
104
105	def query_by_domain_hash(	1✔
106	self, domain_hash: Union[HexStr, bytes, int]
107	) -> Optional[str]:
108	"""
109	Get domain label from domain_hash (keccak of domain name without the TLD, don't confuse with namehash)
110	used for ENS ERC721 token_id. Use another method for caching purposes (use same parameter type)
111
112	:param domain_hash: keccak of domain name without the TLD, don't confuse with namehash. E.g. For
113	batman.eth it would be just keccak('batman')
114	:return: domain label if found
115	"""
116	domain_hash_str = self.domain_hash_to_hex_str(domain_hash)	×
117	return self._query_by_domain_hash(domain_hash_str)	×
118
119	def query_by_account(self, account: str) -> Optional[List[Dict[str, Any]]]:	1✔
120	"""
121	:param account: ethereum account to search for ENS registered addresses
122	:return: None if there's a problem or not found, otherwise example of dictionary returned:
123	{
124	"registrations": [
125	{
126	"domain": {
127	"isMigrated": true,
128	"labelName": "gilfoyle",
129	"labelhash": "0xadfd886b420023026d5c0b1be0ffb5f18bb2f37143dff545aeaea0d23a4ba910",
130	"name": "gilfoyle.eth",
131	"parent": {
132	"name": "eth"
133	}
134	},
135	"expiryDate": "1905460880"
136	}
137	]
138	}
139	"""
140	query = """query getRegistrations {	×
141	account(id: "account_id") {
142	registrations {
143	expiryDate
144	domain {
145	labelName
146	labelhash
147	name
148	isMigrated
149	parent {
150	name
151	}
152	}
153	}
154	}
155	}""".replace(
156	"account_id", account.lower()
157	)
158	try:	×
159	response = self.request_session.post(	×
160	self.config.url,
161	json={"query": query},
162	timeout=self.request_timeout,
163	)
164	except IOError:	×
165	return None	×
166
167	if response.ok:	×
168	data = response.json()	×
169	if data:	×
170	return data.get("data", {}).get("account")	×
171	return None	×

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