I wrote Django commands to make it easier to debug Celery tasks

I wanted to call tasks synchronously (not asynchronously) in a Django + Celery project to see how they work. I thought it would be easier to call it by specifying a task from Django's management command, so I wrote it.

Place the following files under management / commands / of the 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)}")

If you call it like this

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

It behaves the same as calling it like this.

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

I also tried to start the debugger by specifying --pdb.

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

Then, the debugger starts with the breakpoint set in myproject.tasks.example_task. I wrote it for the title, but Celery didn't really matter.

Put it in gist. https://gist.github.com/TakesxiSximada/d986ef7d8fbf5555d8e12586226dc389