- fix(address/tagger.py): simplify yield statement and update addr di…

…ctionary with tag values

- fix(token/erc20.py): refactor ERC20TokenService to use singleton pattern and add initialization lock
- fix(searcher.py): remove unused imports and fix eth_balance_changes dictionary keys
- fix(translate.py): add import statement for ERC20TokenService and update balance_changed dictionary keys
- fix(tx_type.py): update contract_created type to Optional[str]
- chore(pyproject.toml): update package version to 0.5.8

decodex/convert/address/tagger.py

@@ -104,11 +104,9 @@ def lazy_tag(
  yield None
  continue
  tag: Dict[str, Any] = self._addr_tags.get(_relying_addr.lower(), {})
- extra_fields = {
- "name": tag.get("name", ""),
- "labels": tag.get("labels", []),
- }
- yield {**addr, **extra_fields}
+ addr["name"] = addr["name"] or tag.get("name", "")
+ addr["labels"] = tag.get("labels", [])
+ yield addr
  else:
  tag: Dict[str, Any] = self._addr_tags.get(addr.lower(), {})
  yield {

decodex/convert/token/erc20.py

@@ -9,12 +9,8 @@
 import diskcache
 from cachetools import cached
 from cachetools import LRUCache
-from eth_utils.address import to_checksum_address
 from multicall import Call
 from multicall import Multicall
-from web3 import Web3
-from web3.exceptions import BadFunctionCallOutput
-from web3.exceptions import ContractLogicError
 
 from decodex.constant import DECODEX_DIR
 from decodex.constant import NULL_ADDRESS_0x0
@@ -23,9 +19,22 @@
 
 
 class ERC20TokenService:
+ _instance = None
+ _initialize = False
+ _singleton_lock = Lock()
+
+ # Singleton pattern
+ def __new__(cls, *args, **kwargs):
+ if cls._instance is None:
+ cls._instance = super(ERC20TokenService, cls).__new__(cls)
+ return cls._instance
+
  def __init__(self, mc: Multicall, *, cache_path: Optional[str] = None) -> None:
- self._mc = mc
- self._cache = diskcache.Cache(cache_path or DECODEX_DIR.joinpath("erc20"))
+ with self._singleton_lock:
+ if not self._initialize:
+ self._mc = mc
+ self._cache = diskcache.Cache(cache_path or DECODEX_DIR.joinpath("erc20"))
+ self._initialize = True
 
  @cached(cache=LRUCache(maxsize=131072), lock=Lock())
  def get_erc20(
@@ -52,8 +61,9 @@ def get_erc20(
  return token
 
  if address in {NULL_ADDRESS_0x0, NULL_ADDRESS_0xF}:
+ suffix = "ETH Transfer" if address == NULL_ADDRESS_0x0 else "Gas Fee"
  return {
- "name": "Platform Token",
+ "name": f"Platform Token ({suffix})",
  "address": address,
  "contract_name": None,
  "symbol": "ETH",

decodex/search/searcher.py

@@ -15,8 +15,9 @@
 from web3 import Web3
 from web3.types import Wei
 
+from decodex.constant import NULL_ADDRESS_0x0
+from decodex.constant import NULL_ADDRESS_0xF
 from decodex.exceptions import RPCException
-from decodex.type import AccountBalanceChanged
 from decodex.type import Log
 from decodex.type import RawTraceCallResponse
 from decodex.type import RawTraceCallResult
@@ -102,7 +103,6 @@ def get_tx(self, txhash: str, *, max_workers: int = 2, show_revert_reason: bool
  }
  for log in tx_receipt["logs"]
  ]
-
  status = tx_receipt["status"]
  reason = ""
  if status == 0 and show_revert_reason:
@@ -125,13 +125,18 @@ def get_tx(self, txhash: str, *, max_workers: int = 2, show_revert_reason: bool
  gas_price = tx["gasPrice"]
  value = tx["value"]
 
- eth_balance_changes = {from_addr: {"ETH": -value, "Gas Fee": -gas_used * gas_price}}
+ eth_balance_changes = {
+ from_addr: {
+ NULL_ADDRESS_0x0: -value if status else 0,
+ NULL_ADDRESS_0xF: -gas_used * gas_price,
+ }
+ }
 
  # If to_address is an eoa, add the value to the balance change
  if to_addr and status == 1 and self.web3.eth.get_code(to_addr).hex() == "0x":
  if to_addr not in eth_balance_changes:
  eth_balance_changes[to_addr] = {}
- eth_balance_changes[to_addr]["ETH"] = value
+ eth_balance_changes[to_addr][NULL_ADDRESS_0x0] = value
 
  return {
  "txhash": tx_receipt["transactionHash"].hex(),

decodex/translate/translate.py

@@ -25,6 +25,7 @@
 from decodex.convert.address import TaggerFactory
 from decodex.convert.signature import SignatureFactory
 from decodex.convert.signature import SignatureLookUp
+from decodex.convert.token import ERC20TokenService
 from decodex.decode import eth_decode_input
 from decodex.decode import eth_decode_log
 from decodex.search import SearcherFactory
@@ -105,6 +106,7 @@ def __init__(
  self.hdlrs: Dict[str, EventHandleFunc] = {}
  self.web3 = Web3(Web3.HTTPProvider(provider_uri))
  self.__register__(self.evt_opts.keys() if defis == "all" else defis)
+ self._erc_svc = ERC20TokenService(self.mc)
 
  self.verbose = verbose
  if logger is None and verbose:
@@ -316,7 +318,7 @@ def proxy(addr: str) -> int:
  def _process_transfer(
  self,
  actions: List[TransferAction],
- eth_balance_changes: Dict[str, Dict[Literal["ETH", "Gas Fee"], int]],
+ eth_balance_changes: Dict[str, Dict[str, int]],
  ) -> List[AccountBalanceChanged]:
  tagged_mapping: Dict[str, TaggedAddr] = {}
  addr_token_pairs = set()
@@ -361,16 +363,20 @@ def _process_transfer(
  if balance_changed.get(account) is None:
  balance_changed[account] = {}
 
- if changes.get("ETH", 0):
- balance_changed[account]["ETH"] = {
- "asset": {"address": NULL_ADDRESS_0x0, "name": "Platform (Ether)", "labels": []},
- "balance_change": changes["ETH"] / 10**18,
+ if changes.get(NULL_ADDRESS_0x0, None):
+ platform_token = self._erc_svc.get_erc20(NULL_ADDRESS_0x0)
+ (asset,) = self.tagger([platform_token])
+ balance_changed[account][NULL_ADDRESS_0x0] = {
+ "asset": asset,
+ "balance_change": changes[NULL_ADDRESS_0x0] / 10 ** asset["decimals"],
  }
 
- if changes.get("Gas Fee", 0):
- balance_changed[account]["Gas Fee"] = {
- "asset": {"address": NULL_ADDRESS_0xF, "name": "Platform (Gas Fee)", "labels": []},
- "balance_change": changes["Gas Fee"] / 10**18,
+ if changes.get(NULL_ADDRESS_0xF, None):
+ platform_token = self._erc_svc.get_erc20(NULL_ADDRESS_0xF)
+ (asset,) = self.tagger([platform_token])
+ balance_changed[account][NULL_ADDRESS_0xF] = {
+ "asset": asset,
+ "balance_change": changes[NULL_ADDRESS_0xF] / 10 ** asset["decimals"],
  }
 
  account_balance_changed_list: List[AccountBalanceChanged] = [
@@ -425,8 +431,6 @@ def _process_tx(self, tx: Tx, max_workers: int) -> TaggedTx:
  msg = parse_utf8(tx["input"])
  if msg:
  actions = [UTF8Message(tx_from, tx_to, msg)]
- else:
- actions = []
 
  return {
  "txhash": tx["txhash"],

decodex/type/tx_type.py

@@ -59,7 +59,7 @@ class ERC20Compatible(TaggedAddr):
  "txhash": str, # transaction hash, hex string. 0x prefixed.
  "from": str, # from address, hex string. 0x prefixed.
  "to": Optional[str], # to address, hex string. 0x prefixed.
- "contract_created": Optional[TaggedAddr], # contract created by the transaction, hex string. 0x prefixed.
+ "contract_created": Optional[str], # contract created by the transaction, hex string. 0x prefixed.
  "block_number": int, # block number of the transaction
  "block_timestamp": int, # timestamp of the block, in seconds.
  "value": int, # value of the transaction, in wei.
@@ -69,7 +69,7 @@ class ERC20Compatible(TaggedAddr):
  "status": int, # status of the transaction
  "reason": str, # reason of the transaction if failed
  "logs": List[Log], # logs of the transaction
- "eth_balance_changes": Dict[str, Dict[Literal["ETH", "Gas Fee"], int]], # ETH balance change of the transaction
+ "eth_balance_changes": Dict[str, Dict[str, int]], # ETH balance change of the transaction
  },
  total=False,
 )

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "decodex"
-version = "0.5.7"
+version = "0.5.8"
 description = "Python package to decode dex actions"
 authors = ["alan890104 <alan890104@gmail.com>"]
 exclude = ["tests/**"]