مرحبا يا هبر! سأخبركم بقصة الإرهاق الاحترافي.

حدث ذلك أنني أكره المطاحن الروتينية. لدي العديد من المشاريع باستخدام الكرفس . في كل مرة تصبح المهمة أكثر تعقيدًا من 2 + 2 = 5 ، يتم تقليل قالب الحل إلى إنشاء فصل دراسي يؤدي المهمة ، ووظيفة بداية تستطيع Celery التعامل معها - وهي عبارة عن لوحة. سأخبرك في هذا المقال كيف واجهت صعوبة في الحصول على الغليان وماذا جاء.

نقطة الانطلاق

النظر في المهمة العادية للكرفس. هناك فئة تقوم بتنفيذ المهمة ، ووظيفة بداية تقوم بإنشائها ، وتبدأ إحدى طرقها ، حيث يتم تنفيذ كل منطق المهمة وتورث معالجة الأخطاء:

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

في الوقت نفسه ، full_task طريقة full_task على استدعاء main ، ومع ذلك ، فهي تتعامل أيضًا مع معالجة الأخطاء وتسجيل الأخطاء وغير ذلك من الهراء الذي لا يرتبط مباشرة بالمهمة الرئيسية.

فكرة Taskclass

تكمن جذر فئة المهام في فكرة بسيطة: في الفئة الأساسية ، يمكنك تحديد طريقة لفئة task ، وتنفيذ سلوك وظيفة المبتدئين فيها ، ثم ترث:

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

جميع الملل المساعدة التي تم جمعها في الفئة الأساسية. لن نعود إليها مرة أخرى. ندرك منطق المهمة:

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

لا مزيد من قشر ، أفضل بكثير. ومع ذلك ، ماذا عن نقطة الدخول؟

 MyTask.task.delay(...) 

يحتوي MyTask.task على كافة أساليب المهمة المعتادة: delay ، و apply_async ، وبشكل عام ، إنه كذلك.

الآن حجج الديكور. من الممتع بشكل خاص سحب bind = True في كل مهمة. هل من الممكن تمرير الوسائط الافتراضية من خلال فئة أساسية؟

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

تحتوي فئة MetaTask المتداخلة على وسيطات افتراضية وستكون متاحة لجميع الفئات الفرعية. ومن المثير للاهتمام ، أنه يمكن أيضًا توريثها:

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

@app.taskcls تم تمريرها إلى @app.taskcls decorator بالأولوية القصوى:

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

نتيجة لذلك ، ستكون مهلة المهمة 20.

الذهاب إلى أبعد من ذلك

في تطبيقات الويب ، غالبًا ما تكون هناك حاجة لبدء مهمة من العرض. في حالة التصاق عالية ، يمكن الجمع بين العرض والمهمة:

 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) 

بالمناسبة ، لتجنب تصادم الأسماء ، تسمى الفئة المتداخلة MetaTask ، وليس Meta ، كما هو الحال في django.

استنتاج

هذه الوظيفة متوقعة في Celery 4.5 . ومع ذلك ، قمت أيضًا بإعداد حزمة تسمح لك بتجربة أداة تزيين المهام اليوم. فكرة الحزمة هي أنه عندما تقوم بترقية Celery إلى الإصدار 4.5 ، يمكنك إزالة الاستيراد دون تغيير سطر واحد من التعليمات البرمجية.