10629724120

Committed 30 Aug 2024 08:49AM UTC coverage: 93.903%. Remained the same

Build # 10629724120

Build Type

Pull #1309

github

Committed by

web-flow

Commit Message

Merge 175bdc70e into d0479e88e

Pull Request Pull Request #1309: Add reference to main in safe-eth-py dependency in addresses actions

Run Details

8671 of 9234 relevant lines covered (93.9%)

3.76 hits per line

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

60.32

/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


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

safe-global / safe-eth-py / 10629724120

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