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()

Remove all items from the dict.

close()

commit()

copy()

Return a shallow copy of the dict.

create_cache_entry(call_results)

Decode one token's Multicall results into a reusable cache record.

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)

Encode metadata reads only for missing or expired negative entries.

get(key[, default])

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

get_file_size()

items()

Return a set-like object providing a view on the dict's items.

iteritems()

iterkeys()

itervalues()

keys()

Return a set-like object providing a view on the dict's keys.

load_token_details_with_multicall(chain_id, ...)

Populate the shared token cache with bounded concurrent Multicalls.

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()

Return an object providing a view on the dict's values.
__init__(filename=PosixPath('/home/runner/.tradingstrategy/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

generate_calls(chain_id, addresses)

Encode metadata reads only for missing or expired negative entries.

Bulk warmup follows the same daily negative retry policy as individual fetch_erc20_details calls. Timeless legacy negatives refresh once; valid positive metadata remains reusable across scanner phases and vaults.

Parameters
Returns

Lazy encoded subcalls for entries that need a network read.

Return type

Iterable[eth_defi.event_reader.multicall_batcher.EncodedCall]

create_cache_entry(call_results)

Decode one token’s Multicall results into a reusable cache record.

Preserve unavailable fields instead of inventing valid token metadata. checked_at bounds retry of negative symbols; it does not make cached supply a current observation. Numeric caps below defend serialisation against malformed contracts, not validation of economic token values.

Parameters

call_results (dict[str, eth_defi.event_reader.multicall_batcher.EncodedCallResult]) – symbol/name/decimals/totalSupply results for one token.

Returns

Address, decoded metadata and naive UTC observation timestamp.

Return type

dict

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

Populate the shared token cache with bounded concurrent Multicalls.

Lead metadata and historical reader preparation use this to amortise repeated ERC-20 reads across vaults. The cache stores metadata rather than historical observations, so warming it requests no block timestamps.

Parameters

  • chain_id (int) – Chain namespace for cache entries.

  • web3factory (eth_defi.event_reader.web3factory.Web3Factory) – Worker connection factory with optional RPC counters.

  • addresses (list[eth_typing.evm.HexAddress]) – Token addresses whose metadata may need warming.

  • display_progress (Union[str, bool]) – Whether to display warmup progress.

  • max_workers (int) – Maximum threaded request concurrency.

  • block_identifier (Union[Literal['latest', 'earliest', 'pending', 'safe', 'finalized'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, int]) – Source block for uncached metadata reads.

  • batch_size (int) – Regular encoded-subcall limit, including cold-cache reads.

  • checkpoint (int) – Number of entries between cache syncs.

Returns

Token and encoded-subcall counts as a TokenCacheWarmupResult.

Return type

eth_defi.token.TokenCacheWarmupResult

__new__(**kwargs)
clear()

Remove all items from the dict.

property conn: sqlite3.Connection

One connection per thread

copy()

Return a shallow copy of the dict.

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()

Return a set-like object providing a view on the dict’s items.

keys()

Return a set-like object providing a view on the dict’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()

Return an object providing a view on the dict’s values.