Python使用flask-caching緩存數(shù)據(jù)的示例代碼

更新時(shí)間：2024年12月20日 10:19:43 作者：大雄野比

Flask-Caching 是 Flask 的一個(gè)擴(kuò)展,為任何 Flask 應(yīng)用程序添加了對(duì)各種后端的緩存支持,它基于 cachelib 運(yùn)行,并通過(guò)統(tǒng)一的 API 支持 werkzeug 的所有原始緩存后端,本文給大家介紹了Python使用flask-caching緩存數(shù)據(jù),需要的朋友可以參考下

簡(jiǎn)介

Flask-Caching 是 Flask 的一個(gè)擴(kuò)展，為任何 Flask 應(yīng)用程序添加了對(duì)各種后端的緩存支持。它基于 cachelib 運(yùn)行，并通過(guò)統(tǒng)一的 API 支持 werkzeug 的所有原始緩存后端。開(kāi)發(fā)者還可以通過(guò)繼承 flask_caching.backends.base.BaseCache 類(lèi)來(lái)開(kāi)發(fā)自己的緩存后端。

安裝

pip install Flask-Caching

設(shè)置

緩存通過(guò)緩存實(shí)例來(lái)管理

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來(lái)延后配置緩存實(shí)例

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

還可以提供一個(gè)備用的配置字典，如果有多個(gè)Cache緩存實(shí)例，每個(gè)實(shí)例使用不同的后端，這將非常有用。

#: 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'})

緩存視圖函數(shù)

使用cached()裝飾器緩存視圖函數(shù)，默認(rèn)使用path作為緩存的key

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

cached 裝飾器還有另一個(gè)可選參數(shù)叫做 unless。這個(gè)參數(shù)接受一個(gè)可調(diào)用對(duì)象，它返回 True 或 False。如果 unless 返回 True，那么將完全跳過(guò)緩存機(jī)制。

為了在視圖中動(dòng)態(tài)確定超時(shí)時(shí)間，可以返回 CachedResponse，這是 flask.Response 的子類(lèi)。

@app.route("/")
@cache.cached()
def index():
    return CachedResponse(
        response=make_response(render_template('index.html')),
        timeout=50,
    )

緩存插拔式視圖類(lèi)

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

緩存其它函數(shù)

使用相同的 @cached 裝飾器，還可以緩存其他非視圖相關(guān)的函數(shù)的結(jié)果。需要注意替換 key_prefix，否則它將使用 request.path 作為 cache_key。鍵控制從緩存中獲取什么內(nèi)容。例如，如果一個(gè)鍵在緩存中不存在，將會(huì)在緩存中創(chuàng)建一個(gè)新的鍵值對(duì)條目。否則，將會(huì)返回該鍵的值（即緩存的結(jié)果）。

@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()

自定義緩存鍵

有時(shí)您希望為每個(gè)路由定義自己的緩存鍵。使用 @cached 裝飾器，您可以指定如何生成這個(gè)鍵。當(dāng)緩存鍵不應(yīng)僅僅是默認(rèn)的 key_prefix，而是必須從請(qǐng)求中的其他參數(shù)派生時(shí)，這可能會(huì)非常有用。例如，在緩存 POST 路由時(shí)，緩存鍵應(yīng)該根據(jù)請(qǐng)求中的數(shù)據(jù)而不僅僅是路由或視圖本身來(lái)確定，這時(shí)就可以使用這個(gè)功能。

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():
   ....

記憶化

在記憶化中，函數(shù)參數(shù)也會(huì)包含在cache_key中

注意：對(duì)于不接收參數(shù)的函數(shù)來(lái)說(shuō)，cached() 和 memoize() 實(shí)際上是相同的。

Memoize 也適用于方法，因?yàn)樗鼤?huì)將 self或 cls 參數(shù)的身份作為緩存鍵的一部分。

記憶化背后的理論是，如果你有一個(gè)函數(shù)需要在一次請(qǐng)求中多次調(diào)用，那么它只會(huì)在第一次使用這些參數(shù)調(diào)用該函數(shù)時(shí)進(jìn)行計(jì)算。例如，一個(gè) sqlalchemy 對(duì)象用來(lái)確定一個(gè)用戶是否具有某個(gè)角色。在一次請(qǐng)求中，你可能需要多次調(diào)用這個(gè)函數(shù)。為了避免每次需要這些信息時(shí)都訪問(wèn)數(shù)據(jù)庫(kù)，你可能會(huì)做如下操作：

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

