Как сохранить имя задачи в расширенной модели django_celery_result - PullRequest
Новая
       52

Как сохранить имя задачи в расширенной модели django_celery_result

0 голосов
/ 30 августа 2018

Это дерево моего проекта. 

└── 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. Я расширил модель DjangoCeleryResultsTaskresult из django_celery_results в файле etl / models.py. 

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()

Поэтому я хочу сохранить имя задачи для каждой задачи в файле celery.py, когда одна из задач выполняется с ее идентификатором_ задачи и именем-задачи в расширенной модели TaskResultsExtension. Я использую python3 и django 1.11 celery4.2.1 django-celery-beat1.1.1 django-celery-results1.0.1. Пожалуйста, помогите мне с этой проблемой.

1 Ответ

0 голосов
/ 30 августа 2018

Вы должны создать свой собственный класс и обрабатывать 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

Вы можете найти больше в Исходный код для celery.app.task

...