Advertisement
Guest User

Untitled

a guest
Mar 22nd, 2019
80
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
text 2.37 KB | None | 0 0
  1. import datetime
  2. import functools
  3. from typing import Callable, Optional
  4.  
  5.  
  6. def timed_cache(timedelta: datetime.timedelta,
  7. maxsize: Optional[int] = 128,
  8. typed: Optional[bool] = False) -> Callable:
  9. """Timeout is applied to the whole cache."""
  10. def decorator(func):
  11. # Apply @lru_cache to func
  12. func = functools.lru_cache(maxsize=maxsize, typed=typed)(func)
  13.  
  14. next_update = datetime.datetime.utcnow() - timedelta
  15.  
  16. @functools.wraps(func)
  17. def wrapped(*args, **kwargs):
  18. nonlocal next_update
  19.  
  20. now = datetime.datetime.utcnow()
  21. if now > next_update:
  22. func.cache_clear()
  23. next_update = now + timedelta
  24. return func(*args, **kwargs)
  25.  
  26. wrapped.cache_info = func.cache_info
  27. wrapped.cache_clear = func.cache_clear
  28.  
  29. return wrapped
  30.  
  31. return decorator
  32.  
  33.  
  34. def granular_timed_cache(timedelta: datetime.timedelta,
  35. maxsize: Optional[int] = 128,
  36. typed: Optional[bool] = False) -> Callable:
  37. """The timeout is applied to the cache of each individual result."""
  38. def decorator(func):
  39. _args_ = {}
  40.  
  41. # Apply @lru_cache to new func
  42. @functools.lru_cache(maxsize=maxsize, typed=typed)
  43. def _wrapped(expiration, *args, **kwargs):
  44. return func(*args, **kwargs)
  45.  
  46. @functools.wraps(func)
  47. def wrapped(*args, **kwargs):
  48. hashable_args = tuple(args) + tuple((k, v)
  49. for k, v in kwargs.items())
  50.  
  51. expiration = _args_.get(hashable_args, None)
  52.  
  53. now = datetime.datetime.utcnow()
  54. if not expiration or now >= expiration:
  55. expiration = now + timedelta
  56. _args_[hashable_args] = expiration
  57.  
  58. return _wrapped(expiration, *args, **kwargs)
  59.  
  60. wrapped.cache_info = _wrapped.cache_info
  61. wrapped.cache_clear = _wrapped.cache_clear
  62.  
  63. return wrapped
  64.  
  65. return decorator
  66.  
  67.  
  68. if __name__ == '__main__':
  69. import time
  70.  
  71. for decorator in timed_cache, granular_timed_cache:
  72. @decorator(datetime.timedelta(seconds=5))
  73. def f(num):
  74. print(f'The function was called with argument {num}')
  75. return num
  76.  
  77. for i in range(10):
  78. print(decorator.__name__, f(i % 2))
  79. time.sleep(1)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement