如何在扩展的django_celery_result模型中保存任务名称

Question

这是我项目的树。

└── elt-ui2
├── Etl_ui
│   ├── celerybeat.pid
│   ├── celerybeat-schedule
│   ├── celery_tasks
│   │   ├── admin.py
│   │   ├── apps.py
│   │   ├── __init__.py
│   │   ├── migrations
│   │   │   ├── __init__.py
│   │   │   └── __pycache__
│   │   │       └── __init__.cpython-35.pyc
│   │   ├── models.py
│   │   ├── __pycache__
│   │   │   ├── admin.cpython-35.pyc
│   │   │   ├── __init__.cpython-35.pyc
│   │   │   └── models.cpython-35.pyc
│   │   ├── scripts
│   │   │   ├── __pycache__
│   │   │   │   ├── ssl_extract.cpython-35.pyc
│   │   │   │   └── ssl_transform.cpython-35.pyc
│   │   │   ├── ssl_extract.py
│   │   │   └── ssl_transform.py
│   │   ├── tests.py
│   │   └── views.py
│   ├── etl
│   │   ├── admin.py
│   │   ├── apps.py
│   │   ├── __init__.py
│   │   ├── migrations
│   │   │   ├── 0001_initial.py
│   │   │   ├── 0002_djangoceleryresultstaskresult_taskresultsextension.py
│   │   │   ├── __init__.py
│   │   │   └── __pycache__
│   │   │       ├── 0001_initial.cpython-35.pyc
│   │   │       ├── 0002_djangoceleryresultstaskresult_taskresultsextension.cpython-35.pyc
│   │   │       └── __init__.cpython-35.pyc
│   │   ├── models.py
│   │   ├── __pycache__
│   │   │   ├── admin.cpython-35.pyc
│   │   │   ├── __init__.cpython-35.pyc
│   │   │   ├── models.cpython-35.pyc
│   │   │   ├── urls.cpython-35.pyc
│   │   │   └── views.cpython-35.pyc
│   │   ├── tasks.py
│   │   ├── templates
│   │   │   ├── Etl_status.html
│   │   │   ├── Home_page.html
│   │   │   ├── login_page.html
│   │   │   └── Upload_data.html
│   │   ├── tests.py
│   │   ├── urls.py
│   │   └── views.py
│   ├── Etl_ui
│   │   ├── celery_app.py
│   │   ├── celeryconfig.py
│   │   ├── __init__.py
│   │   ├── __pycache__
│   │   │   ├── celery_app.cpython-35.pyc
│   │   │   ├── celeryconfig.cpython-35.pyc
│   │   │   ├── __init__.cpython-35.pyc
│   │   │   ├── settings.cpython-35.pyc
│   │   │   ├── urls.cpython-35.pyc
│   │   │   └── wsgi.cpython-35.pyc
│   │   ├── settings.py
│   │   ├── urls.py
│   │   └── wsgi.py
│   ├── geckodriver.log
│   ├── manage.py
│   └── requirement.txt
└── README.md

我在Etl_ui / celery_app.py中有两个任务。

from __future__ import absolute_import, unicode_literals
from celery import Celery
from celery.schedules import crontab
import os
from . import celeryconfig
from django.conf import settings
from celery_tasks import scripts
# from django.db import models
# from django_celery_results.models import TaskResult

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'Etl_ui.settings')

app = Celery('Etl_ui')
app.config_from_object(celeryconfig)

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
     @app.on_after_configure.connect
    def setup_periodic_tasks(sender, **kwargs):
        sender.add_periodic_task(900.0,task_ssl_extract.s(),name='add every 15 minmutes')
        sender.add_periodic_task(300.0,task_ssl_transform.s(),name='add every 5 minutes')
        #sender.add_periodic_task(30.0, test2.s('world'), expires=10)

    @app.task(max_retries=5,default_retry_delay=300,name='ssl_extract')
    def task_ssl_extract():
        from celery_tasks.scripts.ssl_extract import main
        main(username, password, brandname, client_name, partner, path)

    @app.task(name='ssl_transform')
    def task_ssl_transform():
        from celery_tasks.scripts.ssl_transform import main
        main(input_file, output_file, url, username, password, error_file)

我使用django_celery_results作为后端，它使用许多字段保存任务结果。但是这个模型没有task_name列。我在etl / models.py中扩展了django_celery_results的模型DjangoCeleryResultsTaskresult。

class TaskResultsExtension(models.Model):
task_name = models.CharField(max_length=255)
task = models.OneToOneField(DjangoCeleryResultsTaskresult,on_delete=models.CASCADE)

我在etl / tasks.py中创建了一个任务。

from __future__ import absolute_import, unicode_literals
from celery.decorators import task
from .models import TaskResultsExtension, DjangoCeleryResultsTaskresult

@task(name="save the new task")
def save_task():
    task_result = DjangoCeleryResultsTaskresult.objects.all()

因此，当其中一个任务在扩展模型TaskResultsExtension中运行task_id和task_name时，我想在celery.py中保存每个任务的任务名称。我正在使用python3和django 1.11 celery4.2.1 django-celery-beat1.1.1 django-celery-results1.0.1。请帮我解决这个问题。

Answer 1

您应该创建自定义类并处理on_success和on_failure案例。基本上，你需要一个班级：

from celery import Task


class MyCalbackTask(Task):

    def run(self, *args, **kwargs):
        # The body of the task executed by workers. Required.
        pass

    def on_success(self, retval, task_id, *args, **kwargs):
        # do something with usefull values as retval and task_id
        pass

    def on_failure(self, exc, task_id, args, kwargs, einfo):
        # do something
        pass

然后你只需要使用这个类作为基础：

@app.task(base=MyCallbackTask)
def my_dear_task(**kwargs):
    # task code

你可以在Source code for celery.app.task找到更多