如何在Flask-Restful中使用Flask-Cache的@cache.cached()装饰器？例如，我有一个继承自Resource的Foo类，其中包含get、post、put和delete方法。
如何在POST后使缓存的结果无效？
@api.resource('/whatever')
class Foo(Resource):
@cache.cached(timeout=10)
def get(self):
return expensive_db_operation()
def post(self):
update_db_here()
## How do I invalidate the value cached in get()?
return something_useful()
from app import cache # The Flask-Cache object
from config import CACHE_REDIS_HOST, CACHE_REDIS_PORT # The Flask-Cache config
from redis import Redis
from flask import request
import urllib
redis_client = Redis(CACHE_REDIS_HOST, CACHE_REDIS_PORT)
def cache_key():
args = request.args
key = request.path + '?' + urllib.urlencode([
(k, v) for k in sorted(args) for v in sorted(args.getlist(k))
])
return key
@api.resource('/whatever')
class Foo(Resource):
@cache.cached(timeout=10, key_prefix=cache_key)
def get(self):
return expensive_db_operation()
def post(self):
update_db_here()
self.clear_cache()
return something_useful()
def clear_cache(self):
# Note: we have to use the Redis client to delete key by prefix,
# so we can't use the 'cache' Flask extension for this one.
key_prefix = request.path
keys = [key for key in redis_client.keys() if key.startswith(key_prefix)]
nkeys = len(keys)
for key in keys:
redis_client.delete(key)
if nkeys > 0:
log.info("Cleared %s cache keys" % nkeys)
log.info(keys)
cache.clear()方法来使缓存失效。更多详细信息请参见：https://pythonhosted.org/Flask-Cache/#flask.ext.cache.Cache.clear和https://pythonhosted.org/Flask-Cache/中的“清除缓存”部分。@JahMyst的回答对我没用。 Flask-Cache与Flask restful框架不兼容。根据他们的文档，@cache.Cached和@cache.memoize不能处理可变对象。
Using mutable objects (classes, etc) as part of the cache key can become tricky. It is suggested to not pass in an object instance into a memoized function. However, the memoize does perform a repr() on the passed in arguments so that if the object has a __repr__ function that returns a uniquely identifying string for that object, that will be used as part of the cache key.
我不得不自己实现。在这里留下代码片段，以防其他人遇到同样的问题。
cache_key 函数将用户请求转换为哈希值。
cache_res_pickled 函数用于对数据进行 pickle 或 unpickle。
|-flask-app
|-app.py
|-resource
|--some_resource.py
import json
import logging
import pickle
import time
import urllib
from flask import Response, abort, request
from redis import Redis
redis_client = Redis("127.0.0.1", "6379")
exp_setting_s = 1500
def json_serial(obj):
"""
JSON serializer for objects not serializable by default json code"
Args:
obj: JSON serialized object for dates
Returns:
serialized JSON data
"""
if isinstance(obj, datetime.datetime):
return obj.__str__()
def cache_key():
""" ""
Returns: Hashed string of request made by the user.
"""
args = request.args
key = (
request.path
+ "?"
+ urllib.parse.urlencode(
[(k, v) for k in sorted(args) for v in sorted(args.getlist(k))]
)
)
key_hashed = hashlib.sha256(key.encode())
return key_hashed.hexdigest()
def cache_res_pickled(data, encode):
"""
Args:
data (dict): Data in dict format
encode (Boolean): Encode (true) or decode (false) the data
Returns: Result after pickling
"""
if encode:
return pickle.dumps(data)
else:
data = pickle.loads(data)
return data
class SomeResource(Resource):
@auth.login_required
def get(self):
# Get the key for request in hashed format SHA256
key = cache_key()
result = redis_client.get(key)
def generate():
"""
A lagging generator to stream JSON so we don't have to hold everything in memory
This is a little tricky, as we need to omit the last comma to make valid JSON,
thus we use a lagging generator, similar to https://dev59.com/fnI-5IYBdhLWcg3w48tO
"""
releases = res.__iter__()
try:
prev_release = next(releases) # get first result
# We have some releases. First, yield the opening json
yield '{"data": ['
# Iterate over the releases
for release in releases:
yield json.dumps(prev_release, default=json_serial) + ", "
prev_release = release
logging.info(f"For {key} # records returned = {len(res)}")
# Now yield the last iteration without comma but with the closing brackets
yield json.dumps(prev_release, default=json_serial) + "]}"
except StopIteration:
# StopIteration here means the length was zero, so yield a valid releases doc and stop
logging.info(f"For {key} # records returned = {len(res)}")
yield '{"data": []}'
if result is None:
# Secure a key on Redis server.
redis_client.set(key, cache_res_pickled({}, True), ex=exp_setting_s)
try:
# Do the querying to the DB or math here to get res. It should be in dict format as shown below
res = {"A": 1, "B": 2, "C": 2}
# Update the key on Redis server with the latest data
redis_client.set(key, cache_res_pickled(res, True), ex=exp_setting_s)
return Response(generate(), content_type="application/json")
except Exception as e:
logging.exception(e)
abort(505, description="Resource not found. error - {}".format(e))
else:
res = cache_res_pickled(result, False)
if res:
logging.info(
f"The data already exists! loading the data form Redis cache for Key - {key} "
)
return Response(generate(), content_type="application/json")
else:
logging.info(
f"There is already a request for this key. But there is no data in it. Key: {key}."
)
s = time.time()
counter = 0
# loops aimlessly till the data is available on the Redis
while not any(res):
result = redis_client.get(key)
res = cache_res_pickled(result, False)
counter += 1
logging.info(
f"The data was available after {time.time() - s} seconds. Had to loop {counter} times."
)
return Response(generate(), content_type="application/json")
##create a decarator
from werkzeug.contrib.cache import SimpleCache
CACHE_TIMEOUT = 300
cache = SimpleCache()
class cached(object):
def __init__(self, timeout=None):
self.timeout = timeout or CACHE_TIMEOUT
def __call__(self, f):
def decorator(*args, **kwargs):
response = cache.get(request.path)
if response is None:
response = f(*args, **kwargs)
cache.set(request.path, response, self.timeout)
return response
return decorator
#add this decarator to your views like below
@app.route('/buildingTotal',endpoint='buildingTotal')
@cached()
def eventAlert():
return 'something'
@app.route('/buildingTenants',endpoint='buildingTenants')
@cached()
def buildingTenants():
return 'something'
from src.consts import config
from src.utils.external_services import redis_connector
import json
import jsons
import base64
class cached(object):
def __init__(self, req, timeout=None):
self.timeout = timeout or config.CACHE_DEFAULT_TIMEOUT
self.request = req
self.cache = redis_connector.get_redis_instance()
def __call__(self, f):
def decorator(*args, **kwargs):
redis_healthy = True
if self.cache is not None:
try:
self.cache.ping()
except Exception as ex:
redis_healthy = False
else:
redis_healthy = False
if self.request is not None and self.request.values is not None and self.request.path is not None and redis_healthy:
cache_key = "{}-{}".format(self.request.path, json.dumps(jsons.dump(self.request.values), sort_keys=True))
cache_key_base_64 = base64.b64encode(cache_key.encode("ascii")).decode("ascii")
response = self.cache.get(cache_key_base_64)
if response is None:
response = f(*args, **kwargs)
self.cache.setex(cache_key_base_64, self.timeout, jsons.dumps(response))
else:
response = json.loads(response)
else:
response = f(*args, **kwargs)
return response
return decorator
现在将这个装饰器用在您的API函数上
from flask import g, request
from flask_restful import Resource
from webargs.flaskparser import use_args
class GetProducts(Resource):
@use_args(gen_args.argsGetProducts)
@cached(request)
def get(self, args):
return "hello from products"