[python]使用flask-caching缓存数据

简介

Flask-Caching 是 Flask 的一个扩展，为任何 Flask 应用步伐添加了对各种后端的缓存支持。它基于 cachelib 运行，并通过统一的 API 支持 werkzeug 的全部原始缓存后端。开发者还可以通过继承 flask_caching.backends.base.BaseCache 类来开发自己的缓存后端。
官方文档 - https://flask-caching.readthedocs.io/en/latest/index.html
安装

1. pip install Flask-Caching

复制代码

设置

缓存通过缓存实例来管理

1. from flask import Flask
2. from flask_caching import Cache
4. config = {
5.     "DEBUG": True,          # some Flask specific configs
6.     "CACHE_TYPE": "SimpleCache",  # Flask-Caching related configs
7.     "CACHE_DEFAULT_TIMEOUT": 300
8. }
9. app = Flask(__name__)
10. # tell Flask to use the above defined config
11. app.config.from_mapping(config)
12. cache = Cache(app)

复制代码

也可以使用init_app来延后配置缓存实例

1. cache = Cache(config={'CACHE_TYPE': 'SimpleCache'})
3. app = Flask(__name__)
4. cache.init_app(app)

复制代码

还可以提供一个备用的配置字典，假如有多个Cache缓存实例，每个实例使用不同的后端，这将非常有用。

1. #: Method A: During instantiation of class
2. cache = Cache(config={'CACHE_TYPE': 'SimpleCache'})
3. #: Method B: During init_app call
4. cache.init_app(app, config={'CACHE_TYPE': 'SimpleCache'})

复制代码

缓存视图函数

使用cached()装饰器缓存视图函数，默认使用path作为缓存的key

1. @app.route("/")
2. @cache.cached(timeout=50)
3. def index():
4.     return render_template('index.html')

复制代码

cached 装饰器还有另一个可选参数叫做 unless。这个参数担当一个可调用对象，它返回 True 或 False。假如 unless 返回 True，那么将完全跳过缓存机制。
为了在视图中动态确定超时时间，可以返回 CachedResponse，这是 flask.Response 的子类。

1. @app.route("/")
2. @cache.cached()
3. def index():
4.     return CachedResponse(
5.         response=make_response(render_template('index.html')),
6.         timeout=50,
7.     )

复制代码

缓存插拔式视图类

1. from flask.views import View
3. class MyView(View):
4.     @cache.cached(timeout=50)
5.     def dispatch_request(self):
6.         return 'Cached for 50s'

复制代码

缓存其它函数

使用相同的 @cached 装饰器，还可以缓存其他非视图相关的函数的效果。需要留意更换 key_prefix，否则它将使用 request.path 作为 cache_key。键控制从缓存中获取什么内容。例如，假如一个键在缓存中不存在，将会在缓存中创建一个新的键值对条目。否则，将会返回该键的值（即缓存的效果）。

1. @cache.cached(timeout=50, key_prefix='all_comments')
2. def get_all_comments():
3.     comments = do_serious_dbio()
4.     return [x.author for x in comments]
6. cached_comments = get_all_comments()

复制代码

自定义缓存键

有时您希望为每个路由定义自己的缓存键。使用 @cached 装饰器，您可以指定怎样生成这个键。当缓存键不应仅仅是默认的 key_prefix，而是必须从请求中的其他参数派生时，这大概会非常有用。例如，在缓存 POST 路由时，缓存键应该根据请求中的数据而不仅仅是路由或视图本身来确定，这时就可以使用这个功能。

1. def make_key():
2.    """A function which is called to derive the key for a computed value.
3.       The key in this case is the concat value of all the json request
4.       parameters. Other strategy could to use any hashing function.
5.    :returns: unique string for which the value should be cached.
6.    """
7.    user_data = request.get_json()
8.    return ",".join([f"{key}={value}" for key, value in user_data.items()])
10. @app.route("/hello", methods=["POST"])
11. @cache.cached(timeout=60, make_cache_key=make_key)
12. def some_func():
13.    ....

复制代码

记忆化

在记忆化中，函数参数也会包含在cache_key中

留意：对于不接收参数的函数来说，cached() 和 memoize() 现实上是相同的。

