我要评论简介
flask-caching 是 flask 的一个扩展，为任何 flask 应用程序添加了对各种后端的缓存支持。它基于 cachelib 运行，并通过统一的 api 支持 werkzeug 的所有原始缓存后端。开发者还可以通过继承 flask_caching.backends.base.basecache 类来开发自己的缓存后端。
安装
pip install flask-caching
设置
缓存通过缓存实例来管理
from flask import flask
from flask_caching import cache
config = {
"debug": true, # some flask specific configs
"cache_type": "simplecache", # flask-caching related configs
"cache_default_timeout": 300
}
app = flask(__name__)
# tell flask to use the above defined config
app.config.from_mapping(config)
cache = cache(app)也可以使用init_app来延后配置缓存实例
cache = cache(config={'cache_type': 'simplecache'})
app = flask(__name__)
cache.init_app(app)还可以提供一个备用的配置字典，如果有多个cache缓存实例，每个实例使用不同的后端，这将非常有用。
#: method a: during instantiation of class
cache = cache(config={'cache_type': 'simplecache'})
#: method b: during init_app call
cache.init_app(app, config={'cache_type': 'simplecache'})缓存视图函数
使用cached()装饰器缓存视图函数，默认使用path作为缓存的key
@app.route("/")
@cache.cached(timeout=50)
def index():
return render_template('index.html')cached 装饰器还有另一个可选参数叫做 unless。这个参数接受一个可调用对象，它返回 true 或 false。如果 unless 返回 true，那么将完全跳过缓存机制。
为了在视图中动态确定超时时间，可以返回 cachedresponse，这是 flask.response 的子类。
@app.route("/")
@cache.cached()
def index():
return cachedresponse(
response=make_response(render_template('index.html')),
timeout=50,
)缓存插拔式视图类
from flask.views import view
class myview(view):
@cache.cached(timeout=50)
def dispatch_request(self):
return 'cached for 50s'缓存其它函数
使用相同的 @cached 装饰器，还可以缓存其他非视图相关的函数的结果。需要注意替换 key_prefix，否则它将使用 request.path 作为 cache_key。键控制从缓存中获取什么内容。例如，如果一个键在缓存中不存在，将会在缓存中创建一个新的键值对条目。否则，将会返回该键的值（即缓存的结果）。
@cache.cached(timeout=50, key_prefix='all_comments')
def get_all_comments():
comments = do_serious_dbio()
return [x.author for x in comments]
cached_comments = get_all_comments()自定义缓存键
有时您希望为每个路由定义自己的缓存键。使用 @cached 装饰器，您可以指定如何生成这个键。当缓存键不应仅仅是默认的 key_prefix，而是必须从请求中的其他参数派生时，这可能会非常有用。例如，在缓存 post 路由时，缓存键应该根据请求中的数据而不仅仅是路由或视图本身来确定，这时就可以使用这个功能。
def make_key():
"""a function which is called to derive the key for a computed value.
the key in this case is the concat value of all the json request
parameters. other strategy could to use any hashing function.
:returns: unique string for which the value should be cached.
"""
user_data = request.get_json()
return ",".join([f"{key}={value}" for key, value in user_data.items()])
@app.route("/hello", methods=["post"])
@cache.cached(timeout=60, make_cache_key=make_key)
def some_func():
....记忆化
在记忆化中，函数参数也会包含在cache_key中
注意：对于不接收参数的函数来说，cached() 和 memoize() 实际上是相同的。
memoize 也适用于方法，因为它会将 self或 cls 参数的身份作为缓存键的一部分。
记忆化背后的理论是，如果你有一个函数需要在一次请求中多次调用，那么它只会在第一次使用这些参数调用该函数时进行计算。例如，一个 sqlalchemy 对象用来确定一个用户是否具有某个角色。在一次请求中，你可能需要多次调用这个函数。为了避免每次需要这些信息时都访问数据库，你可能会做如下操作：
class person(db.model):
@cache.memoize(50)
def has_membership(self, role_id):
return group.query.filter_by(user=self, role_id=role_id).count() >= 1将可变对象（类等）作为缓存键的一部分可能会变得棘手。建议不要将对象实例传递给记忆化函数。然而，memoize 会对传入的参数执行 repr()，因此如果对象有一个返回唯一标识字符串的 __repr__ 函数，该字符串将被用作缓存键的一部分。
例如，一个 sqlalchemy 的 person 对象，返回数据库 id 作为唯一标识符的一部分：
class person(db.model):
def __repr__(self):
return "%s(%s)" % (self.__class__.__name__, self.id)删除记忆化缓存
您可能需要按函数删除缓存。使用上述示例，假设您更改了用户的权限并将其分配给某个角色，但现在您需要重新计算他们是否拥有某些成员资格。您可以使用 delete_memoized() 函数来实现这一点：
cache.delete_memoized(user_has_membership)
如果仅将函数名称作为参数提供，那么该函数的所有记忆化版本都将失效。然而，您可以通过提供与缓存时相同的参数值来删除特定的缓存。在下面的示例中，只有用户角色的缓存被删除：
user_has_membership('demo', 'admin')
user_has_membership('demo', 'user')
cache.delete_memoized(user_has_membership, 'demo', 'user')如果一个类方法被记忆化，您必须将类作为第一个 *args 参数提供。
class foobar(object):
@classmethod
@cache.memoize(5)
def big_foo(cls, a, b):
return a + b + random.randrange(0, 100000)
cache.delete_memoized(foobar.big_foo, foobar, 5, 2)缓存jinja2模板
基本使用
{% cache [timeout [,[key1, [key2, ...]]]] %}
...
{% endcache %}默认情况下，“模板文件路径” + “块开始行”的值被用作缓存键。此外，键名也可以手动设置。键会连接成一个字符串，这样可以避免在不同模板中评估相同的块。
将超时设置为 none 以表示没有超时，但可以使用自定义键。
{% cache none, "key" %}
...
{% endcache %}设置timeout为del来删除缓存值
{% cache 'del', key1 %}
...
{% endcache %}如果提供了键，您可以轻松生成模板片段的键，并在模板上下文外部删除它。
from flask_caching import make_template_fragment_key
key = make_template_fragment_key("key1", vary_on=["key2", "key3"])
cache.delete(key)考虑使用render_form_field和render_submit
{% cache 60*5 %}
<div>
<form>
{% render_form_field(form.username) %}
{% render_submit() %}
</form>
</div>
{% endcache %}清空缓存
清空应用缓存的简单示例
from flask_caching import cache
from yourapp import app, your_cache_config
cache = cache()
def main():
cache.init_app(app, config=your_cache_config)
with app.app_context():
cache.clear()
if __name__ == '__main__':
main()某些后端实现不支持完全清除缓存。此外，如果您不使用键前缀，一些实现（例如 redis）会清空整个数据库。请确保您没有在缓存数据库中存储任何其他数据。
显式缓存数据
数据可以通过直接使用代理方法如 cache.set() 和 cache.get() 来显式缓存。通过 cache 类还有许多其他可用的代理方法。
@app.route("/html")
@app.route("/html/<foo>")
def html(foo=none):
if foo is not none:
cache.set("foo", foo)
bar = cache.get("foo")
return render_template_string(
"<html><body>foo cache: {{bar}}</body></html>", bar=bar
)基本使用示例
from flask import flask
from flask_caching import cache
import time
flask_cache = cache(config={'cache_type': 'simplecache'})
app = flask(__name__)
fake_db = {
"zhangsan": "qwerty"
}
def do_io(username: str):
time.sleep(0.01)
return fake_db.get(username, "")
@app.get("/user/<username>")
def get_user(username):
if data := flask_cache.get(username):
print(f"getting data from cache, username: {username}")
return data
else:
print("data not found in cache")
db_data = do_io(username)
flask_cache.set(username, db_data, timeout=10)
return db_data
if __name__ == "__main__":
flask_cache.init_app(app)
app.run("127.0.0.1", 8000)- 测试
wrk -t1 -c10 -d30s http://127.0.0.1:8000/user/zhangsan
simplecache在gunicorn中的问题
gunicorn会创建多个子进程，子进程之间是否共享simplecache？
先写一个普通的service，暴露两个api
get /cache/<key_name>: 根据key name获取缓存值post /cache: 添加缓存
from flask import flask, request
from flask_caching import cache
from typing import optional
flask_config = {
"cache_type": "simplecache",
"cache_default_timeout": 300
}
app = flask(__name__)
app.config.from_mapping(flask_config)
cache = cache(app)
@app.get("/cache/<foo>")
def get_cached_data(foo: optional[str]):
if not foo:
return "foo is none\n"
cache_rst = cache.get(foo)
if not cache_rst:
return f"key {foo} is not in cache\n"
return f"find key {foo} in cache, value is {cache_rst}\n"
@app.post("/cache")
def set_cached_data():
try:
req_body = request.get_json()
except exception as e:
raise exception(f"request body is not json format, error: {e}\n") from e
key = req_body.get("key", none)
value = req_body.get("value", none)
if not key or not value:
return "key or value is none\n"
if cached_data := cache.get(key):
return f"key {key} is already in cache, value is {cached_data}\n"
cache.set(key, value)
return f"set key {key} in cache, value is {value}\n"
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)先用flask默认运行方式运行，测试接口是否正常
# 添加键值对缓存
curl -x post http://127.0.0.1:5000/cache -h 'content-type: application/json' -d '{"key": "k1", "value": "v1"}'
# 获取缓存
curl http://127.0.0.1:5000/cache/k1如果响应正常的话，再用gunicorn启动。如下命令将启动4个工作子进程
gunicorn demo:app -b 0.0.0.0:5000 -w 4 -k gevent --worker-connections 2000
请求测试。第一个请求设置缓存，后面四个获取缓存，可见工作进程之间并不共享flask_cache。如果用gunicorn或多个flask service实例，最好换其他cache type，比如rediscache。
$ curl -x post http://127.0.0.1:5000/cache -h 'content-type: application/json' -d '{"key": "k1", "value": "v1"}'
set key k1 in cache, value is v1
$ curl http://127.0.0.1:5000/cache/k1
key k1 is not in cache
$ curl http://127.0.0.1:5000/cache/k1
key k1 is not in cache
$ curl http://127.0.0.1:5000/cache/k1
find key k1 in cache, value is v1
$ curl http://127.0.0.1:5000/cache/k1
key k1 is not in cache以上就是python使用flask-caching缓存数据的示例代码的详细内容，更多关于python flask-caching缓存数据的资料请关注代码网其它相关文章！
发表评论