[PYTHON] Ich habe Django-Befehle geschrieben, um das Debuggen von Sellerie-Aufgaben zu vereinfachen

Ich wollte Aufgaben in einem Django + Sellerie-Projekt synchron (nicht asynchron) aufrufen, um zu sehen, wie sie funktionieren. Ich dachte, es wäre einfacher, eine Aufgabe anzugeben und sie über den Verwaltungsbefehl von Django aufzurufen, also schrieb ich sie.

Platzieren Sie die folgenden Dateien unter Verwaltung / Befehle / der Anwendung.

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)}")

Wenn du es so nennst

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

Es verhält sich genauso, als würde man es so nennen.

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

Ich habe auch versucht, den Debugger durch Angabe von "--pdb" zu starten.

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

Anschließend startet der Debugger mit dem in myproject.tasks.example_task festgelegten Haltepunkt. Ich habe es für den Titel geschrieben, aber Sellerie war nicht wirklich wichtig.

Setzen Sie es in den Kern. https://gist.github.com/TakesxiSximada/d986ef7d8fbf5555d8e12586226dc389