A Django + Starlette or FastAPI connection handling example (ASGI, with WSGIMiddleware).

asgi.py

# Looking for a good remote Python job? Check out Zapier at https://zapier.com/jobs/.

import functools
import importlib
import os

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")  # noqa
import anyio
from asgiref.sync import sync_to_async
from django.conf import settings
from django.core.wsgi import get_wsgi_application
from django.db import (
    InterfaceError,
    OperationalError,
    close_old_connections,
    connections,
)
from fastapi import FastAPI
from starlette.applications import Starlette
from starlette.concurrency import run_in_threadpool
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.middleware.wsgi import WSGIMiddleware

FILES_TO_IMPORT = {
    "api.py",
    "views.py",
    "urls.py",
}


def run_once(func):
    def wrapper(*args, **kwargs):
        if not wrapper.has_run:
            wrapper.has_run = True
            return func(*args, **kwargs)

    wrapper.has_run = False
    return wrapper


@run_once
def find_and_load_select_modules():
    """
    Use os.walk and importlib.import_module to find all views.py files
    in settings.BASE_DIR and "import" them such that any @app.get()
    decorators are found and executed properly, before any traffic is
    served.
    """
    for root, dirs, files in os.walk(settings.BASE_DIR):
        files = [f for f in files if not f[0] == "."]
        dirs[:] = [d for d in dirs if not d[0] == "."]
        for file in files:
            if any(file.endswith(file_to_import) for file_to_import in FILES_TO_IMPORT):
                module_path = os.path.join(root, file)
                module_dir = os.path.dirname(module_path)
                module_rel_path = os.path.relpath(module_dir, settings.BASE_DIR)
                module_rel_path = module_rel_path.replace(os.sep, ".")
                module_name = module_rel_path + "." + file.replace(".py", "")
                importlib.import_module(module_name)


def close_connections_for_func(func):
    """
    A decorator that will close all connections after calling the function,
    if it throws an appropriate error.
    """

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except (OperationalError, InterfaceError):
            close_old_connections()
            raise

    return wrapper


async def smart_sync_to_async(func, *args, **kwargs):
    """
    A Starlette/FastAPI compatible replacement for Django's, with slightly
    different __call__ semantics (drop currying pattern for immediate invoke).
    """
    wrapped_func = close_connections_for_func(func)
    is_starlette = anyio.get_current_task().name.startswith("starlette.")
    if is_starlette:
        return await run_in_threadpool(wrapped_func, *args, **kwargs)
    else:
        return await sync_to_async(wrapped_func)(*args, **kwargs)


class AsyncCloseConnectionsMiddleware(BaseHTTPMiddleware):
    """
    Using this middleware to call close_old_connections() twice is a pretty yucky hack,
    as it appears that run_in_threadpool (used by Starlette/FastAPI) and sync_to_async
    (used by Django) have divergent behavior, ultimately acquiring the incorrect thread
    in mixed sync/async which has the effect of duplicating connections.

    We could fix the duplicate connections too if we normalized the thread behavior,
    but at minimum we need to clean up connections in each case to prevent persistent
    "InterfaceError: connection already closed" errors when the database connection is
    reset via a database restart or something -- so here we are!

    If we always use smart_sync_to_async(), this double calling isn't necessary, but
    depending on what levels of abstraction we introduce, we might silently break the
    assumptions. Better to be safe than sorry!
    """

    async def dispatch(self, request, call_next):
        await run_in_threadpool(close_old_connections)
        await sync_to_async(close_old_connections)()
        try:
            response = await call_next(request)
        finally:
            # in tests, use @override_settings(CLOSE_CONNECTIONS_AFTER_REQUEST=True)
            if getattr(settings, 'CLOSE_CONNECTIONS_AFTER_REQUEST', False):
                await run_in_threadpool(connections.close_all)
                await sync_to_async(connections.close_all)()
        return response


@functools.cache
def register_fast_api_application(root_path: str, **kwargs) -> FastAPI:
    """
    In your views.py, etc. files, you can do this:

        app = register_fast_api_application("/api/v1/")

        @app.get("/hello")
        def hello():
            return {"hello": "world"}

    """
    application = get_starlette_application()
    fast_api_application = FastAPI(root_path=root_path, debug=settings.DEBUG, **kwargs)
    application.mount(root_path, fast_api_application)
    return fast_api_application


@functools.cache
def get_starlette_application() -> Starlette:
    return Starlette(debug=settings.DEBUG)


@functools.cache
def get_application() -> Starlette:
    django_application = get_wsgi_application()
    application = get_starlette_application()
    application.add_middleware(AsyncCloseConnectionsMiddleware)
    find_and_load_select_modules()  # this must happen before the django mount
    application.mount("/", WSGIMiddleware(django_application))
    return application


# run me with `uvicorn myproject.asgi:application`
application = get_application()

license.txt

Copyright 2022 Zapier, Inc.

Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

A good test file might look something like this:

from django.test import TestCase, override_settings
from fastapi.testclient import TestClient

from myproject.applications import get_base_application


@override_settings(CLOSE_CONNECTIONS_AFTER_REQUEST=True)
class BaseTestCase(TestCase):
    api_client: TestClient

    @classmethod
    def setUpClass(cls) -> None:
        super().setUpClass()
        cls.api_client = TestClient(get_base_application())


class APITestCase(BaseTestCase):
    def test_retrieving_api_v1_add(self):
        response = self.api_client.get("/api/v1/hello")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json(), {"hello": "world"})

Thanks for the gist, Bryan. But I'm facing two errors:

"get_base_application" is not defined

"check_and_close_old_connections" is not defined

Any idea how to resolve this?

@xshapira those were just copy pasta fails from our internal codebase -- note the update.

@xshapira those were just copy pasta fails from our internal codebase -- note the update.

👍

@bryanhelmig I slightly improved the file by fixing RuntimeError with custom middleware when client drops connection before the request is finished.

https://gist.github.com/xshapira/653b8d7dbd4757e183132c3c82f232b0