如何让dogpile捕获pickle序列化异常并返回函数实际值？

以下代码尝试用dogpile.cache缓存返回lambda的函数结果，会触发pickle序列化错误：

from dogpile.cache import make_region
import logging

log = logging.getLogger(__name__)
TIMEOUT_SECONDS = 10 * 60


def my_key_generator(namespace, fn):
    fname = fn.__name__

    def generate_key(*arg):
        key_template = fname + "_" + "_".join(str(s) for s in arg)
        return key_template

    return generate_key


region = make_region(function_key_generator=my_key_generator).configure(
    "dogpile.cache.redis",
    expiration_time=TIMEOUT_SECONDS,
    arguments={
        "host": "localhost",
        "port": 6379,
        "db": 0,
        "redis_expiration_time": TIMEOUT_SECONDS * 2,  # 2 hours
        "distributed_lock": True,
        "thread_local_lock": False,
    },

)


@region.cache_on_arguments()
def load_user_info(user_id):
    log.info(f"Called func {user_id}")
    return lambda: user_id

print(load_user_info(1))

对应的错误栈：

Traceback (most recent call last):
  File "/Users/user/src/python-test/dogpile-test.py", line 120, in <module>
    print(load_user_info(1))
          ^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/lib/python3.11/site-packages/decorator.py", line 232, in fun
    return caller(func, *(extras + args), **kw)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/cache/region.py", line 1577, in get_or_create_for_user_func
    return self.get_or_create(
           ^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/cache/region.py", line 1042, in get_or_create
    with Lock(
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/lock.py", line 185, in __enter__
    return self._enter()
           ^^^^^^^^^^^^^
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/lock.py", line 94, in _enter
    generated = self._enter_create(value, createdtime)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/lock.py", line 178, in _enter_create
    return self.creator()
           ^^^^^^^^^^^^^^
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/cache/region.py", line 1012, in gen_value
    self._set_cached_value_to_backend(key, value)
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/cache/region.py", line 1288, in _set_cached_value_to_backend
    key, self._serialized_cached_value(value)
         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/cache/region.py", line 1258, in _serialized_cached_value
    return self._serialize_cached_value_elements(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/opt/homebrew/lib/python3.11/site-packages/dogpile/cache/region.py", line 1232, in _serialize_cached_value_elements
    serializer(payload),
    ^^^^^^^^^^^^^^^^^^^
AttributeError: Can't pickle local object 'load_user_info.<locals>.<lambda>'

用户问题：是否存在方法让dogpile捕获该pickle序列化异常，记录异常日志并返回函数的实际值，而非导致整个函数调用失败？

方案一：自定义安全缓存装饰器 #

通过包装dogpile原生装饰器，在调用时捕获序列化异常，记录日志后直接执行原函数返回结果：

import logging
import pickle
from dogpile.cache import make_region

log = logging.getLogger(__name__)
TIMEOUT_SECONDS = 10 * 60


def my_key_generator(namespace, fn):
    fname = fn.__name__
    def generate_key(*arg):
        key_template = fname + "_" + "_".join(str(s) for s in arg)
        return key_template
    return generate_key


region = make_region(function_key_generator=my_key_generator).configure(
    "dogpile.cache.redis",
    expiration_time=TIMEOUT_SECONDS,
    arguments={
        "host": "localhost",
        "port": 6379,
        "db": 0,
        "redis_expiration_time": TIMEOUT_SECONDS * 2,
        "distributed_lock": True,
        "thread_local_lock": False,
    },
)

def safe_cache_on_arguments():
    def decorator(func):
        cached_func = region.cache_on_arguments()(func)
        def wrapper(*args, **kwargs):
            try:
                return cached_func(*args, **kwargs)
            except Exception as e:
                # 精准匹配序列化相关异常
                if isinstance(e, (pickle.PicklingError, AttributeError)) or "Can't pickle" in str(e):
                    log.error(f"缓存序列化失败: {str(e)}")
                    # 跳过缓存，直接返回原函数结果
                    return func(*args, **kwargs)
                # 非序列化异常正常抛出
                raise
        return wrapper
    return decorator

@safe_cache_on_arguments()
def load_user_info(user_id):
    log.info(f"Called func {user_id}")
    return lambda: user_id

print(load_user_info(1))

方案二：重写CacheRegion类处理异常 #

继承dogpile的CacheRegion类，重写缓存值到后端的方法，内部捕获序列化异常：

import logging
import pickle
from dogpile.cache import make_region
from dogpile.cache.region import CacheRegion

log = logging.getLogger(__name__)
TIMEOUT_SECONDS = 10 * 60


class SafeCacheRegion(CacheRegion):
    def _set_cached_value_to_backend(self, key, value):
        try:
            super()._set_cached_value_to_backend(key, value)
        except Exception as e:
            # 识别序列化异常并记录日志
            if isinstance(e, (pickle.PicklingError, AttributeError)) or "Can't pickle" in str(e):
                log.error(f"缓存存储失败，key: {key}, 错误信息: {str(e)}")
            else:
                # 其他异常继续抛出
                raise


def my_key_generator(namespace, fn):
    fname = fn.__name__
    def generate_key(*arg):
        key_template = fname + "_" + "_".join(str(s) for s in arg)
        return key_template
    return generate_key


# 使用自定义Region类
region = SafeCacheRegion(function_key_generator=my_key_generator).configure(
    "dogpile.cache.redis",
    expiration_time=TIMEOUT_SECONDS,
    arguments={
        "host": "localhost",
        "port": 6379,
        "db": 0,
        "redis_expiration_time": TIMEOUT_SECONDS * 2,
        "distributed_lock": True,
        "thread_local_lock": False,
    },
)

@region.cache_on_arguments()
def load_user_info(user_id):
    log.info(f"Called func {user_id}")
    return lambda: user_id

print(load_user_info(1))

补充方案：替换序列化器解决根本问题 #

如果只是想让lambda这类对象可以被序列化，可使用dill替代默认pickle，既正常缓存又不抛出异常：

import dill
from dogpile.cache import make_region
import logging

log = logging.getLogger(__name__)
TIMEOUT_SECONDS = 10 * 60

def my_key_generator(namespace, fn):
    fname = fn.__name__
    def generate_key(*arg):
        key_template = fname + "_" + "_".join(str(s) for s in arg)
        return key_template
    return generate_key

region = make_region(function_key_generator=my_key_generator).configure(
    "dogpile.cache.redis",
    expiration_time=TIMEOUT_SECONDS,
    arguments={
        "host": "localhost",
        "port": 6379,
        "db": 0,
        "redis_expiration_time": TIMEOUT_SECONDS * 2,
        "distributed_lock": True,
        "thread_local_lock": False,
        # 指定dill作为序列化器
        "serializer": dill.dumps,
        "deserializer": dill.loads,
    },
)

@region.cache_on_arguments()
def load_user_info(user_id):
    log.info(f"Called func {user_id}")
    return lambda: user_id

print(load_user_info(1))

注意：使用dill需先安装pip install dill。

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