Source code for eth_defi.token

"""ERC-20 token deployment and manipulation.

Deploy ERC-20 tokens to be used within your test suite.

`Read also unit test suite for tokens to see how ERC-20 can be manipulated in pytest <>`_.
from collections import OrderedDict
from dataclasses import dataclass
from decimal import Decimal
from functools import cached_property
from typing import Optional, Union

import cachetools
from eth_tester.exceptions import TransactionFailed
from eth_typing import HexAddress
from web3 import Web3
from web3.contract import Contract
from web3.exceptions import BadFunctionCallOutput, ContractLogicError

from eth_defi.abi import get_deployed_contract
from eth_defi.deploy import deploy_contract
from eth_defi.utils import sanitise_string

#: List of exceptions JSON-RPC provider can through when ERC-20 field look-up fails
#: TODO: Add exceptios from real HTTPS/WSS providers
#: `ValueError` is raised by Ganache
_call_missing_exceptions = (TransactionFailed, BadFunctionCallOutput, ValueError, ContractLogicError)

#: By default we cache 1024 token details using LRU.
DEFAULT_TOKEN_CACHE = cachetools.LRUCache(1024)

[docs]@dataclass class TokenDetails: """ERC-20 token Python presentation. - A helper class to work with ERC-20 tokens. - Read on-chain data, deal with token value decimal conversions. - Any field can be ``None`` for non-well-formed tokens. Example how to get USDC details on Polygon: .. code-block:: python usdc = fetch_erc20_details(web3, "0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174") # USDC on Polygon formatted = f"Token {} ({usdc.symbol}) at {usdc.address} on chain {usdc.chain_id}" assert formatted == "Token USD Coin (PoS) (USDC) at 0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174 on chain 137" """ #: The underlying ERC-20 contract proxy class instance contract: Contract #: Token name e.g. ``USD Circle`` name: Optional[str] = None #: Token symbol e.g. ``USDC`` symbol: Optional[str] = None #: Token supply as raw units total_supply: Optional[int] = None #: Number of decimals decimals: Optional[int] = None def __eq__(self, other): """Token is the same if it's on the same chain and has the same contract address.""" assert isinstance(other, TokenDetails) return (self.contract.address == other.contract.address) and (self.chain_id == other.chain_id) def __hash__(self): """Token hash.""" return hash((self.chain_id, self.contract.address)) def __repr__(self): return f"<{} ({self.symbol}) at {self.contract.address}, {self.decimals} decimals, on chain {self.chain_id}>" @cached_property def chain_id(self) -> int: """The EVM chain id where this token lives.""" return self.contract.w3.eth.chain_id @property def address(self) -> HexAddress: """The address of this token.""" return self.contract.address
[docs] def convert_to_decimals(self, raw_amount: int) -> Decimal: """Convert raw token units to decimals. Example: .. code-block:: python details = fetch_erc20_details(web3, token_address) # Convert 1 wei units to edcimals assert details.convert_to_decimals(1) == Decimal("0.0000000000000001") """ return Decimal(raw_amount) / Decimal(10**self.decimals)
[docs] def convert_to_raw(self, decimal_amount: Decimal) -> int: """Convert decimalised token amount to raw uint256. Example: .. code-block:: python details = fetch_erc20_details(web3, token_address) # Convert 1.0 USDC to raw unit with 6 decimals assert details.convert_to_raw(1) == 1_000_000 """ return int(decimal_amount * 10**self.decimals)
[docs] def fetch_balance_of(self, address: HexAddress | str, block_identifier="latest") -> Decimal: """Get an address token balance. :param block_identifier: A specific block to query if doing archive node historical queries :return: Converted to decimal using :py:meth:`convert_to_decimal` """ raw_amount = self.contract.functions.balanceOf(address).call(block_identifier=block_identifier) return self.convert_to_decimals(raw_amount)
[docs] @staticmethod def generate_cache_key(chain_id: int, address: str) -> int: """Generate a cache key for this token. - Cached by (chain, address) tuple - Validate the inputs before generating the key """ assert type(chain_id) == int assert type(address) == str assert address.startswith("0x") return hash((chain_id, address.lower()))
[docs]class TokenDetailError(Exception): """Cannot extract token details for an ERC-20 token for some reason."""
[docs]def create_token( web3: Web3, deployer: str, name: str, symbol: str, supply: int, decimals: int = 18, ) -> Contract: """Deploys a new ERC-20 token on local dev, testnet or mainnet. - Uses `ERC20Mock <>`_ contract for the deployment. - Waits until the transaction has completed Example: .. code-block:: # Deploys an ERC-20 token where 100,000 tokens are allocated ato the deployer address token = create_token(web3, deployer, "Hentai books token", "HENTAI", 100_000 * 10**18) print(f"Deployed token contract address is {token.address}") print(f"Deployer account {deployer} has {token.functions.balanceOf(user_1).call() / 10**18} tokens") Find more examples in :ref:`tutorials` and unit testing source code. :param web3: Web3 instance :param deployer: Deployer account as 0x address. Make sure this account has enough ETH or native token to cover the gas cost. :param name: Token name :param symbol: Token symbol :param supply: Token starting supply as raw units. E.g. ``500 * 10**18`` to have 500 tokens minted to the deployer at the start. :param decimals: How many decimals ERC-20 token values have :return: Instance to a deployed Web3 contract. """ return deploy_contract(web3, "ERC20MockDecimals.json", deployer, name, symbol, supply, decimals)
[docs]def fetch_erc20_details( web3: Web3, token_address: Union[HexAddress, str], max_str_length: int = 256, raise_on_error=True, contract_name="ERC20MockDecimals.json", cache: cachetools.Cache | None = DEFAULT_TOKEN_CACHE, chain_id: int = None, ) -> TokenDetails: """Read token details from on-chain data. Connect to Web3 node and do RPC calls to extract the token info. We apply some sanitazation for incoming data, like length checks and removal of null bytes. The function should not raise an exception as long as the underlying node connection does not fail. Example: .. code-block:: python details = fetch_erc20_details(web3, token_address) assert == "Hentai books token" assert details.decimals == 6 :param web3: Web3 instance :param token_address: ERC-20 contract address: :param max_str_length: For input sanitisation :param raise_on_error: If set, raise `TokenDetailError` on any error instead of silently ignoring in and setting details to None. :param contract_name: Contract ABI file to use. The default is ``ERC20MockDecimals.json``. For USDC use ``centre/FiatToken.json``. :param cache: Use this cache for cache token detail calls. The main purpose is to easily reduce JSON-RPC API call count. By default, we use LRU cache of 1024 entries. Set to ``None`` to disable the cache. Instance of :py:class:`cachetools.Cache'. See `cachetools documentation for details <>`__. :param chain_id: Chain id hint for the cache. If not given do ``eth_chainId`` RPC call to figure out. :return: Sanitised token info """ if not chain_id: chain_id = web3.eth.chain_id erc_20 = get_deployed_contract(web3, contract_name, token_address) key = TokenDetails.generate_cache_key(chain_id, token_address) if cache is not None: cached = cache.get(key) if cached is not None: return TokenDetails( erc_20, cached["name"], cached["symbol"], cached["supply"], cached["decimals"], ) try: symbol = sanitise_string(erc_20.functions.symbol().call()[0:max_str_length]) except _call_missing_exceptions as e: if raise_on_error: raise TokenDetailError(f"Token {token_address} missing symbol") from e symbol = None except OverflowError: # OverflowError: Python int too large to convert to C ssize_t # Que? # Sai Stablecoin uses bytes32 instead of string for name and symbol information # symbol = None try: name = sanitise_string([0:max_str_length]) except _call_missing_exceptions as e: if raise_on_error: raise TokenDetailError(f"Token {token_address} missing name") from e name = None except OverflowError: # OverflowError: Python int too large to convert to C ssize_t # Que? # Sai Stablecoin uses bytes32 instead of string for name and symbol information # name = None try: decimals = erc_20.functions.decimals().call() except _call_missing_exceptions as e: if raise_on_error: raise TokenDetailError(f"Token {token_address} missing decimals") from e decimals = 0 try: supply = erc_20.functions.totalSupply().call() except _call_missing_exceptions as e: if raise_on_error: raise TokenDetailError(f"Token {token_address} missing totalSupply") from e supply = None token_details = TokenDetails(erc_20, name, symbol, supply, decimals) if cache is not None: cache[key] = { "name": name, "symbol": symbol, "supply": supply, "decimals": decimals, } return token_details
[docs]def reset_default_token_cache(): """Purge the cached token data. See :py:data:`DEFAULT_TOKEN_CACHE` """ global DEFAULT_TOKEN_CACHE # Cache has a horrible API DEFAULT_TOKEN_CACHE.__dict__["_LRUCache__order"] = OrderedDict() DEFAULT_TOKEN_CACHE.__dict__["_Cache__currsize"] = 0 DEFAULT_TOKEN_CACHE.__dict__["_Cache__data"] = dict()