將可變對(duì)象（類(lèi)等）作為緩存鍵的一部分可能會(huì)變得棘手。建議不要將對(duì)象實(shí)例傳遞給記憶化函數(shù)。然而，memoize 會(huì)對(duì)傳入的參數(shù)執(zhí)行 repr()，因此如果對(duì)象有一個(gè)返回唯一標(biāo)識(shí)字符串的 __repr__ 函數(shù)，該字符串將被用作緩存鍵的一部分。

例如，一個(gè) sqlalchemy 的 person 對(duì)象，返回?cái)?shù)據(jù)庫(kù) ID 作為唯一標(biāo)識(shí)符的一部分：

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

刪除記憶化緩存

您可能需要按函數(shù)刪除緩存。使用上述示例，假設(shè)您更改了用戶的權(quán)限并將其分配給某個(gè)角色，但現(xiàn)在您需要重新計(jì)算他們是否擁有某些成員資格。您可以使用 delete_memoized() 函數(shù)來(lái)實(shí)現(xiàn)這一點(diǎn)：

cache.delete_memoized(user_has_membership)

如果僅將函數(shù)名稱(chēng)作為參數(shù)提供，那么該函數(shù)的所有記憶化版本都將失效。然而，您可以通過(guò)提供與緩存時(shí)相同的參數(shù)值來(lái)刪除特定的緩存。在下面的示例中，只有用戶角色的緩存被刪除：

user_has_membership('demo', 'admin')
user_has_membership('demo', 'user')
 
cache.delete_memoized(user_has_membership, 'demo', 'user')

如果一個(gè)類(lèi)方法被記憶化，您必須將類(lèi)作為第一個(gè) *args 參數(shù)提供。

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 %}

默認(rèn)情況下，“模板文件路徑” + “塊開(kāi)始行”的值被用作緩存鍵。此外，鍵名也可以手動(dòng)設(shè)置。鍵會(huì)連接成一個(gè)字符串，這樣可以避免在不同模板中評(píng)估相同的塊。

將超時(shí)設(shè)置為 None 以表示沒(méi)有超時(shí)，但可以使用自定義鍵。

{% cache None, "key" %}
...
{% endcache %}

設(shè)置timeout為del來(lái)刪除緩存值

{% 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 %}

清空緩存

清空應(yīng)用緩存的簡(jiǎn)單示例

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()

某些后端實(shí)現(xiàn)不支持完全清除緩存。此外，如果您不使用鍵前綴，一些實(shí)現(xiàn)（例如 Redis）會(huì)清空整個(gè)數(shù)據(jù)庫(kù)。請(qǐng)確保您沒(méi)有在緩存數(shù)據(jù)庫(kù)中存儲(chǔ)任何其他數(shù)據(jù)。

顯式緩存數(shù)據(jù)

數(shù)據(jù)可以通過(guò)直接使用代理方法如 Cache.set() 和 Cache.get() 來(lái)顯式緩存。通過(guò) Cache 類(lèi)還有許多其他可用的代理方法。

@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)

測(cè)試

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

SimpleCache在gunicorn中的問(wèn)題

gunicorn會(huì)創(chuàng)建多個(gè)子進(jìn)程，子進(jìn)程之間是否共享simplecache？

先寫(xiě)一個(gè)普通的service，暴露兩個(gè)api

GET /cache/<key_name>: 根據(jù)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默認(rèn)運(yùn)行方式運(yùn)行，測(cè)試接口是否正常

# 添加鍵值對(duì)緩存
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

如果響應(yīng)正常的話，再用gunicorn啟動(dòng)。如下命令將啟動(dòng)4個(gè)工作子進(jìn)程

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

請(qǐng)求測(cè)試。第一個(gè)請(qǐng)求設(shè)置緩存，后面四個(gè)獲取緩存，可見(jiàn)工作進(jìn)程之間并不共享flask_cache。如果用gunicorn或多個(gè)flask service實(shí)例，最好換其他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緩存數(shù)據(jù)的示例代碼的詳細(xì)內(nèi)容，更多關(guān)于Python flask-caching緩存數(shù)據(jù)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: