v0.1.29

implemented faster to_checksum_address with caching as it is surprisingly slow. Also speed up HexBytes conversion in Subsquid get_filter function

Author: simone1999

IceCreamSwapWeb3/EthAdvanced.py

@@ -299,7 +299,7 @@ def main(
         node_url="https://rpc-core.icecreamswap.com",
         usdt_address="0x900101d06A7426441Ae63e9AB3B9b0F63Be145F1",
 ):
-    from eth_utils import to_checksum_address
+    from IceCreamSwapWeb3 import to_checksum_address
 
     usdt_address = to_checksum_address(usdt_address)
 

IceCreamSwapWeb3/FastChecksumAddress.py

@@ -0,0 +1,27 @@
+import os
+from typing import cast
+from eth_hash.auto import keccak
+from eth_typing import ChecksumAddress
+
+CACHE_SIZE = int(os.getenv("CHECKSUM_CACHE_SIZE", 1000))
+CHECKSUM_CACHE = {}
+
+def to_checksum_address(address: str) -> ChecksumAddress:
+    """Fast checksum address with caching."""
+    normalized_address = address.lower().replace("0x", "")
+    assert len(normalized_address) == 40, "Address has incorrect length"
+
+    if normalized_address in CHECKSUM_CACHE:
+        return CHECKSUM_CACHE[normalized_address]
+
+    hashed_address = keccak(normalized_address.encode()).hex()
+    checksum_address = "0x" + "".join(
+        char.upper() if int(hashed_address[i], 16) >= 8 else char
+        for i, char in enumerate(normalized_address)
+    )
+
+    if len(CHECKSUM_CACHE) >= CACHE_SIZE:
+        CHECKSUM_CACHE.pop(next(iter(CHECKSUM_CACHE)))
+
+    CHECKSUM_CACHE[normalized_address] = checksum_address
+    return cast(ChecksumAddress, checksum_address)

IceCreamSwapWeb3/Multicall.py

@@ -5,12 +5,12 @@
 import eth_abi
 import eth_utils
 import rlp
-from eth_utils import to_checksum_address, to_bytes
+from eth_utils import to_bytes
 from eth_utils.abi import get_abi_output_types, get_abi_input_types
 from web3.contract.contract import ContractFunction, ContractConstructor
 from web3.exceptions import ContractLogicError
 
-from IceCreamSwapWeb3 import Web3Advanced
+from IceCreamSwapWeb3 import Web3Advanced, to_checksum_address
 
 # load multicall abi
 with files("IceCreamSwapWeb3").joinpath("./abi/Multicall.abi").open('r') as f:

IceCreamSwapWeb3/Subsquid.py

@@ -1,4 +1,7 @@
+from typing import cast
+
 import requests
+from IceCreamSwapWeb3 import to_checksum_address
 from eth_utils import to_checksum_address
 from hexbytes import HexBytes
 from tqdm import tqdm
@@ -103,10 +106,10 @@ def get_filter(
                     address=to_checksum_address(log['address']),
                     blockHash=block["header"]["hash"],
                     blockNumber=block["header"]["number"],
-                    data=HexBytes(log["data"]),
+                    data=cast(HexBytes, bytes.fromhex(log["data"][2:])),
                     logIndex=log["logIndex"],
-                    topics=[HexBytes(topic) for topic in log["topics"]],
-                    transactionHash=HexBytes(log["transactionHash"]),
+                    topics=[cast(HexBytes, bytes.fromhex(topic[2:])) for topic in log["topics"]],
+                    transactionHash=cast(HexBytes, bytes.fromhex(log["transactionHash"][2:])),
                     transactionIndex=log["transactionIndex"],
                     removed=False,
                 ))

IceCreamSwapWeb3/Web3Advanced.py

@@ -2,7 +2,7 @@
 from importlib.resources import files
 from time import sleep
 
-from eth_utils import to_checksum_address
+from IceCreamSwapWeb3 import to_checksum_address
 from web3 import Web3
 from web3.exceptions import ContractLogicError
 from web3.main import get_default_modules

IceCreamSwapWeb3/__init__.py

@@ -1 +1,2 @@
 from .Web3Advanced import Web3Advanced
+from .FastChecksumAddress import to_checksum_address

setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup, find_packages
 
-VERSION = '0.1.28'
+VERSION = '0.1.29'
 DESCRIPTION = 'IceCreamSwap Web3.py wrapper'
 LONG_DESCRIPTION = 'IceCreamSwap Web3.py wrapper with automatic retries, multicall and other advanced functionality'