How to share (initialize and close) aiohttp.ClientSession between Django async views to use connection pooling

Django supports async views since version 3.1, so it's great for non-blocking calls to e.g. external HTTP APIs (using, for example, aiohttp).

I often see the following code sample, which I think is conceptually wrong (although it works perfectly fine):

import aiohttp
from django.http import HttpRequest, HttpResponse

async def view_bad_example1(request: HttpRequest):
    async with aiohttp.ClientSession() as session:
        async with session.get("https://example.com/") as example_response:
            response_text = await example_response.text()
            return HttpResponse(response_text[:42], content_type="text/plain")

This code creates a ClientSession for each incoming request, which is inefficient. aiohttp cannot then use e.g. connection pooling.

Don’t create a session per request. Most likely you need a session per application which performs all requests altogether.

Source: https://docs.aiohttp.org/en/stable/client_quickstart.html#make-a-request

The same applies to httpx:

On the other hand, a Client instance uses HTTP connection pooling. This means that when you make several requests to the same host, the Client will reuse the underlying TCP connection, instead of recreating one for every single request.

Source: https://www.python-httpx.org/advanced/#why-use-a-client

Is there any way to globally instantiate aiohttp.ClientSession in Django so that this instance can be shared across multiple requests? Don't forget that ClientSession must be created in a running eventloop (Why is creating a ClientSession outside of an event loop dangerous?), so we can't instantiate it e.g. in Django settings or as a module-level variable.

The closest I got is this code. However, I think this code is ugly and doesn't address e.g. closing the session.

CLIENT_SESSSION = None

async def view_bad_example2(request: HttpRequest):
    global CLIENT_SESSSION

    if not CLIENT_SESSSION:
        CLIENT_SESSSION = aiohttp.ClientSession()

    example_response = await CLIENT_SESSSION.get("https://example.com/")
    response_text = await example_response.text()

    return HttpResponse(response_text[:42], content_type="text/plain")

Basically I'm looking for the equivalent of Events from FastAPI that can be used to create/close some resource in an async context.

By the way here is a performance comparison using k6 between the two views:

view_bad_example1: avg=1.32s min=900.86ms med=1.14s max=2.22s p(90)=2s p(95)=2.1s
view_bad_example2: avg=930.82ms min=528.28ms med=814.31ms max=1.66s p(90)=1.41s p(95)=1.52s

Solution

Django doesn't implement the ASGI Lifespan protocol.
Ref: https://github.com/django/django/pull/13636

Starlette does. FastAPI directly uses Starlette's implementation of event handlers.

Here's how you can achieve that with Django:

Implement the ASGI Lifespan protocol in a subclass of Django's ASGIHandler.

import django
from django.core.asgi import ASGIHandler


class MyASGIHandler(ASGIHandler):
    def __init__(self):
        super().__init__()
        self.on_shutdown = []

    async def __call__(self, scope, receive, send):
        if scope['type'] == 'lifespan':
            while True:
                message = await receive()
                if message['type'] == 'lifespan.startup':
                    # Do some startup here!
                    await send({'type': 'lifespan.startup.complete'})
                elif message['type'] == 'lifespan.shutdown':
                    # Do some shutdown here!
                    await self.shutdown()
                    await send({'type': 'lifespan.shutdown.complete'})
                    return
        await super().__call__(scope, receive, send)

    async def shutdown(self):
        for handler in self.on_shutdown:
            if asyncio.iscoroutinefunction(handler):
                await handler()
            else:
                handler()


def my_get_asgi_application():
    django.setup(set_prefix=False)
    return MyASGIHandler()

Replace the application in asgi.py.

# application = get_asgi_application()
application = my_get_asgi_application()

Implement a helper get_client_session to share the instance:

import asyncio
import aiohttp
from .asgi import application

CLIENT_SESSSION = None

_lock = asyncio.Lock()


async def get_client_session():
    global CLIENT_SESSSION

    async with _lock:
        if not CLIENT_SESSSION:
            CLIENT_SESSSION = aiohttp.ClientSession()
            application.on_shutdown.append(CLIENT_SESSSION.close)

    return CLIENT_SESSSION

Usage:

async def view(request: HttpRequest):
    session = await get_client_session()
    
    example_response = await session.get("https://example.com/")
    response_text = await example_response.text()

    return HttpResponse(response_text[:42], content_type="text/plain")