Как унаследовать метод класса django из views.py в tasks.py для планирования заданий с помощью Celery
Я пытаюсь в django с помощью Celery планировать задания. Проблема в том, что я не могу заставить мою функцию @shared_task в tasks.py наследовать метод класса из views.py. Я установил Celery вместе с Flower и Beat, и конфигурация работает хорошо, когда я пишу просто отдельную функцию в tasks.py. Есть ли у вас какие-нибудь предложения, как решить эту проблему с наследованием? Спасибо за помощь.
my views.py
from rest_framework import generics
from .serializers import TickerSerializer
from ticker.models import Ticker
import requests
import logging
from itertools import chain
from importlib import reload
import sys
logger = logging.getLogger(__name__)
class TickerList(generics.ListAPIView):
serializer_class = TickerSerializer
queryset = Ticker.objects.all()
class TickerRetrieve(generics.RetrieveAPIView):
serializer_class = TickerSerializer
queryset = Ticker.objects.all()
lookup_field = 'id'
def get_object(request, **args):
url = 'https://api.kraken.com/0/public/Ticker?pair=1INCHEUR,1INCHUSD'
response = requests.get(url)
data = response.json()
Ticker.objects.all().delete()
for key in data['result']:
if isinstance(data['result'][key], int):
continue
crypto_data =data['result'][key]
ticker_data = Ticker(crypto = (key),
a_0 = (crypto_data.get('a')[0]),
a_1 = (crypto_data.get('a')[1]),
)
ticker_data.save()
my tasks.py
from __future__ import absolute_import, unicode_literals
from celery.utils.log import get_task_logger
import sys
from celery import shared_task
from ticker.views import TickerRetrieve
from django.core.management import call_command
a = TickerRetrieve()
@shared_task(name="updatemy_ticker")
def updatemy_ticker_task(ticker_data):
return a.get_object(ticker_data)
На Flower и локальных серверах запланированные задачи работают без ошибок, однако я не могу заставить tasks.py запустить метод класса в views.py. Я просмотрел несколько примеров здесь и прочитал документацию, но ни один из них не добавляет аналогичный пример.
Решением стало возвращение имени функции a.get_object() без аргументов в tasks.py Теперь автоматизированная задача работает как часы.
tasks.py
from __future__ import absolute_import, unicode_literals
from celery.utils.log import get_task_logger
import sys
from celery import shared_task
from ticker.views import TickerRetrieve
from django.core.management import call_command
a = TickerRetrieve()
@shared_task(name="updatemy_ticker")
def updatemy_ticker_task():
return a.get_object()