如何在装饰器下运行Celery-Beat任务？解决任务未注册KeyError问题

这个错误的核心原因是装饰器顺序错误加上自定义装饰器没有保留原函数元信息，导致Celery无法正确识别和注册你的任务。让我一步步帮你解决：

1. 调整装饰器顺序 #

Celery的@celery_app.task装饰器必须直接作用于你的任务函数，也就是要放在自定义装饰器@lock_task的下方（靠近函数的一侧）。装饰器是从下往上执行的，如果顺序反了，Celery会把lock_task返回的wrapper函数注册为任务，而不是你真正的test函数，自然找不到指定的任务名。

修改后的任务代码：

@lock_task
@celery_app.task  # 这个装饰器要紧贴任务函数
def test():
    call_command('test')

2. 给自定义装饰器添加元信息保留 #

你的lock_task装饰器返回的wrapper函数会覆盖原函数的名字等元信息，这会导致Celery注册的任务名变成wrapper而不是test，Beat调用时自然找不到project.celery.test。我们需要用functools.wraps来保留原函数的元数据。

修改后的装饰器代码：

from functools import wraps
import redis

def lock_task(func):
    @wraps(func)  # 关键：保留原函数的名称、文档字符串等元信息
    def wrapper(*args, **kwargs):
        if redis.set(func.__name__, 'lock', nx=True):
            try:
                result = func(*args, **kwargs)
            finally:
                redis.delete(func.__name__)
            return result or True
        else:
            return 'Skipped'
    return wrapper

3. 额外检查：确保任务模块被Celery Worker导入 #

如果调整完装饰器还是报错，要确认Celery Worker能正确加载你的任务模块。可以在Celery配置里添加imports选项，明确指定任务所在的模块：

celery_app.conf.imports = ('project.celery',)  # 替换成你实际的任务模块路径

或者在启动Worker时通过--app参数指定模块，比如：

celery -A project.celery worker --beat --loglevel=info

这样调整后，Celery就能正确注册test任务，Beat调度时也能找到对应的任务了。

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