[PYTHON] J'ai écrit des commandes Django pour faciliter le débogage des tâches Celery

Je voulais appeler des tâches de manière synchrone (et non asynchrone) dans un projet Django + Celery pour voir comment elles fonctionnent. Je l'ai écrit parce que je pensais qu'il serait facile de spécifier une tâche et de l'appeler à partir de la commande de gestion de Django.

Placez les fichiers suivants sous gestion / commandes / de l'application.

run_task.py

import importlib
import inspect
import json
import pdb

from django.core.management.base import BaseCommand


class Command(BaseCommand):
    help = "Django commands that make debugging celery tasks easier"

    def add_arguments(self, parser):
        parser.add_argument(
            "dottedname", help="The dotted name of the callable object."
        )
        parser.add_argument(
            "--task-args",
            default="[]",
            type=json.loads,
            help="Arguments passed to the task. (default: '[]')",
        )
        parser.add_argument(
            "--task-kwargs",
            default="{}",
            type=json.loads,
            help="Keyword arguments passed to the task. (default: '{}')",
        )
        parser.add_argument(
            "--pdb", action="store_true", help="Stop execution by debugger."
        )
        parser.add_argument(
            "--pdb-offset",
            default=0,
            type=int,
            help="Offset for debugger to create breakpoint. (default: 0)",
        )

    def handle(self, **options):
        dotted_list = options["dottedname"].strip().split(".")
        module_name = ".".join(dotted_list[:-1])
        func_name = dotted_list[-1]

        try:
            module = importlib.import_module(module_name)
        except ModuleNotFoundError:
            self.stderr.write(f"No module: {module_name}")
            return

        try:
            func = getattr(module, func_name)
        except AttributeError:
            self.stderr.write(f"No attribute: {func_name} not in {module_name}")
            return

        if not callable(func):
            self.stderr.write(f"Not function: {module_name}.{func_name}")
            return

        if options["pdb"]:
            lineno = inspect.getsourcelines(func)[1] + options["pdb_offset"]
            debugger = pdb.Pdb()
            debugger.set_break(module.__file__, lineno=lineno, funcname=func_name)
            debugger.set_trace()

        result = func(*options["task_args"], **options["task_kwargs"])
        self.stdout.write(f"Return: {json.dumps(result)}")

Si tu l'appelles comme ça

$ python manage.py run_task myproject.tasks.example_task --task-args '[1, 2, 3]'

Il se comporte de la même manière que l'appeler ainsi.

  >>> from myproject.tasks import example_task
  >>> example_task(1, 2, 3)

J'ai également essayé de démarrer le débogueur en spécifiant --pdb.

$ python manage.py run_task myproject.tasks.example_task --pdb --task-args '[1, 2, 3]'

Ensuite, le débogueur démarre avec le point d'arrêt défini dans myproject.tasks.example_task. Je l'ai écrit pour le titre, mais Celery n'avait pas vraiment d'importance.

Mettez-le dans l'essentiel. https://gist.github.com/TakesxiSximada/d986ef7d8fbf5555d8e12586226dc389