flask_caching.memoize()与Dash background_callback_manager缓存兼容问题

问题现象 #

原本正常工作的非回调函数memoize缓存，在引入background_callback_manager后出现异常：同一回调内两次调用该函数可复用缓存，但再次触发回调时必须重新生成缓存。推测缓存键中混入了会话变量，但对缓存后端机制不熟悉，相关文档匮乏。同时使用两套缓存系统是为了实现dash.callback的progress参数功能。当前缓存配置代码如下：

import uuid
from uuid import uuid4
import os
from dash import Dash
from dash.long_callback import CeleryManager, DiskcacheManager

launch_uid = uuid4()

EXPIRE = 600 

if "REDIS_URL" in os.environ:
    # 生产环境使用Redis + Celery
    from celery import Celery

    celery_app = Celery(__name__, broker=os.environ['REDIS_URL'], backend=os.environ['REDIS_URL'])
    background_callback_manager = CeleryManager(
        celery_app, cache_by=[lambda: launch_uid], expire=EXPIRE
    )

else:
    # 本地开发使用Diskcache
    import diskcache

    cache = diskcache.Cache("./cache")
    background_callback_manager = DiskcacheManager(
        cache, cache_by=[lambda: launch_uid], expire=EXPIRE
    )

# 初始化Dash应用
app = Dash(
    __name__,
    background_callback_manager=background_callback_manager
)

# Flask缓存配置（用于非回调函数memoize）
from flask_caching import Cache
flask_cache = Cache(
    app.server,
    config={
        "CACHE_TYPE": "RedisCache",
        "CACHE_TYPE": "FileSystemCache",
        "CACHE_DIR": "cache-directory",
        "CACHE_THRESHOLD": 200,
        "CACHE_DEFAULT_TIMEOUT": 0,
    },
)

问题根源 #

1. 两套独立缓存系统隔离：同时配置了用于后台回调的CeleryManager/DiskcacheManager和用于Flask的flask_cache，两者使用不同的缓存存储或上下文，导致缓存无法跨系统共享。
2. 后台回调执行环境隔离：后台回调运行在独立进程/worker中，Flask默认的memoize缓存键生成逻辑会混入进程上下文或会话相关变量，跨进程/回调时生成的键不一致，无法命中原有缓存。

解决方案 #

方案1：统一缓存后端，共用同一存储 #

让Flask缓存和后台回调使用同一缓存后端，消除隔离：

生产环境（Redis）

修改flask_cache配置，直接复用Redis连接：

flask_cache = Cache(
    app.server,
    config={
        "CACHE_TYPE": "RedisCache",
        "CACHE_REDIS_URL": os.environ['REDIS_URL'],
        "CACHE_THRESHOLD": 200,
        "CACHE_DEFAULT_TIMEOUT": 0,
    },
)

开发环境（Diskcache）

复用后台回调的diskcache实例作为Flask缓存的底层存储：

# 先初始化diskcache实例
import diskcache
cache = diskcache.Cache("./cache")
# ... 后台回调manager配置 ...

flask_cache = Cache(
    app.server,
    config={
        "CACHE_TYPE": "SimpleCache",
        "CACHE_DEFAULT_TIMEOUT": 0,
    }
)
# 替换SimpleCache的底层存储为diskcache
flask_cache.cache._cache = cache

方案2：自定义缓存键生成规则，排除会话/上下文变量 #

通过memoize的make_name参数，自定义只基于函数参数生成缓存键的逻辑，避免混入上下文相关内容：

def custom_cache_key(f, **kwargs):
    # 仅用函数名、位置参数和关键字参数生成缓存键
    args = kwargs.get('args', ())
    kwargs_sorted = sorted(kwargs.get('kwargs', {}).items())
    return f"{f.__name__}_{str(args)}_{str(kwargs_sorted)}"

# 使用自定义键生成器装饰非回调函数
@flask_cache.memoize(make_name=custom_cache_key)
def your_non_callback_func(param1, param2):
    # 函数业务逻辑
    pass

方案3：确保多进程环境下launch_uid全局一致 #

如果是多进程部署（如Gunicorn多worker），需保证所有进程使用相同的launch_uid：

• 启动时统一生成launch_uid并写入共享存储（如Redis），所有进程从共享存储读取
• 或通过启动参数传递launch_uid给每个worker，避免各进程独立生成不同的ID导致缓存隔离

内容的提问来源于stack exchange，提问作者doyle