
Flask-Caching 完整示例 包含内存缓存、Redis缓存、文件系统缓存、视图缓存、函数缓存、手动清除缓存 import time import jsonfromflaskimportFlask,jsonify,request,render_template_stringfromflask_caching import Cache# 1. 创建 Flask 应用 appFlask(__name__)# 2. 缓存配置 # 以下展示三种常用缓存配置可以根据需要选择使用其中一种# -------------------- 配置 ASimpleCache内存缓存开发测试用--------------------# SimpleCache 使用 Python 字典在内存中存储非线程安全适合单进程开发环境# 注意生产环境不要使用 SimpleCache因为它不支持多进程/多服务器共享SIMPLE_CACHE_CONFIG{CACHE_TYPE:SimpleCache,# 缓存类型SimpleCache内存缓存 CACHE_DEFAULT_TIMEOUT:300,# 默认过期时间秒300秒5分钟 CACHE_THRESHOLD:500,# 最大缓存条目数超过后自动淘汰最旧的 CACHE_IGNORE_ERRORS:True,# 遇到错误时是否忽略True忽略}# -------------------- 配置 BRedisCache生产环境推荐支持分布式--------------------# Redis 是生产环境最常用的选择支持多进程/多服务器共享缓存# 使用前需要1) 安装 Redis 服务器并启动 2) pip install redis# 启动 Redis: redis-server 或 docker run -d -p 6379:6379 redisREDIS_CACHE_CONFIG{CACHE_TYPE:RedisCache,# 缓存类型RedisCache CACHE_REDIS_HOST:localhost,# Redis 服务器主机地址 CACHE_REDIS_PORT:6379,# Redis 服务器端口默认6379CACHE_REDIS_PASSWORD:,# Redis 密码没有则留空 CACHE_REDIS_DB:0,# 使用的 Redis 数据库编号0-15 CACHE_DEFAULT_TIMEOUT:300,# 默认过期时间秒 CACHE_KEY_PREFIX:flask_cache_,# 缓存键的前缀避免与其他应用冲突}# -------------------- 配置 CFileSystemCache单服务器场景--------------------# 使用文件系统存储缓存适合单机部署且不方便使用 Redis 的场景# 确保缓存目录有读写权限FILESYSTEM_CACHE_CONFIG{CACHE_TYPE:FileSystemCache,# 缓存类型FileSystemCache CACHE_DIR:./cache_data,# 缓存存储的目录路径 CACHE_DEFAULT_TIMEOUT:300,# 默认过期时间秒 CACHE_THRESHOLD:500,# 最大缓存文件数 CACHE_IGNORE_ERRORS:True,# 遇到错误时是否忽略}# 选择并应用配置 # 这里使用 SimpleCache 作为演示无需额外安装开箱即用# 如果安装了 Redis可以把下面这行注释掉改使用REDIS_CACHE_CONFIGapp.config.from_mapping(SIMPLE_CACHE_CONFIG)# 也可以从环境变量动态决定使用哪个配置生产环境推荐# import os# if os.environ.get(CACHE_TYPE) redis:# app.config.from_mapping(REDIS_CACHE_CONFIG)# else:# app.config.from_mapping(SIMPLE_CACHE_CONFIG)# 3. 初始化缓存实例 cacheCache(app)# 传入 app 对象自动读取 app.config 中的缓存配置# 也可以使用延迟初始化方式适合应用工厂模式# cache Cache() # 先不传 app# cache.init_app(app) # 后面再初始化# 4. 路由缓存使用示例 # ---------- 示例 1简单的视图缓存 ----------app.route(/)cache.cached(timeout60)# 缓存此视图60秒defindex(): 使用 cache.cached 装饰器缓存视图的返回结果-第一次访问时执行函数结果存入缓存-60秒内的后续访问直接从缓存返回绕过函数体-默认缓存键基于请求的 URL path 生成 current_timetime.strftime(%Y-%m-%d%H:%M:%S)print(f[INDEX] 函数被执行了当前时间{current_time})# 只有第一次或过期后才会打印returnrender_template_string(htmlbodyh1Flask-Caching 演示/h1p当前时间{{current_time}}/pp如果60秒内刷新页面时间不会变化因为视图被缓存了/phrulliahref/user/1查看用户1缓存30秒-使用 key_prefix/a/liliahref/memoized/5斐波那契计算memoize-演示函数级缓存/a/liliahref/manual手动缓存示例/a/liliahref/delete/user/1删除 user/1的缓存/a/liliahref/clear-all清空全部缓存/a/li/ul/body/html,current_timecurrent_time)# ---------- 示例 2带参数的视图缓存 ----------app.route(/user/int:user_id)cache.cached(timeout30,key_prefixuser_view)# key_prefix 指定缓存键前缀defget_user(user_id): 有参数的视图缓存-key_prefix缓存键前缀不指定则默认为 view/request.path-当 query_stringTrue 时缓存键会包含 URL 查询参数适合 GET 请求print(f[USER] 查询用户 {user_id}执行了数据库查询演示)# 模拟数据库查询time.sleep(0.5)# 模拟耗时操作 user_data{id:user_id,name:f用户{user_id},cached_at:time.strftime(%Y-%m-%d%H:%M:%S)}returnjsonify(user_data)# ---------- 示例 3函数级缓存memoize- 缓存任意函数的返回值 ----------app.route(/fib/int:n)deffib_demo(n):演示 memoize 函数缓存 resultexpensive_fibonacci(n)returnjsonify({n:n,result:result})cache.memoize(timeout60)# memoize 根据函数参数自动生成缓存键defexpensive_fibonacci(n): 计算斐波那契数列的第 n 项递归版本非常耗时 使用 cache.memoize 装饰器根据参数 n 的值缓存结果-第一次调用传入 n35时花费时间计算并缓存-后续再调用expensive_fibonacci(35)时直接从缓存返回-注意参数必须是可哈希hashable的类型常见类型如int、str、tuple 都可以print(f[FIB] 正在计算斐波那契 fib({n})... 这很耗时)ifn1:returnnreturnexpensive_fibonacci(n-1)expensive_fibonacci(n-2)# ---------- 示例 4条件缓存unless----------app.route(/admin)cache.cached(timeout60,unlesslambda:notis_admin())defadmin_panel(): unless 参数如果返回 True则跳过缓存 这里只有当用户是管理员时才缓存普通用户每次都要重新计算 实际项目中可以从 session 或 request 中判断用户角色returnjsonify({role:admin,data:敏感数据})defis_admin():模拟判断当前用户是否是管理员# 实际项目中从 request.headers 或 session 中获取# 这里简单演示可以通过 URL 参数 ?admin1 模拟returnrequest.args.get(admin)1# ---------- 示例 5手动缓存控制get/set/delete----------app.route(/manual,methods[GET,POST])defmanual_cache_demo(): 手动控制缓存不依赖装饰器直接使用 cache.get()和 cache.set()这种方式的优点是可以完全控制缓存逻辑比如-POST 请求cache.cached 默认基于 URL无法区分 body 内容-需要根据请求体内容决定缓存键-需要更精细的缓存粒度控制ifrequest.methodPOST:# 从 POST 请求中获取数据datarequest.jsonor{}user_iddata.get(user_id)ifnotuser_id:returnjsonify({error:需要提供 user_id}),400# 构造缓存键手动方式更灵活cache_keyfuser_data_{user_id}# 检查缓存是否存在cached_datacache.get(cache_key)ifcached_dataisnotNone:# 缓存命中返回缓存数据并标记来源resultjson.loads(cached_data)ifisinstance(cached_data,str)elsecached_data result[from_cache]Truereturnjsonify(result)# 缓存未命中查询数据库或调用APIprint(f[MANUAL] 缓存未命中查询用户 {user_id} 的数据)# 模拟耗时操作time.sleep(1)user_info{user_id:user_id,name:f用户{user_id},data:f这是用户{user_id}的详细信息,fetched_at:time.strftime(%Y-%m-%d%H:%M:%S),from_cache:False}# 存入缓存设置 120 秒过期# 注意存储到 Redis 的值会被序列化建议先转成字符串cache.set(cache_key,json.dumps(user_info),timeout120)returnjsonify(user_info)# GET 请求返回使用说明页面returnrender_template_string(htmlbodyh2手动缓存示例/h2p使用 POST 请求发送 JSON 数据例如/pprecurl-XPOSThttp://localhost:5000/manual \\-HContent-Type: application/json\\-d {user_id:1}/prep第一次请求会耗时查询逻辑第二次请求从缓存返回。/phrformmethodpostaction/manualstyledisplay: inline;inputtypehiddennameuser_idvalue99buttontypesubmit测试手动缓存user_id99/button/form/body/html)# ---------- 示例 6删除指定缓存 ----------app.route(/delete/user/int:user_id)defdelete_user_cache(user_id): 删除指定用户的缓存 使用 cache.delete(key)方法# 方式 1删除视图缓存需要知道缓存的键名# cache.cached 默认生成的 cache_key 格式为view/ request.path# 对于 /user/1 这个路径cache_key 是 view//user/1cache_keyfview//user/{user_id}deletedcache.delete(cache_key)# 方式 2如果使用了 memoize需要用 delete_memoized# cache.delete_memoized(expensive_fibonacci, n35) # 删除特定参数的缓存# cache.delete_memoized(expensive_fibonacci) # 删除该函数的所有缓存returnjsonify({status:deletedifdeletedelsenot_foundoralready expired,cache_key:cache_key})# ---------- 示例 7清理所有缓存 ----------app.route(/clear-all)defclear_all_cache(): 清空整个缓存谨慎使用 生产环境建议只用于管理员接口 cache.clear()returnjsonify({status:all cache cleared})# ---------- 示例 8获取缓存统计信息 ----------app.route(/cache-stats)defcache_stats(): 获取缓存统计信息 注意SimpleCache 和 Redis 的部分统计方法可能不可用# 获取当前缓存实例的类型cache_typetype(cache.cache).__name__# 尝试获取缓存大小SimpleCache 支持cache_sizeNoneifhasattr(cache.cache,_cache):cache_sizelen(cache.cache._cache)returnjsonify({cache_type:cache_type,cache_size:cache_size,info:SimpleCache 不使用 Redis如需详细统计请改用 RedisCache})# ---------- 示例 9POST 请求的手动缓存复杂场景----------app.route(/api/search,methods[POST])defsearch_api(): 演示 POST 请求的手动缓存 cache.cached 默认基于 URL path 生成缓存键无法区分 POST body 内容 因此 POST 请求需要手动实现缓存逻辑 datarequest.jsonor{}# 根据请求体内容生成唯一的缓存键# 注意字典的键顺序可能不同需要排序以确保相同内容生成相同键querydata.get(query,)pagedata.get(page,1)# 构造缓存键先排序确保一致性cache_keyfsearch:query{query}page{page}# 尝试从缓存获取cached_resultcache.get(cache_key)ifcached_resultisnotNone:returnjsonify({from_cache:True,data:json.loads(cached_result)ifisinstance(cached_result,str)elsecached_result,cached_at:None # 可以单独存储})# 模拟执行搜索耗时操作print(f[SEARCH] 执行搜索query{query}, page{page})importrandomtime.sleep(1)# 模拟搜索延迟# 生成搜索结果results{query:query,page:page,total:random.randint(10,100),items:[f结果{i1}foriinrange(5)],searched_at:time.strftime(%Y-%m-%d%H:%M:%S)}# 存入缓存超时时间 60 秒cache.set(cache_key,json.dumps(results),timeout60)returnjsonify({from_cache:False,data:results})# 5. 启动应用 if__name____main__:print(*50)print(Flask-Caching 演示应用)print(f缓存类型: {app.config.get(CACHE_TYPE, 未配置)})print(f缓存默认超时: {app.config.get(CACHE_DEFAULT_TIMEOUT, 300)} 秒)print(*50)app.run(debugTrue,port5000)》》测试命令# 1. 启动应用python app.py# 2. 测试视图缓存首次访问会打印日志curlhttp://localhost:5000/# 3. 测试带参数的视图缓存curlhttp://localhost:5000/user/1# 第一次缓存未命中执行函数curlhttp://localhost:5000/user/1# 第二次缓存命中不执行函数# 4. 测试 memoize 函数缓存curlhttp://localhost:5000/fib/35# 第一次计算耗时curlhttp://localhost:5000/fib/35# 第二次直接从缓存返回# 5. 测试手动缓存POST 请求curl-XPOSThttp://localhost:5000/manual-HContent-Type: application/json-d {user_id:1} curl-XPOSThttp://localhost:5000/manual-HContent-Type: application/json-d {user_id:1}# 6. 测试删除缓存curlhttp://localhost:5000/delete/user/1# 7. 清除全部缓存curlhttp://localhost:5000/clear-all