Add async methods to source contracts clients

safe_eth/eth/clients/blockscout_client.py

@@ -1,7 +1,9 @@
 import json
+import os
 from typing import Any, Dict, Optional
 from urllib.parse import urljoin
 
+import aiohttp
 import requests
 from eth_typing import ChecksumAddress
 
@@ -150,14 +152,26 @@ class BlockscoutClient:
         EthereumNetwork.EXSAT_TESTNET: "https://scan-testnet.exsat.network/api/v1/graphql",
     }
 
-    def __init__(self, network: EthereumNetwork):
+    def __init__(
+        self,
+        network: EthereumNetwork,
+        request_timeout: int = int(
+            os.environ.get("BLOCKSCOUT_CLIENT_REQUEST_TIMEOUT", 10)
+        ),
+        max_requests: int = int(os.environ.get("BLOCKSCOUT_CLIENT_MAX_REQUESTS", 100)),
+    ):
         self.network = network
         self.grahpql_url = self.NETWORK_WITH_URL.get(network, "")
+        self.request_timeout = request_timeout
         if not self.grahpql_url:
             raise BlockScoutConfigurationProblem(
                 f"Network {network.name} - {network.value} not supported"
             )
         self.http_session = requests.Session()
+        # Limit simultaneous connections to the same host.
+        self.async_session = aiohttp.ClientSession(
+            connector=aiohttp.TCPConnector(limit_per_host=max_requests)
+        )
 
     def build_url(self, path: str):
         return urljoin(self.grahpql_url, path)
@@ -169,6 +183,18 @@ def _do_request(self, url: str, query: str) -> Optional[Dict[str, Any]]:
 
         return response.json()
 
+    async def _async_do_request(self, url: str, query: str) -> Optional[Dict[str, Any]]:
+        """
+        Asynchronous version of _do_request
+        """
+        async with self.async_session.post(
+            url, son={"query": query}, timeout=self.request_timeout
+        ) as response:
+            if not response.ok:
+                return None
+
+            return await response.json()
+
     def get_contract_metadata(
         self, address: ChecksumAddress
     ) -> Optional[ContractMetadata]:
@@ -185,3 +211,20 @@ def get_contract_metadata(
                 smart_contract["name"], json.loads(smart_contract["abi"]), False
             )
         return None
+
+    async def async_get_contract_metadata(
+        self, address: ChecksumAddress
+    ) -> Optional[ContractMetadata]:
+        query = '{address(hash: "%s") { hash, smartContract {name, abi} }}' % address
+        result = await self._async_do_request(self.grahpql_url, query)
+        if (
+            result
+            and "error" not in result
+            and result.get("data", {}).get("address", {})
+            and result["data"]["address"]["smartContract"]
+        ):
+            smart_contract = result["data"]["address"]["smartContract"]
+            return ContractMetadata(
+                smart_contract["name"], json.loads(smart_contract["abi"]), False
+            )
+        return None

safe_eth/eth/clients/sourcify_client.py

@@ -3,6 +3,8 @@
 from typing import Any, Dict, List, Optional
 from urllib.parse import urljoin
 
+import aiohttp
+
 from ...util.http import prepare_http_session
 from .. import EthereumNetwork
 from ..utils import fast_is_checksum_address
@@ -38,12 +40,17 @@ def __init__(
         request_timeout: int = int(
             os.environ.get("SOURCIFY_CLIENT_REQUEST_TIMEOUT", 10)
         ),
+        max_requests: int = int(os.environ.get("SOURCIFY_CLIENT_MAX_REQUESTS", 100)),
     ):
         self.network = network
         self.base_url_api = base_url_api
         self.base_url_repo = base_url_repo
-        self.http_session = prepare_http_session(10, 100)
+        self.http_session = prepare_http_session(10, max_requests)
         self.request_timeout = request_timeout
+        # Limit simultaneous connections to the same host.
+        self.async_session = aiohttp.ClientSession(
+            connector=aiohttp.TCPConnector(limit_per_host=max_requests)
+        )
 
         if not self.is_chain_supported(network.value):
             raise SourcifyClientConfigurationProblem(
@@ -66,6 +73,18 @@ def _do_request(self, url: str) -> Optional[Dict[str, Any]]:
 
         return response.json()
 
+    async def _async_do_request(self, url: str) -> Optional[Dict[str, Any]]:
+        """
+        Asynchronous version of _do_request
+        """
+        async with self.async_session.get(
+            url, timeout=self.request_timeout
+        ) as response:
+            if not response.ok:
+                return None
+
+            return await response.json()
+
     def is_chain_supported(self, chain_id: int) -> bool:
         chains = self.get_chains()
         if not chains:
@@ -107,3 +126,25 @@ def get_contract_metadata(
                 name = self._get_name_from_metadata(metadata)
                 return ContractMetadata(name, abi, match_type == "partial_match")
         return None
+
+    async def async_get_contract_metadata(
+        self, contract_address: str
+    ) -> Optional[ContractMetadata]:
+        """
+        Asynchronous version of get_contract_metadata
+        """
+        assert fast_is_checksum_address(
+            contract_address
+        ), "Expecting a checksummed address"
+
+        for match_type in ("full_match", "partial_match"):
+            url = urljoin(
+                self.base_url_repo,
+                f"/contracts/{match_type}/{self.network.value}/{contract_address}/metadata.json",
+            )
+            metadata = await self._async_do_request(url)
+            if metadata:
+                abi = self._get_abi_from_metadata(metadata)
+                name = self._get_name_from_metadata(metadata)
+                return ContractMetadata(name, abi, match_type == "partial_match")
+        return None