CeleryのタスクをデバッグしやすいようにDjangoのコマンドを書いた

Django + Celeryのプロジェクトでタスクを同期的(非同期ではなく)に呼び出して動きを確認したかった。
Djangoのmanagementコマンドからタスクを指定して呼び出せると楽かなと思ったので書いた。

以下のファイルをアプリケーションのmanagement/commands/配下に置く。

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

こんな感じで呼び出すと

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

こんな感じで呼び出したのと同じ動きをする。

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

また --pdb を指定するとデバッガが起動するようにもしてみた。

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

とすると、myproject.tasks.example_task内にbreakpointが設定された状態でデバッガが起動する。
標題のために書いたが、Celeryはあまり関係なかったな。

gistに置いておく。
https://gist.github.com/TakesxiSximada/d986ef7d8fbf5555d8e12586226dc389