Last active
March 22, 2020 17:58
-
-
Save boechat107/fba89766fe33ffedb22d8252f500d705 to your computer and use it in GitHub Desktop.
A Python decorator to cache functions and clear the cache as scheduled
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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