Migrating from Django to FastAPI has been quite a trip.

Note: This article applies only to Django 5.1. Things might change in later versions.

The why

We have an old application using Django REST Framework and gunicorn. These both provide unique challenges we had to accept for a long time, but have grown such distaste for that I started to form a plan for a gradual migration out of Django and into more modern technology.

Before we begin, let's go over our requirements:

1. All existing code must continue to work as-is. Users of the API should not be able to observe the change.
2. New code should be possible to write using FastAPI without worrying about existing code.
3. We should be able to use the Django ORM from FastAPI code.

Understanding ASGI/WSGI

Here it is important to be aware of WSGI. It is a Python standard for how a generic web server (HTTP or otherwise) can communicate with the actual application that produces responses. It is quite widely supported in the Python land and allows you to mix-and-match almost any server implementation with almost any backend library. It is a synchronous protocol, meaning the server calls the application's entry function for each incoming request and waits for it to produce a response.

This would of course be terrible if a request would block all following requests and they would have to wait for the first one to finish. Gunicorn and practically all servers support running multiple workers (processes) and multiple threads in each process.

Here's the important detail: Django assumes every request comes in a separate thread. This assumption is built deep into Django and has to be understood as we move forward.

There's also a standard designed to supersede WSGI, called ASGI. It is largely compatible with the WSGI specification, but every request is handled asynchronously inside a single thread (but again optionally across multiple workers). Modern libraries like FastAPI are built to take full advantage of ASGI.

Enter a2wsgi

So, we need a way to run Django ORM queries outside the usual Django runtime. Luckily, the documentation puts us on the right path directly. We just need to initialize Django's configuration.

# asgi.py
import os
import django

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "settings")
django.setup(set_prefix=False)

After running django.setup, we can import and use our models as normal. So, let's add in our application.

# asgi.py
from django.core.handlers.wsgi import WSGIHandler
from fastapi import FastAPI
from a2wsgi import WSGIMiddleware

# Our FastAPI routes, defined later
from .routes import api_router

app = FastAPI(
    title="My App",
)

# Include all api endpoints
app.include_router(api_router)
app.mount("/", WSGIMiddleware(WSGIHandler(), workers=10))

The WSGIMiddleware provided by a2wsgi is an extremely convenient and well implemented wrapper that can take a full WSGI application and make it look like an ASGI application. It has a threadpool that runs each incoming request. Since it goes through the whole Django stack, we can assume all middleware, connection management, etc work as usual.

We have to be mindful that any middleware we add to our FastAPI application will also apply to the mounted Django application. This is not always what you want, as seen later.

This application can then be executed simply like so:

uvicorn asgi:app --host 0.0.0.0 --port 8000

def get_wsgi_application():
    django.setup(set_prefix=False)
    return WSGIHandler()

Adding FastAPI routes

Let's start with the code.

# routes.py
from asgiref.sync import SyncToAsync, ThreadSensitiveContext, sync_to_async
from fastapi import APIRouter, Depends
import django.db

@sync_to_async
def _close():
    django.db.connection.close()

async def _django_request_wrapper():
    # This seems to be the minimum viable solution for safely executing Django
    # ORM code outside the usual Django views, given an async context.
    # This is not entirely ideal, as we end up opening a new thread for each
    # request that uses the database.

    async with ThreadSensitiveContext() as ctx:
        try:
            yield
        finally:
            # Close the database connection if it was used.
            # This avoids starting the worker unnecessarily on requests that
            # don't touch the DB.
            executor = SyncToAsync.context_to_thread_executor.get(ctx, None)
            if executor:
                await _close()


api_router = APIRouter(dependencies=[Depends(_django_request_wrapper)])

class DummyOut(BaseModel):
    id: str

@router.get("/dummy")
async def get_dummy() -> list[Dummy]:
    result = []
    async for dummy in Dummy.objects.all():
        result.append(DummyOut(id=dummy.id))
    return result

The mystical django_request_wrapper function is all we need to make Django queries work properly in async code. It must not be run for the usual Django endpoints, so while it behaves like a middleware here, we use it as a dependency in our base router.

If we look at how Django actually implements all of its async functions on database models, we see a pattern like this:

async def afirst(self):
    return await sync_to_async(self.first)()

Every async function just delegates the call to asgiref.sync_to_async. It runs the given function in a single-worker thread pool. This pool is defined by the current ThreadSensitiveContext or if one isn't used, a single global worker is used. This means that if we forget to set up the context, only a single database query can happen at a time across all incoming FastAPI requests.

Setting up the context is not necessary for non-async endpoints, but there's also no cost to doing it. The thread is only started when the wrapper returned by sync_to_async is first called.

Note however that if you do use non-async endpoints, there is no great way to close the database connections they reserve, so you must call connection.close() manually in each.

Database pooling

Note how in the above solution we run a thread for each incoming request that needs to use the database. Each of these threads gets its own database connection by default, which can easily blow up the database's limits. For example, postgres defaults to a maximum of 100 open connections. The simplest way to deal with this is to just enable database connection pooling.

DATABASES = {
    "default": {
        "ENGINE": "django.db.backends.postgresql",
        "NAME": config[Env.DB_NAME],
        "USER": config[Env.DB_USER],
        "PASSWORD": config[Env.DB_PASSWORD],
        "HOST": config[Env.DB_HOST],
        "PORT": config[Env.DB_PORT],
        "OPTIONS": {
            "pool": {
                "min_size": 4,
                "max_size": 20,
                "max_lifetime": 60 * 60, # 1 hour
                "max_idle": 10 * 60 # close idle connections after 10 minutes
            },
        },
    }
}

That should be all.

Oh, also, if you detect terrible performance in local environment using the above setup, it's a psycopg bug. If there is zero network delay, psycopg hangs up in a multithreaded setup.