كان لدى الكرفس سابقًا علامة --autoreload تمت إزالتها منذ ذلك الحين. ومع ذلك، يحتوي Django على ميزة إعادة التحميل التلقائية المضمنة في أمر manager.py runserver الخاص به. يؤدي غياب إعادة التحميل التلقائي في عمال Celery إلى إنشاء تجربة تطوير مربكة: تحديث كود Python يؤدي إلى إعادة تحميل خادم Django بالرمز الحالي، ولكن أي مهام يطلقها الخادم ستعمل على تشغيل تعليمات برمجية قديمة في عامل Celery.

ستوضح لك هذه المشاركة كيفية إنشاء أمر Manage.py runworker المخصص الذي يعيد تحميل عمال Celery تلقائيًا أثناء التطوير. سيتم تصميم الأمر على غرار خادم التشغيل، وسوف نلقي نظرة على كيفية عمل إعادة التحميل التلقائي لـ Django تحت الغطاء.

قبل أن نبدأ

يفترض هذا المنشور أن لديك تطبيق Django مع تثبيت Celery بالفعل (الدليل). ويفترض أيضًا أن لديك فهمًا للاختلافات بين المشاريع والتطبيقات في جانغو.

ستكون جميع الروابط الخاصة بالكود المصدري والوثائق مخصصة للإصدارات الحالية من Django وCelery في وقت النشر (يوليو 2024). إذا كنت تقرأ هذا في المستقبل البعيد، فربما تغيرت الأمور.

أخيرًا، سيتم تسمية دليل المشروع الرئيسي باسم my_project في أمثلة المنشور.

الحل: أمر مخصص

سنقوم بإنشاء أمر Manage.py مخصص يسمى runworker. نظرًا لأن Django يوفر إعادة تحميل تلقائي عبر أمر runever الخاص به، فسنستخدم كود مصدر runserver كأساس لأمرنا المخصص.

يمكنك إنشاء أمر في Django عن طريق إنشاء دليل إدارة/أوامر/ داخل أي من تطبيقات مشروعك. بمجرد إنشاء الأدلة، يمكنك بعد ذلك وضع ملف Python باسم الأمر الذي ترغب في إنشائه داخل هذا الدليل (docs).

بافتراض أن مشروعك يحتوي على تطبيق يسمى استطلاعات الرأي، فسنقوم بإنشاء ملف على polls/management/commands/runworker.py وإضافة الكود التالي:

# polls/management/commands/runworker.py

import sys
from datetime import datetime

from celery.signals import worker_init

from django.conf import settings
from django.core.management.base import BaseCommand
from django.utils import autoreload

from my_project.celery import app as celery_app


