diff --git a/evmspec/data/_cache.py b/evmspec/data/_cache.py new file mode 100644 index 0000000..c5be924 --- /dev/null +++ b/evmspec/data/_cache.py @@ -0,0 +1,50 @@ +from importlib.metadata import version +from time import monotonic + +from cachetools import cached, keys +from cachetools.func import TTLCache, _UnboundTTLCache + + +_CACHETOOLS_VERSION = tuple(int(i) for i in version("cachetools").split(".")) + + +def ttl_cache(maxsize=128, ttl=600, timer=monotonic, typed=False): + """Decorator to wrap a function with a memoizing callable that saves + up to `maxsize` results based on a Least Recently Used (LRU) + algorithm with a per-item time-to-live (TTL) value. + """ + if maxsize is None: + return _cache(_UnboundTTLCache(ttl, timer), None, typed) + elif callable(maxsize): + return _cache(TTLCache(128, ttl, timer), 128, typed)(maxsize) + else: + return _cache(TTLCache(maxsize, ttl, timer), maxsize, typed) + + +def _cache(cache, maxsize, typed, info: bool = False): + # reimplement ttl_cache with no RLock for race conditions + + key = keys.typedkey if typed else keys.hashkey + get_params = lambda: {"maxsize": maxsize, "typed": typed} + + # `info` param was added in 5.3 + if _CACHETOOLS_VERSION >= (5, 3): + + def decorator(func): + wrapper = cached(cache=cache, key=key, lock=None, info=info)(func) + wrapper.cache_parameters = get_params + return wrapper + + elif info: + raise ValueError( + "You cannot use the `info` param with cachetools versions < 5.3" + ) + + else: + + def decorator(func): + wrapper = cached(cache=cache, key=key, lock=None)(func) + wrapper.cache_parameters = get_params + return wrapper + + return decorator diff --git a/evmspec/data/_main.py b/evmspec/data/_main.py index 66dda85..3f0aa50 100644 --- a/evmspec/data/_main.py +++ b/evmspec/data/_main.py @@ -4,16 +4,18 @@ from functools import cached_property from typing import TYPE_CHECKING, Any, Callable, Tuple, Type, TypeVar, Union -from cachetools.func import ttl_cache from cchecksum import to_checksum_address from hexbytes import HexBytes from msgspec import Raw, Struct, json from typing_extensions import Self +from evmspec.data._cache import ttl_cache + if TYPE_CHECKING: from evmspec.structs.log import Log from evmspec.structs.receipt import TransactionReceipt + _T = TypeVar("_T") """A generic type variable.""" @@ -88,7 +90,7 @@ def _decode_hook(cls, typ: Type["Address"], obj: str): return cls.checksum(obj) @classmethod - @ttl_cache(ttl=600) + @ttl_cache(maxsize=None, ttl=600) def checksum(cls, address: str) -> Self: """Returns the checksummed version of the address.