TokenDiskCache

Documentation for eth_defi.token.TokenDiskCache Python class.

class TokenDiskCache

Bases: eth_defi.sqlite_cache.PersistentKeyValueStore

Token cache that stores tokens in disk.

Use with fetch_erc20_details()
For loading hundreds of tokens once
Shared across chains
Enable fast cache warmup with load_token_details_with_multicall()
Persistent: Make sure subsequent batch jobs do not refetch token data over RPC as it is expensive
Store as a SQLite database

Example:

addresses = [
    "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",  # USDC
    "0x4200000000000000000000000000000000000006",  # WETH
    "0x50c5725949A6F0c72E6C4a641F24049A917DB0Cb",  # DAI
    "0x554a1283cecca5a46bc31c2b82d6702785fc72d9",  # UNI
]

cache = TokenDiskCache(tmp_path / "disk_cache.sqlite")
web3factory = MultiProviderWeb3Factory(JSON_RPC_BASE)
web3 = web3factory()

#
# Do single token lookups against cache
#
token = fetch_erc20_details(
    web3,
    token_address="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",
    chain_id=web3.eth.chain_id,
    cache=cache,
)
assert token.extra_data["cached"] == False
assert len(cache) == 1
# After one look up, we should have it cached
token = fetch_erc20_details(
    web3,
    token_address="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",
    chain_id=web3.eth.chain_id,
    cache=cache,
)
assert token.extra_data["cached"] == True
cache.purge()

#
# Warm up multiple on dry cache
#
result = cache.load_token_details_with_multicall(
    chain_id=web3.eth.chain_id,
    web3factory=web3factory,
    addresses=addresses,
    max_workers=max_workers,
    display_progress=False,
)
assert result["tokens_read"] == 4
assert "8453-0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913".lower() in cache
assert "8453-0x4200000000000000000000000000000000000006".lower() in cache

cache_data = cache["8453-0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913".lower()]
assert cache_data["name"] == "USD Coin"
assert cache_data["symbol"] == "USDC"
assert cache_data["decimals"] == 6
assert cache_data["supply"] > 1_000_000

token = fetch_erc20_details(
    web3,
    token_address="0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",
    chain_id=web3.eth.chain_id,
    cache=cache,
)
assert token.extra_data["cached"] == True

Parameters

filename – Path to the sqlite database
autocommit – Whether to autocommit every time new entry is added to the database

Attributes summary

`DEFAULT_TOKEN_DISK_CACHE_PATH`
`conn`	One connection per thread

Methods summary

`__init__`([filename, max_str_length])	param filename Path to the sqlite database
`clear`()
`close`()
`commit`()
`copy`()
`create_cache_entry`(call_results)	Map multicall results to token details data for one address
`decode_value`(value)	Hook to convert SQLite values to Python objects
`encode_multicalls`(address)	Generate multicalls for each token address
`encode_value`(value)	Hook to convert Python objects to cache format
`fromkeys`([value])	Create a new dictionary with keys from iterable and values set to value.
`generate_calls`(chain_id, addresses)
`get`(key[, default])	Return the value for key if key is in the dictionary, else default.
`get_file_size`()
`items`()
`iteritems`()
`iterkeys`()
`itervalues`()
`keys`()
`load_token_details_with_multicall`(chain_id, ...)	Warm up cache and load token details for multiple
`pop`(k[,d])	If the key is not found, return the default if given; otherwise, raise a KeyError.
`popitem`()	Remove and return a (key, value) pair as a 2-tuple.
`purge`()	Delete all keys and save.
`setdefault`(key[, default])	Insert key with a value of default if key is not in the dictionary.
`update`([E, ]**F)	If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]
`values`()

__init__(filename=PosixPath('~/.cache/eth-defi-tokens.sqlite'), max_str_length=256)

Parameters

filename – Path to the sqlite database
autocommit – Whether to autocommit every time new entry is added to the database
max_str_length (int) –

encode_value(value)

Hook to convert Python objects to cache format

Parameters: value (dict) –
Return type: Any

decode_value(value)

Hook to convert SQLite values to Python objects

Parameters: value (str) –
Return type: Any

encode_multicalls(address)

Generate multicalls for each token address

Parameters: address (eth_typing.evm.HexAddress) –
Return type: eth_defi.event_reader.multicall_batcher.EncodedCall

create_cache_entry(call_results)

Map multicall results to token details data for one address

Parameters: call_results (dict[str, eth_defi.event_reader.multicall_batcher.EncodedCallResult]) –
Return type: dict

load_token_details_with_multicall(chain_id, web3factory, addresses, display_progress=False, max_workers=8, block_identifier='latest', checkpoint=32)

Warm up cache and load token details for multiple

Parameters

chain_id (int) –
web3factory (eth_defi.event_reader.web3factory.Web3Factory) –
addresses (list[eth_typing.evm.HexAddress]) –
display_progress (str | bool) –
checkpoint (int) –

Return type

eth_defi.token.TokenCacheWarmupResult

__new__(**kwargs)

clear() → None. Remove all items from D.

property conn: sqlite3.Connection: One connection per thread

copy() → a shallow copy of D

fromkeys(value=None, /): Create a new dictionary with keys from iterable and values set to value.

get(key, default=None): Return the value for key if key is in the dictionary, else default.

items() → a set-like object providing a view on D's items

keys() → a set-like object providing a view on D's keys

pop(k[, d]) → v, remove specified key and return the corresponding value.: If the key is not found, return the default if given; otherwise, raise a KeyError.

popitem()

Remove and return a (key, value) pair as a 2-tuple.

Pairs are returned in LIFO (last-in, first-out) order. Raises KeyError if the dict is empty.

purge(): Delete all keys and save.

setdefault(key, default=None, /)

Insert key with a value of default if key is not in the dictionary.

Return the value for key if key is in the dictionary, else default.

update([E, ]**F) → None. Update D from mapping/iterable E and F.: If E is present and has a .keys() method, then does: for k in E.keys(): D[k] = E[k] If E is present and lacks a .keys() method, then does: for k, v in E: D[k] = v In either case, this is followed by: for k in F: D[k] = F[k]

values() → an object providing a view on D's values