Skip to content

Instantly share code, notes, and snippets.

@bryanhelmig
Last active April 13, 2023 12:40
Show Gist options
  • Save bryanhelmig/6fb091f23c1a4b7462dddce51cfaa1ca to your computer and use it in GitHub Desktop.
Save bryanhelmig/6fb091f23c1a4b7462dddce51cfaa1ca to your computer and use it in GitHub Desktop.
A Django + Starlette or FastAPI connection handling example (ASGI, with WSGIMiddleware).
# 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()
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.
@bryanhelmig
Copy link
Author

bryanhelmig commented Feb 3, 2022

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

@xshapira
Copy link

xshapira commented Jul 8, 2022

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?

@bryanhelmig
Copy link
Author

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

@xshapira
Copy link

xshapira commented Jul 9, 2022

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

👍

@xshapira
Copy link

@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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment