如何解决无法在 Django Celery Beat 中获取 kwargs 数据
我正在使用 Django celery-beat 在应用程序中实现一个提醒模块,我正在周期性任务中创建 cron tab 并在 kwargs 参数中传递字典。它已成功保存在 Django 周期任务表中,但是当计划任务按时运行并调用上述函数时,它不会获取 kwargs 数据并通过异常。
settings.py
INSTALLED_APPS = [
'django_celery_beat',]
# CELERY STUFF
CELERY_BROKER_URL = "redis://localhost:6379"
CELERY_RESULT_BACKEND = "redis://localhost:6379"
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
init.py
from .celery import app as celery_app
all = ("celery_app",)
celery.py
import os
from celery import Celery
os.environ.setdefault("DJANGO_SETTINGS_MODULE","config.settings_local")
app = Celery("baseapp")
app.config_from_object("django.conf:settings",namespace="CELERY")
app.autodiscover_tasks()
这是创建周期性任务的任务调度器功能:
def task_scheduler(raw_data):
try:
if raw_data["start"] and raw_data["reminder"]:
reminder_time = datetime.datetime.strptime(raw_data["start"],'%Y-%m-%dT%H:%M:%S.%fZ') - datetime.timedelta(minutes=raw_data["reminder"])
reminder,_ = CrontabSchedule.objects.get_or_create(
minute=reminder_time.minute,hour=reminder_time.hour,day_of_week="*",day_of_month=reminder_time.day,month_of_year=reminder_time.month,)
PeriodicTask.objects.update_or_create(
name=raw_data["summary"],task="reminder_notification",crontab=reminder,expire_seconds=7200,kwargs=json.dumps({'test132':'123'}),)
except Exception as error:
print(error)
这是一个提醒通知函数,当任务按时运行时调用:
@shared_task(name="reminder_notification")
def reminder_notification(*args,**kwargs):
print("hello task")
这里是错误:
[2021-01-07 06:13:00,021: ERROR/Beat] Message Error: Couldn't apply scheduled task celery15: reminder_notification() got an unexpected keyword argument 'test132'
[' File "/home/usman/Documents/venv/bin/celery",line 8,in <module>\n sys.exit(main())\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/__main__.py",line 15,in main\n sys.exit(_main())\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/bin/celery.py",line 213,in main\n return celery(auto_envvar_prefix="CELERY")\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/click/core.py",line 829,in __call__\n return self.main(*args,**kwargs)\n',line 782,in main\n rv = self.invoke(ctx)\n',line 1259,in invoke\n return _process_result(sub_ctx.command.invoke(sub_ctx))\n',line 1066,in invoke\n return ctx.invoke(self.callback,**ctx.params)\n',line 610,in invoke\n return callback(*args,' File "/home/usman/Documents/venv/lib/python3.8/site-packages/click/decorators.py",line 21,in new_func\n return f(get_current_context(),*args,' File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/bin/base.py",line 132,in caller\n return f(ctx,' File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/bin/worker.py",line 327,in worker\n worker.start()\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/worker/worker.py",line 203,in start\n self.blueprint.start(self)\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/bootsteps.py",line 116,in start\n step.start(parent)\n',line 365,in start\n return self.obj.start()\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/billiard/process.py",line 124,in start\n self._popen = self._Popen(self)\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/billiard/context.py",line 276,in _Popen\n return _default_context.get_context().Process._Popen(process_obj)\n',line 333,in _Popen\n return Popen(process_obj)\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/billiard/popen_fork.py",line 24,in __init__\n self._launch(process_obj)\n',line 79,in _launch\n code = process_obj._bootstrap()\n',in _bootstrap\n self.run()\n',' File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/beat.py",line 703,in run\n self.service.start(embedded_process=True)\n',line 627,in start\n interval = self.scheduler.tick()\n',line 339,in tick\n self.apply_entry(entry,producer=self.producer)\n',line 272,in apply_entry\n exc,traceback.format_stack(),exc_info=True)\n']
Traceback (most recent call last):
File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/beat.py",line 386,in apply_async
return task.apply_async(entry_args,entry_kwargs,File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/app/task.py",line 526,in apply_async
check_arguments(*(args or ()),**(kwargs or {}))
TypeError: reminder_notification() got an unexpected keyword argument 'test132'
During handling of the above exception,another exception occurred:
Traceback (most recent call last):
File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/beat.py",line 269,in apply_entry
result = self.apply_async(entry,producer=producer,advance=False)
File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/beat.py",line 394,in apply_async
reraise(SchedulingError,SchedulingError(
File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/exceptions.py",line 104,in reraise
raise value.with_traceback(tb)
File "/home/usman/Documents/venv/lib/python3.8/site-packages/celery/beat.py",**(kwargs or {}))
celery.beat.SchedulingError: Couldn't apply scheduled task celery15: reminder_notification() got an unexpected keyword argument 'test132'
我做错了什么?
解决方法
我正在成功获取kwargs 数据。
解决方案:
当前运行的worker没有从函数中获取更新的变化,所以我们需要停止worker并再次运行。
celery -A baseapp worker --beat --scheduler django --loglevel=info
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。