Lifespan signals (low-level)

Tip

Developers are advised to prefer asynchronous context managers. Using signals directly is fully supported, and until v0.2.0 this was the only way to use this plugin — the only downside is that it is a more low-level way to use it.

Before you start

This low-level approach uses ASGI lifespan signals directly. The disadvantage is that the developer has to figure out where to store the global shared state. Prefer the more modern approach via context manager — the global state will be managed by the ASGI server.

Example use case for lifespan signals

Consider a Django application named library. This application retrieves book information from an external API using HTTPX. To ensure efficient utilization of network resources (e.g., connection pooling, see: https://www.python-httpx.org/advanced/#why-use-a-client), we intend to use the HTTPX client.

Typehinting

Firstly, we're going to define a Protocol for typehinting.

This Protocol (HTTPXAppConfig) provides an explicit interface for a request that contains a reference to an instance of httpx.AsyncClient in its state dictionary.

from typing import Protocol

import httpx


class HTTPXAppConfig(Protocol):
    httpx_client: httpx.AsyncClient

Signal receivers

Next, we create receivers for lifespan signals. A startup receiver that creates an instance of the HTTPX client, and a shutdown receiver that closes the client.

import httpx

from .types import HTTPXAppConfig


class ASGILifespanSignalHandler:
    app_config: HTTPXAppConfig

    def __init__(self, app_config: HTTPXAppConfig):
        self.app_config = app_config

    async def startup(self, **_):
        self.app_config.httpx_client = httpx.AsyncClient()

    async def shutdown(self, **_):
        await self.app_config.httpx_client.aclose()

Connect the receivers to the lifespan signals:

import httpx
from django.apps import AppConfig

from django_asgi_lifespan.signals import asgi_shutdown, asgi_startup

from .handlers import ASGILifespanSignalHandler


class ExampleAppConfig(AppConfig):
    httpx_client: httpx.AsyncClient

    def ready(self):
        handler = ASGILifespanSignalHandler(app_config=self)

        asgi_startup.connect(handler.startup, weak=False)
        asgi_shutdown.connect(handler.shutdown, weak=False)

Access shared state in view

Now we can access the HTTPX client instance in our view:

from typing import cast

from django.apps import apps
from django.http import HttpResponse

from .types import HTTPXAppConfig


async def my_library_view(*_) -> HttpResponse:
    library_app = cast(HTTPXAppConfig, apps.get_app_config("library"))
    httpx_client = library_app.httpx_client
    external_api_response = await httpx_client.get("https://www.example.com/")

    return HttpResponse(
        f"{external_api_response.text[:42]}",  # pyright: ignore
        content_type="text/plain",
    )