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

Recommended Posts

J'ai écrit des commandes Django pour faciliter le débogage des tâches Celery
Développer la source devicetree pour faciliter la lecture
J'ai essayé de déboguer.
Comment utiliser Decorator dans Django et comment le créer
[Git] J'ai essayé de faciliter la compréhension de l'utilisation de git stash en utilisant un exemple concret.
Vous qui coloriez le journal pour le rendre plus facile à voir
Je veux créer un éditeur de blog avec l'administrateur de django
Une doublure qui formate JSON pour le rendre plus facile à voir
Lorsque j'essaye de créer Apache SSL, cela ne démarre pas.
Je l'ai écrit en langage Go pour comprendre le principe SOLID
Rejoignez CSV normalisé par les pandas Python pour faciliter la vérification
Facilitez le test des programmes qui fonctionnent avec les API avec vcrpy
Je veux déboguer avec Python
Je veux le faire avec Python lambda Django, mais je vais m'arrêter
Je souhaite utiliser Django Debug Toolbar dans les applications Ajax
[Zaif] J'ai essayé de faciliter le commerce de devises virtuelles avec Python