Coverage for expiring_lru_cache/__init__.py: 100%

1"""Core functionality.""" 

2 

3import functools 

4import logging 

5from datetime import datetime, timedelta 

6from typing import Any, Callable, Optional, Union 

7 

8from expiring_lru_cache.version import __version__ # noqa: F401 

9 

10logging.getLogger("expiring_lru_cache").addHandler(logging.NullHandler()) 

11 

12MINUTES = 60 

13HOURS = 60 * MINUTES 

14DAYS = 24 * HOURS 

15 

16 

17# Bug in type hints for `functools.lru_cache` + mypy 

18# https://github.com/python/mypy/issues/5858#issuecomment-932773127 

19def _init_cache( 

20 func: Callable, expires_after: Optional[int], *args: Any, **kwargs: Any 

21) -> Callable: 

22 """Initialize cached function by wrapping it in `functools.lru_cache`.""" 

23 cached_func = functools.lru_cache(*args, **kwargs)(func) 

24 if expires_after: 

25 setattr( 

26 cached_func, 

27 "__expires_at", 

28 datetime.now() + timedelta(seconds=expires_after), 

29 ) 

30 return cached_func 

31 

32 

33def _expired(cached_func: Callable) -> bool: 

34 """Check if cached function is expired.""" 

35 if hasattr(cached_func, "__expires_at"): 

36 return getattr(cached_func, "__expires_at") < datetime.now() 

37 return False 

38 

39 

40def lru_cache( 

41 expires_after: Optional[int] = None, 

42 *lru_args: Union[int, bool], 

43 **lru_kwargs: Union[int, bool], 

44) -> Callable: 

45 """ 

46 LRU caching with expiration period. 

47 

48 Acts as a drop-in replacement of `functools.lru_cache`. Arguments valid for 

49 `functools.lru_cache` can also be passed. 

50 :param expires_after: number of seconds after which to invalidate cache - `None` 

51 will never invalidate based on time. Convenience variables `MINUTES`, `HOURS` 

52 and `DAYS` are available (using `lru_cache(expires_after=2 * DAYS)`) 

53 :param args: `functools.lru_cache`'s positional arguments 

54 :param kwargs: `functools.lru_cache`'s keyword arguments 

55 :return: cached function 

56 """ 

57 

58 def decorate(func: Callable) -> Callable: 

59 cached_func = _init_cache(func, expires_after, *lru_args, **lru_kwargs) 

60 

61 @functools.wraps(func) 

62 def wrapper(*args: Union[int, bool], **kwargs: Union[int, bool]) -> Callable: 

63 nonlocal cached_func 

64 if _expired(cached_func): 

65 logging.debug("Resetting cache") 

66 cached_func = _init_cache(func, expires_after, *lru_args, **lru_kwargs) 

67 return cached_func(*args, **kwargs) 

68 

69 wrapper.cache_info = lambda: cached_func.cache_info() 

70 return wrapper 

71 

72 return decorate