Memoize 也实用于方法，因为它会将 self或 cls 参数的身份作为缓存键的一部分。
记忆化背后的理论是，假如你有一个函数需要在一次请求中多次调用，那么它只会在第一次使用这些参数调用该函数时举行盘算。例如，一个 sqlalchemy 对象用来确定一个用户是否具有某个脚色。在一次请求中，你大概需要多次调用这个函数。为了避免每次需要这些信息时都访问数据库，你大概会做如下操作：

1. class Person(db.Model):
2.     @cache.memoize(50)
3.     def has_membership(self, role_id):
4.         return Group.query.filter_by(user=self, role_id=role_id).count() >= 1

复制代码

将可变对象（类等）作为缓存键的一部分大概会变得棘手。建议不要将对象实例转达给记忆化函数。然而，memoize 会对传入的参数执行 repr()，因此假如对象有一个返回唯一标识字符串的 __repr__ 函数，该字符串将被用作缓存键的一部分。
例如，一个 sqlalchemy 的 person 对象，返回数据库 ID 作为唯一标识符的一部分：

1. class Person(db.Model):
2.     def __repr__(self):
3.         return "%s(%s)" % (self.__class__.__name__, self.id)

复制代码

删除记忆化缓存

您大概需要按函数删除缓存。使用上述示例，假设您更改了用户的权限并将其分配给某个脚色，但现在您需要重新盘算他们是否拥有某些成员资格。您可以使用 delete_memoized() 函数来实现这一点：

1. cache.delete_memoized(user_has_membership)

复制代码

假如仅将函数名称作为参数提供，那么该函数的全部记忆化版本都将失效。然而，您可以通过提供与缓存时相同的参数值来删除特定的缓存。在下面的示例中，只有用户脚色的缓存被删除：

1. user_has_membership('demo', 'admin')
2. user_has_membership('demo', 'user')
4. cache.delete_memoized(user_has_membership, 'demo', 'user')

复制代码

假如一个类方法被记忆化，您必须将类作为第一个 *args 参数提供。

1. class Foobar(object):
2.     @classmethod
3.     @cache.memoize(5)
4.     def big_foo(cls, a, b):
5.         return a + b + random.randrange(0, 100000)
7. cache.delete_memoized(Foobar.big_foo, Foobar, 5, 2)

复制代码

缓存Jinja2模板

根本使用

1. {% cache [timeout [,[key1, [key2, ...]]]] %}
2. ...
3. {% endcache %}

复制代码

默认情况下，“模板文件路径” + “块开始行”的值被用作缓存键。此外，键名也可以手动设置。键会连接成一个字符串，这样可以避免在不同模板中评估相同的块。
将超时设置为 None 以表现没有超时，但可以使用自定义键。

1. {% cache None, "key" %}
2. ...
3. {% endcache %}

复制代码

设置timeout为del来删除缓存值

1. {% cache 'del', key1 %}
2. ...
3. {% endcache %}

复制代码

假如提供了键，您可以轻松生成模板片段的键，并在模板上下文外部删除它。

1. from flask_caching import make_template_fragment_key
2. key = make_template_fragment_key("key1", vary_on=["key2", "key3"])
3. cache.delete(key)

复制代码

思量使用render_form_field和render_submit

1. {% cache 60*5 %}
3.     <form>
4.     {% render_form_field(form.username) %}
5.     {% render_submit() %}
6.     </form>
8. {% endcache %}

复制代码

清空缓存

清空应用缓存的简单示例

1. from flask_caching import Cache
3. from yourapp import app, your_cache_config
5. cache = Cache()
8. def main():
9.     cache.init_app(app, config=your_cache_config)
11.     with app.app_context():
12.         cache.clear()
14. if __name__ == '__main__':
15.     main()

复制代码

某些后端实现不支持完全扫除缓存。此外，假如您不使用键前缀，一些实现（例如 Redis）会清空整个数据库。请确保您没有在缓存数据库中存储任何其他数据。
显式缓存数据

数据可以通过直接使用署理方法如 Cache.set() 和 Cache.get() 来显式缓存。通过 Cache 类还有很多其他可用的署理方法。

