With Django 6.0 a new tasks framework was added. Instead of being a full framework like celery , it provides the core interface for other tools to build off of. I created django-inspect-tasks to help me test.

Running Perodic Tasks

By default, the tasks framework does not have anything similar to celery beat to run perodic tasks. This can be provided by the django-crontask package. Crontask needs to find all the scheduled tasks, which it does using a similar snippet.

import importlib
from django.apps import apps

for app in apps.get_app_configs():
    try:
        importlib.import_module(f"{app.name}.tasks")
        print(f"Loaded tasks from {app.name}.")
    except ImportError:
        pass

We typically assume that {app.name}.tasks will be the module that our tasks live in. We can loop through all of our Django AppConfig and try to import the module with import_module. Because crontask and the tasks framework use decorators like @task and @cron, just importing the module is enough to initalize things.

Finding Task objects

In our case, we want the actual Task objects, so we need to do some additional work. We want to look at each object in the module and find any Task objects so we modify our code a bit.

import importlib
from django.apps import apps
from django.tasks import Task

tasks = {}
for app in apps.get_app_configs():
    try:
        module = importlib.import_module(f"{app.name}.tasks")
    except ImportError:
        pass
    else:
        for key in dir(module):
            obj = getattr(module, key)
            if isinstance(obj, Task):
                tasks[obj.module_path] = obj

Supporting other backends

Other task backends (such as the database backend from django-tasks) may provide their own Task object, so we need to check multiple types. isinstance allows us to provide a tuple of classes to check against which we can get from looping through all task backends.

from django.tasks import task_backends
def task_class(self, backend, **options) -> tuple[Task]:
    return tuple({backend.task_class for backend in task_backends.all()})

...

for key in dir(module):
    obj = getattr(module, key)
    if isinstance(obj, TASK_CLASS):
        tasks[obj.module_path] = obj

Putting things together

django-inspect-tasks uses these methods to find all the tasks, so that we can now see all the tasks in our project, and optionally queue them manually.

uv run manage.py tasks
crontask.tasks.heartbeat                    cron[month='*', day='*', day_of_week='*', hour='*', minute='*']
myapp.tasks.one
myapp.tasks.two
otherapp.tasks.another_task

and we can optionally call a specific task

uv run manage.py tasks myapp.tasks.one

While there is more I want to add to my tasks sub command, it was interesting digging into django-crontask and the tasks Framework to see how it works. This should help me in a future migration from celery to the tasks framework. Perhaps in the future, some kind of task inspect commands for list and execute will be added upstream.