django-celery-beat：如何从shared_task访问触发的PeriodicTask实例？

1. 访问触发任务的PeriodicTask实例 #

默认Celery Beat不会自动把PeriodicTask实例的信息传给任务，你需要手动将实例的唯一标识（比如ID）加入任务参数，再通过标识查询实例：

• 步骤1：给PeriodicTask添加标识参数
  在后台管理页面或代码配置PeriodicTask时，修改其kwargs字段，加入task_id（或自定义键名），值设为该PeriodicTask的ID。例如：

  {"bar": "your_value", "task_id": "periodic_task_unique_id_here"}
  

• 步骤2：在任务中查询实例
  在任务函数里通过传入的ID查询对应的PeriodicTask实例：

  from celery import shared_task
  from django_celery_beat.models import PeriodicTask
  
  @shared_task
  def foo(**kwargs):
      # 执行原有业务逻辑
      whatever_method(kwargs["bar"])
      
      # 获取并查询触发任务的PeriodicTask实例
      task_id = kwargs.get("task_id")
      if task_id:
          try:
              periodic_task = PeriodicTask.objects.get(id=task_id)
              # 可在此使用periodic_task的属性，如name、schedule等
              print(f"当前任务由PeriodicTask [{periodic_task.name}] 触发")
          except PeriodicTask.DoesNotExist:
              # 处理实例不存在的异常情况
              pass
  

2. 区分同一任务的不同PeriodicTask触发源 #

完全可以识别对应的源PeriodicTask。只要给每个调用同一shared_task的PeriodicTask配置唯一标识参数（如ID、自定义标识）并传入任务，任务执行时通过该标识即可区分触发源：

• 示例：两个PeriodicTask调用foo任务
  • 第一个PeriodicTask的kwargs：{"bar": "value1", "task_id": "task_001"}
  • 第二个PeriodicTask的kwargs：{"bar": "value2", "task_id": "task_002"}
• 在foo任务中，通过kwargs["task_id"]的值就能明确判断本次运行是由哪个PeriodicTask触发的。

如果不想手动配置每个PeriodicTask的标识，也可以通过自定义Celery Beat调度器实现自动注入，但手动添加标识是简单场景下最直接的方案。

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