1. @app.route("/html")
2. @app.route("/html/<foo>")
3. def html(foo=None):
4.     if foo is not None:
5.         cache.set("foo", foo)
6.     bar = cache.get("foo")
7.     return render_template_string(
8.         "<html><body>foo cache: {{bar}}</body></html>", bar=bar
9.     )

复制代码

根本使用示例

1. from flask import Flask
2. from flask_caching import Cache
3. import time
5. flask_cache = Cache(config={'CACHE_TYPE': 'SimpleCache'})
7. app = Flask(__name__)
9. fake_db = {
10.     "zhangsan": "qwerty"
11. }
13. def do_io(username: str):
14.     time.sleep(0.01)
15.     return fake_db.get(username, "")
17. @app.get("/user/<username>")
18. def get_user(username):
19.     if data := flask_cache.get(username):
20.         print(f"getting data from cache, username: {username}")
21.         return data
22.     else:
23.         print("data not found in cache")
24.    
25.     db_data = do_io(username)
26.     flask_cache.set(username, db_data, timeout=10)
27.     return db_data
29. if __name__ == "__main__":
30.     flask_cache.init_app(app)
31.     app.run("127.0.0.1", 8000)

复制代码

• 测试
  

1. wrk -t1 -c10 -d30s http://127.0.0.1:8000/user/zhangsan

复制代码

SimpleCache在gunicorn中的问题

gunicorn会创建多个子历程，子历程之间是否共享simplecache？
先写一个普通的service，暴露两个api

• GET /cache/: 根据key name获取缓存值
  
• POST /cache: 添加缓存
  

1. from flask import Flask, request
2. from flask_caching import Cache
3. from typing import Optional
5. flask_config = {
6.     "CACHE_TYPE": "SimpleCache",
7.     "CACHE_DEFAULT_TIMEOUT": 300
8. }
10. app = Flask(__name__)
11. app.config.from_mapping(flask_config)
12. cache = Cache(app)
14. @app.get("/cache/<foo>")
15. def get_cached_data(foo: Optional[str]):
16.     if not foo:
17.         return "foo is None\n"
18.     cache_rst = cache.get(foo)
19.     if not cache_rst:
20.         return f"key {foo} is not in cache\n"
21.     return f"find key {foo} in cache, value is {cache_rst}\n"
23. @app.post("/cache")
24. def set_cached_data():
25.     try:
26.         req_body = request.get_json()
27.     except Exception as e:
28.         raise Exception(f"request body is not json format, error: {e}\n") from e
29.    
30.     key = req_body.get("key", None)
31.     value = req_body.get("value", None)
32.     if not key or not value:
33.         return "key or value is None\n"
34.     if cached_data := cache.get(key):
35.         return f"key {key} is already in cache, value is {cached_data}\n"
36.     cache.set(key, value)
37.     return f"set key {key} in cache, value is {value}\n"
39. if __name__ == "__main__":
40.     app.run(host="0.0.0.0", port=5000)

复制代码

先用flask默认运行方式运行，测试接口是否正常

1. # 添加键值对缓存
2. curl -X POST http://127.0.0.1:5000/cache -H 'Content-Type: application/json' -d '{"key": "k1", "value": "v1"}'
4. # 获取缓存
5. curl http://127.0.0.1:5000/cache/k1

复制代码

假如响应正常的话，再用gunicorn启动。如下命令将启动4个工作子历程

1. gunicorn demo:app -b 0.0.0.0:5000 -w 4 -k gevent --worker-connections 2000

复制代码

请求测试。第一个请求设置缓存，后面四个获取缓存，可见工作历程之间并不共享flask_cache。假如用gunicorn或多个flask service实例，最好换其他cache type，比如RedisCache。

1. $ curl -X POST http://127.0.0.1:5000/cache -H 'Content-Type: application/json' -d '{"key": "k1", "value": "v1"}'
2. set key k1 in cache, value is v1
4. $ curl http://127.0.0.1:5000/cache/k1
5. key k1 is not in cache
7. $ curl http://127.0.0.1:5000/cache/k1
8. key k1 is not in cache
10. $ curl http://127.0.0.1:5000/cache/k1
11. find key k1 in cache, value is v1
13. $ curl http://127.0.0.1:5000/cache/k1
14. key k1 is not in cache

复制代码

参考

• 官方文档：https://flask-caching.readthedocs.io/en/latest/index.html
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。