On the Kraken Customer side of the business, we mostly use Django in our applications. But what happens when you want to build an async application with Django? Read on to find out.

Why async?

In most of our applications, we typically use task queues for background tasks (historically we’ve used celery and RabbitMQ though we’re pivoting to Amazon SQS for some services). While this means some work is completed asynchronously, this does not make the application itself asynchronous and non-blocking.

My team is working on applications for internal tooling for Kraken staff where we’re leveraging LLMs and classic machine learning techniques. Our projects include a Slack chatbot for searching documentation and generating release notes for applications.

LLMs do not return results instantaneously. When using LLMs, you will also often achieve better results when creating dedicated independent agents for specific tasks, e.g. an agent which classifies a block of text into distinct categories, another which summarises the text etc. This is a perfect use case for async programming. Because each agent is doing its own work and isn’t dependent on the outcome of others, we can make them do the work alongside each other, instead of one at a time.

The async stack

Python handles async natively very well, providing the asyncio package to add additional async functionality. For our LLM orchestration, we use pydantic-ai which is also async native, making it very easy to integrate it into async applications.

A typical workflow when interacting with LLMs could be as follows

async def respond_to_user(user_question: str, slack_client: SlackClient) -> str:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(slack_client.update_status("Finding a response..."))
        tg.create_task(domain.store_user_question(user_question))
        response_task = tg.create_task(
            agent.run(prompt=user_question, model="gpt-5")
        )
    return response_task.result()

Here, we do the following asynchronously:

• Update the user to tell them we’re generating a response
• Storing the user question in the database
• Calling an agent to actually generate a response

These are all independent - the outcome of each does not affect the others. Therefore, they can all be independent tasks. The tasks are executed concurrently, making the process more efficient. We exit the context manager when all tasks have completed and move on, in this case returning the response of the agent.

Enter Django

Our first LLM-heavy internal tool was built using SQLAlchemy as it supports async very well. We also used Slack and APIs as our sole interfaces. However, for any project where you want to build a webpage, it is always worth considering Django, as it gives you a lot of the scaffolding of web development for free. So we thought we’d give it a go.

How well does Django do with async? While we’re at the start of our project and there may be things we haven’t discovered yet, overall my verdict would be “okay”.

💡 We’re working with Django v5.2 in this project. When trying to use async with Django, a newer Django version is recommended as async support has significantly improved in recent versions.

Don’t forget the a !

Interacting with the ORM in an async context is pretty straightforward. Unfortunately, compared to e.g. SQLAlchemy, you do have to use slightly different syntax, though it’s easy to remember. Instead of

def update_book_title(book: Book, title: str) -> None:
    book.title = title
    book.save()

you do

async def update_book_title(book: Book, title: str) -> None:
    book.title = title
    await book.asave()

Similarly, there are async equivalents like .aget(), .afirst() and so on.

One thing to remember is that you need to ensure you always prefetch the data linked via foreign keys or many-to-many fields if you’re planning on using it. Not only will Django not let you access any related fields (accessing them would be a synchronous operation and you’re in an async environment), you would also risk working with inconsistent state. Below is a simple example:

# data/models.py

class Book(django_models.Model):
    title = django_models.CharField(max_length=255)
    authors = django_models.ManyToManyField("Author")


# domain/queries.py

@dataclasses.dataclass
class Author:
    name: str
    
@dataclasses.dataclass
class Book:
    title: str
    authors: list[Author]
    
async def get_book(book_id: int) -> Book:
    book_record = await models.Book.objects.prefetch_related("authors").aget(id=book_id)
    # We're accessing book_record.authors here, so we need to have prefetched them in the ORM query.
    authors = [
        Author(name=author_record.name)
        for author_record in book_record.authors.all()
    ]
    return Book(
        title=book_record.title,
        authors=authors,
    )

Atomic transactions

Transaction management unfortunately doesn’t work particularly well in an async context, but you can work around it. Django provides useful decorators via asgiref (which comes with Django by default) to help you achieve atomicity. However, the relevant code will always be executed synchronously. Have a look at the example below:

# domain/operations.py

async def update_book_title(book: Book, title: str) -> None:
    book.title = title
    await book.asave()


async def update_reader_name(reader: Reader, name: str) -> None:
    reader.name = name
    await reader.asave()


# application/reading_list.py

# You will need these to convert between async and sync.
from asgiref.sync import async_to_sync, sync_to_async
# ...other imports...

async def add_to_reading_list(..) -> None:
    await some_async_operation()
    await some_other_async_operation()
    await _update_book_and_reader(
        book=book,
        reader=reader,
        book_title=book_title,
        reader_name=reader_name
    )


@sync_to_async(thread_sensitive=True)
@transaction.atomic
def _update_book_and_reader(
        book: Book,
        reader: Reader,
        book_title: str,
        reader_name: str,
) -> None:
    async_to_sync(operations.update_book_title)(book, book_title)
    async_to_sync(operations.update_reader_name)(reader, reader_name)

What’s going on here?

The function that we want to execute within an atomic transaction, as indicated by the @transaction.atomic decorator, has to be synchronous, as atomic transactions are not supported asynchronously. Therefore, _update_book_and_reader is a synchronous function (def not async def).

However, we’re calling it from an async function (add_to_reading_list) so it needs to look like an async function, meaning it needs to be awaitable (we’re calling it like await _update_book_and_reader(..)). This is what the @sync_to_async decorator achieves. add_to_reading_list can now successfully call _update_book_and_reader . thread_sensitive=True is the default for sync_to_async but is worth calling out as it’s recommended by Django to be used when doing ORM interactions; it ensures all synchronous processes run in the same thread.