class Command(BaseCommand):
    help = "Starts a Celery worker instance with auto-reloading for development."

    # Validation is called explicitly each time the worker instance is reloaded.
    requires_system_checks = []
    suppressed_base_arguments = {"--verbosity", "--traceback"}

    def add_arguments(self, parser):
        parser.add_argument(
            "--skip-checks",
            action="store_true",
            help="Skip system checks.",
        )
        parser.add_argument(
            "--loglevel",
            choices=("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "FATAL"),
            type=str.upper,  # Transforms user input to uppercase.
            default="INFO",
        )

    def handle(self, *args, **options):
        autoreload.run_with_reloader(self.run_worker, **options)

    def run_worker(self, **options):
        # If an exception was silenced in ManagementUtility.execute in order
        # to be raised in the child process, raise it now.
        autoreload.raise_last_exception()

        if not options["skip_checks"]:
            self.stdout.write("Performing system checks...\n\n")
            self.check(display_num_errors=True)

        # Need to check migrations here, so can't use the
        # requires_migrations_check attribute.
        self.check_migrations()

        # Print Django info to console when the worker initializes.
        worker_init.connect(self.on_worker_init)

        # Start the Celery worker.
        celery_app.worker_main(
            [
                "--app",
                "my_project",
                "--skip-checks",
                "worker",
                "--loglevel",
                options["loglevel"],
            ]
        )

    def on_worker_init(self, sender, **kwargs):
        quit_command = "CTRL-BREAK" if sys.platform == "win32" else "CONTROL-C"

        now = datetime.now().strftime("%B %d, %Y - %X")
        version = self.get_version()
        print(
            f"{now}\n"
            f"Django version {version}, using settings {settings.SETTINGS_MODULE!r}\n"
            f"Quit the worker instance with {quit_command}.",
            file=self.stdout,
        )

هام: تأكد من استبدال جميع مثيلات my_project باسم مشروع Django الخاص بك.

إذا كنت تريد نسخ هذا الرمز ولصقه ومتابعة البرمجة، فيمكنك التوقف بأمان هنا دون قراءة بقية هذا المنشور. هذا حل أنيق سيخدمك جيدًا أثناء تطوير مشروع Django & Celery الخاص بك. ومع ذلك، إذا كنت تريد معرفة المزيد حول كيفية عمله، فاستمر في القراءة.

كيف يعمل (اختياري)

بدلاً من مراجعة هذا الكود سطرًا تلو الآخر، سأناقش الأجزاء الأكثر إثارة للاهتمام حسب الموضوع. إذا لم تكن على دراية بأوامر Django المخصصة، فقد ترغب في مراجعة المستندات قبل المتابعة.

إعادة التحميل التلقائي

هذا الجزء يبدو الأكثر سحراً. داخل نص طريقة Handle () للأمر، يوجد استدعاء لـ autoreload.run_with_reloader () الداخلي لـ Django. يقبل وظيفة رد الاتصال التي سيتم تنفيذها في كل مرة يتم فيها تغيير ملف Python في المشروع. كيف يعمل هذا في الواقع؟

دعونا نلقي نظرة على نسخة مبسطة من الكود المصدري لوظيفة autoreload.run_with_reloader(). تقوم الوظيفة المبسطة بإعادة كتابة التعليمات البرمجية وإدراجها وحذفها لتوفير الوضوح حول عملها.

# NOTE: This has been dramatically pared down for clarity.

def run_with_reloader(callback_func, *args, **kwargs):
    # NOTE: This will evaluate to False the first time it is run.
    is_inside_subprocess = os.getenv("RUN_MAIN") == "true"

    if is_inside_subprocess:
        # The reloader watches for Python file changes.
        reloader = get_reloader()

        django_main_thread = threading.Thread(
            target=callback_func, args=args, kwargs=kwargs
        )
        django_main_thread.daemon = True
        django_main_thread.start()

        # When the code changes, the reloader exits with return code 3.
        reloader.run(django_main_thread)

    else:
        # Returns Python path and the arguments passed to the command.
        # Example output: ['/path/to/python', './manage.py', 'runworker']
        args = get_child_arguments()

        subprocess_env = {**os.environ, "RUN_MAIN": "true"}
        while True:
            # Rerun the manage.py command in a subprocess.
            p = subprocess.run(args, env=subprocess_env, close_fds=False)
            if p.returncode != 3:
                sys.exit(p.returncode)

عند تشغيل Manage.py runworker في سطر الأوامر، فإنه سيستدعي أولاً الأسلوب Handle() الذي سيستدعي run_with_reloader().

داخل run_with_reloader()، سيتم التحقق لمعرفة ما إذا كان متغير البيئة المسمى RUN_MAIN له قيمة "صحيح". عندما يتم استدعاء الدالة لأول مرة، يجب ألا يكون لـ RUN_MAIN أي قيمة.

عندما لا يتم تعيين RUN_MAIN على "صحيح"، سوف يدخل run_with_reloader() في حلقة. داخل الحلقة، ستبدأ عملية فرعية تعيد تشغيل Manage.py [command_name] الذي تم تمريره، ثم تنتظر حتى تخرج هذه العملية الفرعية. إذا خرجت العملية الفرعية برمز الإرجاع 3، فإن التكرار التالي للحلقة سيبدأ عملية فرعية جديدة وينتظر. سيتم تشغيل الحلقة حتى تقوم العملية الفرعية بإرجاع رمز الخروج الذي ليس 3 (أو حتى يخرج المستخدم باستخدام ctrl c). بمجرد حصوله على رمز الإرجاع غير 3، فإنه سيخرج من البرنامج تمامًا.

تقوم العملية الفرعية الناشئة بتشغيل الأمر manager.py مرة أخرى (في حالتنا manager.py runworker)، ومرة ​​أخرى سوف يستدعي الأمر run_with_reloader(). هذه المرة، سيتم تعيين RUN_MAIN على "صحيح" لأن الأمر يعمل في عملية فرعية.

الآن بعد أن عرف run_with_reloader() أنه في عملية فرعية، سيحصل على أداة إعادة تحميل تراقب تغييرات الملف، وتضع وظيفة رد الاتصال المتوفرة في سلسلة رسائل، وتمريرها إلى أداة إعادة التحميل التي تبدأ في مراقبة التغييرات.

عندما تكتشف أداة إعادة التحميل تغييرًا في الملف، فإنها تقوم بتشغيل sys.exit(3). يؤدي هذا إلى الخروج من العملية الفرعية، مما يؤدي إلى التكرار التالي للحلقة من التعليمات البرمجية التي أنتجت العملية الفرعية. وفي المقابل، يتم إطلاق عملية فرعية جديدة تستخدم نسخة محدثة من الكود.

فحوصات النظام والترحيل

افتراضيًا، تقوم أوامر Django بإجراء فحوصات للنظام قبل تشغيل طريقة Handle() الخاصة بها. ومع ذلك، في حالة runserver وأمر runworker المخصص لدينا، سنرغب في تأجيل تشغيلهما حتى نكون داخل رد الاتصال الذي نقدمه إلى run_with_reloader(). في حالتنا، هذه هي طريقة run_worker() الخاصة بنا. يتيح لنا ذلك تشغيل الأمر مع إعادة التحميل التلقائي أثناء إصلاح عمليات فحص النظام المعطلة.

لتأجيل تشغيل فحوصات النظام، يتم تعيين قيمة السمة require_system_checks إلى قائمة فارغة، ويتم إجراء الاختبارات عن طريق استدعاء self.check() في نص run_worker(). مثل runserver، يتحقق أمر runworker المخصص لدينا أيضًا لمعرفة ما إذا تم تشغيل جميع عمليات الترحيل، ويعرض تحذيرًا إذا كانت هناك عمليات ترحيل معلقة.

نظرًا لأننا نقوم بالفعل بإجراء عمليات فحص نظام Django ضمن طريقة run_worker()، فإننا نقوم بتعطيل عمليات فحص النظام في Celery عن طريق تمرير علامة --skip-checks لمنع العمل المكرر.

تم رفع كافة التعليمات البرمجية المتعلقة بعمليات فحص النظام وعمليات الترحيل مباشرةً من التعليمات البرمجية المصدرية لأمر runserver.

celery_app.worker_main()

يطلق تنفيذنا عامل Celery مباشرة من Python باستخدام celery_app.worker_main() بدلاً من إرساله إلى Celery.

on_worker_init()

يتم تنفيذ هذا الكود عند تهيئة العامل، ويعرض التاريخ والوقت وإصدار Django وأمر الإنهاء. تم تصميمه وفقًا للمعلومات التي يتم عرضها عند تشغيل خادم التشغيل.

نموذج آخر لخادم التشغيل

تم أيضًا رفع الأسطر التالية من مصدر خادم التشغيل:

• suppressed_base_arguments = {"--الإسهاب"، "--traceback"}
• autoreload.raise_last_exception()

تسجيل مستوى

يحتوي أمرنا المخصص على مستوى سجل قابل للتكوين في حالة رغبة المطور في ضبط الإعداد من واجهة سطر الأوامر (CLI) دون تعديل الكود.

نذهب أبعد من ذلك

لقد بحثت وحثت على الكود المصدري لـ Django & Celery لبناء هذا التنفيذ، وهناك العديد من الفرص لتوسيعه. يمكنك تكوين الأمر لقبول المزيد من الوسائط العاملة الخاصة بـ Celery. وبدلاً من ذلك، يمكنك إنشاء أمر Manage.py مخصص يعيد تحميل أي أمر Shell تلقائيًا مثلما فعل David Browne في هذا Gist.

إذا وجدت هذا مفيدًا، فلا تتردد في ترك إعجاب أو تعليق. شكرا للقراءة.