Hallo Habr! Ich werde Ihnen die Geschichte meines professionellen Burnouts erzählen.

Es ist so passiert, dass ich Routine-Laufbänder hasse. Ich habe mehrere Projekte mit Sellerie . Jedes Mal, wenn eine Aufgabe komplizierter als 2 + 2 = 5 , wird die Lösungsvorlage darauf reduziert, eine Klasse zu erstellen, die die Aufgabe ausführt, und eine Starterfunktion , mit der Sellerie arbeiten kann - eine Boilerplate. In diesem Artikel werde ich Ihnen erzählen, wie ich mit einem Boilerplate gekämpft habe und was daraus geworden ist.

Ausgangspunkt

Betrachten Sie die gewöhnliche Aufgabe von Sellerie. Es gibt eine Klasse, die die Aufgabe ausführt, und eine Starterfunktion, die die Klasse instanziiert und eine ihrer Methoden startet, in der die gesamte Logik der Aufgabe implementiert und die Fehlerbehandlung vererbt wird:

 class MyTask( FirstMixin, SecondMixin, ThirdMixin, ): def main(self): data = self.do_something() response = self.remote_call(data) parsed = self.parser(response) return self.process(parsed) @app.task(bind=True) def my_task(self, arg1, arg2): instance = MyTask( celery_task=self, arg1=arg1, arg2=arg2, ) return instance.full_task() 

Gleichzeitig enthält die Methode full_task einen Aufruf von main , befasst sich jedoch auch mit Fehlerbehandlung, Protokollierung und anderem Unsinn, der nicht direkt mit der Hauptaufgabe zusammenhängt.

Taskclass-Idee

Die Wurzel der Taskklasse liegt eine einfache Idee: In der Basisklasse können Sie eine Methode der Taskklasse definieren, das Verhalten der Starterfunktion darin implementieren und dann Folgendes erben:

 class BaseTask: def __init__(self, **kwargs): for key, value in kwargs.items(): setattr(self, key, value) def full_task(self): try: return self.main() except: self.celery_task.retry(countdown=30) @classmethod def task(cls, task, **kwargs): self = cls( celery_task=celery_task, **kwargs, ) return self.full_task() 

Alle Hilfslangeweile in der Basisklasse gesammelt. Wir werden nicht wieder zu ihr zurückkehren. Wir erkennen die Logik der Aufgabe:

 @app.taskcls(bind=True) class MyTask( BaseTask, FirstMixin, SecondMixin, ThirdMixin, ): def main(self): data = self.do_something() response = self.remote_call(data) parsed = self.parser(response) return self.process(parsed) 

Keine Schale mehr, viel besser. Was ist jedoch mit dem Einstiegspunkt?

 MyTask.task.delay(...) 

MyTask.task verfügt über alle üblichen Task-Methoden: delay , apply_async und im Allgemeinen auch.

Nun die Argumente des Dekorateurs. Es macht besonders Spaß, bind = True in jede Aufgabe zu ziehen. Ist es möglich, Standardargumente durch eine Basisklasse zu übergeben?

 class BaseTask: class MetaTask: bind = True def __init__(self, **kwargs): for key, value in kwargs.items(): setattr(self, key, value) def full_task(self): try: return self.main() except: self.celery_task.retry(countdown=30) @classmethod def task(cls, task, **kwargs): self = cls( celery_task=celery_task, **kwargs, ) return self.full_task() 

Die verschachtelte MetaTask Klasse enthält Standardargumente und steht allen MetaTask Klassen zur Verfügung. Interessanterweise kann es auch vererbt werden:

 class BaseHasTimeout(BaseTask): class MetaTask(BaseTask.MetaTask): timeout = 42 

Argumente, die an den Dekorator @app.taskcls haben die höchste Priorität:

 @app.taskcls(timeout=20) class MyTask( BaseHasTimeout, FirstMixin, SecondMixin, ThirdMixin, ): def main(self): ... 

Infolgedessen beträgt das Zeitlimit für die Aufgabe 20.

Darüber hinaus gehen

In Webanwendungen muss eine Aufgabe häufig aus der Ansicht heraus gestartet werden. Bei hoher Haftung können Sicht und Aufgabe kombiniert werden:

 class BaseViewTask: @classmethod def task(cls, **kwargs): # Somehow init View class manually self = cls(...) return self.celery() @app.taskcls class MyView( BaseViewTask, FirstMixin, SecondMixin, ThirdMixin, APIView, ): queryset = MyModel.objects.all() def get_some_data(self, *args, **kwargs): # common methed return self.queryset.filtert(...) def get(self, request): data = self.get_some_data(request.field) # used in request handling return Response(json.dumps(data)) def post(self, request): self.task.delay(...) return Response(status=201) def celery(self): data = self.get_some_data(...) # also used in background task return self.do_something(data) 

Um eine Kollision von Namen zu vermeiden, heißt die verschachtelte Klasse MetaTask , nicht Meta wie in Django.

Fazit

Diese Funktionalität wird in Celery 4.5 erwartet. Ich habe jedoch auch ein Paket vorbereitet, mit dem Sie heute den taskcls-Dekorateur ausprobieren können. Die Idee des Pakets ist, dass Sie beim Upgrade von Celery auf Version 4.5 den Import entfernen können, ohne eine einzige Codezeile zu ändern.