Since the atomic transaction only supports synchronous operations, it means the functions we’re calling inside the transaction need to be synchronous. But our operations are async functions! Fortunately, this isn’t a problem. The async_to_sync decorator takes care of that one in turn.

As you can see, while atomic transactions aren’t natively supported in async, you can get around it. If you want to make sure some of what you’re doing is still executed async, this just means you need to group your database updates at the end so that you can do that block (and only that) synchronously.

Management commands

Django management commands are synchronous and therefore can only call sync functions. This isn’t a problem though, as you can just wrap a call to an async function in async_to_sync(...)(...) . Anything that is async within that function will still be executed in an async way.

Async functions, async tests

We still need to test our code, so let’s explore what adjustments we made to our testing setup to accommodate async functions.

Pytest

You can write your tests async for async functions. We use pytest which has an add-on module for async behaviour, pytest-async , which we can use in addition to pytest-django. If you’re using a pyproject.toml file, you can add the following to default all tests to be async:

[tool.pytest.ini_options]
asyncio_mode = "auto"

Otherwise, the same can be achieved by using a decorator.

Tests then look the same, just with the required async and await:

async def test_book_title_is_updated():
    book = models.Book.create(title="Boring title")

    await operations.update_book_title(book=book, title="Unleash the Kraken")

    assert book.title == "Unleash the Kraken"

However, there is a gotcha when using Django. If you want to ensure that each individual test starts with a fresh database state (to e.g. avoid flakey tests through test pollution), you need to ensure that transaction=True when using pytest.mark.django_db . We do this in a conftest.py file like so

def pytest_collection_modifyitems(items):
    for item in items:
        item.add_marker(pytest.mark.django_db(transaction=True))

This is another side effect of Django not supporting transactions in async mode; we cannot roll back the transaction at the end of the test and instead flush the database.

Factory boy

We also like using FactoryBoy in our testing set up. This also needs some adjustment in async which can be a bit fiddly. The road we went down for both our Django project and our SQLAlchemy project is to write new Async factory boy base classes which override certain methods as required to make them awaitable. We do this for both the DjangoModelFactory and the SubFactory . Here is a simple example of the DjangoModelFactory (comments omitted to keep this short):

class AsyncDjangoModelFactory(factory.django.DjangoModelFactory, Generic[_ModelType]):

    @classmethod
    def _create(
        cls,
        model_class: type[django_models.Model],
        *args: Any,
        **kwargs: Any,
    ) -> django_models.Model:
        try:
            asyncio.get_running_loop()
        except RuntimeError:
            pass
        else:
            msg = (
                f"Cannot use {cls.__name__}() in async context. "
                f"Use 'await {cls.__name__}.acreate()' instead."
            )
            raise RuntimeError(msg)

        return super()._create(model_class, *args, **kwargs)

    @classmethod
    async def acreate(cls, **kwargs: Any) -> _ModelType:
        for field_name, declaration in cls._meta.declarations.items():
            if isinstance(declaration, AsyncSubFactory) and field_name not in kwargs:
                subfactory = declaration.get_factory()
                related_instance = await subfactory.acreate()
                kwargs[field_name] = related_instance

        instance = await sync.sync_to_async(cls.build)(**kwargs)

        await instance.asave()
        return instance

A side effect is that you cannot call await factories.SomeFactory() in an async test, but always need to go via .acreate().

From our experience, as you try to have more complex test setup you discover more methods on the inbuilt base classes that you need to override but that’s relatively easy once you have a good starting point. And sometimes it’s not worth the effort, e.g. trying to replace the post_generation decorator with something async compatible.

Now, how do we get this live?

The main thing to know is that you probably want to use ASGI rather than WSGI if you want your application to be fully async. This involves a few different changes compared to e.g. using uwsgi:

• Running the application: you can use a tool like uvicorn to start it up
• Depending on your setup you may need to add additional middleware for intercepting application logs
• Specify ASGI_APPLICATION in the Django settings file

Here’s a short example for the definition of the ASGI application:

# my_project/web/asgi.py
import os

from django.core.asgi import get_asgi_application


os.environ.setdefault("DJANGO_SETTINGS_MODULE", "my_project.settings")

application = get_asgi_application()


# my_project/settings.py
ASGI_APPLICATION = "my_project.web.asgi.application"

There’s a small gotcha around persistent database connections when using async with Django. Django typically keeps database connections open between requests. However, in an async context you don’t want the same connection to be used across different async tasks. Therefore, this should explicitly be disabled in the database settings:

try:
    DATABASES = {"default": env.dj_db_url("DATABASE_URL")}
    DATABASES["default"]["CONN_MAX_AGE"] = 0
except environs.ValidationError:
    # The passed value of DATABASE_URL is invalid.
    raise
except environs.EnvError:
    # There is no DATABASE_URL env var.
    DATABASES = {}

What’s next?

We’ve not really got to a stage yet where we’re dealing with forms and views, so I’m sure there will be more learning there. For now, we have a running application which is executing cronjobs successfully and updating the database!

As you can see from the above, it is possible to work with Django async, as long as you know what the limitations are.

Check out the Django docs on async development here.

Would you like to help us build a better, greener and fairer future?

We're hiring smart people who are passionate about our mission.

Posted by Frederike Jaeger Staff Software Engineer on Jan 12, 2026