Skip to content

Instantly share code, notes, and snippets.

@boechat107
Last active March 22, 2020 17:58
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save boechat107/fba89766fe33ffedb22d8252f500d705 to your computer and use it in GitHub Desktop.
Save boechat107/fba89766fe33ffedb22d8252f500d705 to your computer and use it in GitHub Desktop.
A Python decorator to cache functions and clear the cache as scheduled
from datetime import datetime as dt
from croniter import croniter
from functools import wraps
from typing import Dict, Any
def cache_with_schedule(schedule: str):
"""
Decorator to cache the results of a function "f(x)" and clear the cache as
scheduled.
-> "schedule" is a cron-like string.
-> a "dict" is used as cache, so "x" needs to be hashable.
Example:
# "func" is invoked at most once in a minute.
@cache_with_schedule("*/1 * * * *")
def func(n):
print("I was invoked")
return {"a": n, "b": 20}
"""
nowfn = dt.utcnow
cron = croniter(schedule, nowfn())
cache: Dict[Any, Any] = dict()
def has_expired() -> bool:
"""
Returns "True" if the cache has expired.
"""
now = nowfn()
has_expired = False
while now > cron.get_next(dt):
has_expired = True
# Rewind one step to get the correct "next" date-time next time
# "has_expired" is invoked.
cron.get_prev(dt)
return has_expired
def decorator(f):
@wraps(f)
def wrapper(arg):
nonlocal cache
cache = dict() if has_expired() else cache
res = cache[arg] if arg in cache else f(arg)
cache[arg] = res
return res
return wrapper
return decorator
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment