illagrenan / django-asgi-lifespan

Django ASGI handler with Lifespan Protocol support.
https://illagrenan.github.io/django-asgi-lifespan/
MIT License
12 stars 2 forks source link
asgi asyncio django python

Django ASGI Handler with Lifespan protocol support

License: MIT pypi pypi Python version Supported Django
Build Status codecov

Main features

async def example_view(request) -> HttpResponse:
    # The client is instanciated just once when the application starts,
    # and closed when the server shuts down
    httpx_client: httpx.AsyncClient = request.state["httpx_client"]

Quickstart

  1. Python ^3.10 || ^3.11 || ^3.12 || ^3.13 and Django ^4.2 || ^5.0.3 || ^5.1 are supported. To install this package, run:

    poetry add django-asgi-lifespan@latest

    or

    pip install --upgrade django-asgi-lifespan
  2. Modify asgi.py to use a ASGI Lifespan compatible handler.

    from django_asgi_lifespan.asgi import get_asgi_application
    
    django_application = get_asgi_application()
    
    async def application(scope, receive, send):
        if scope["type"] in {"http", "lifespan"}:
            await django_application(scope, receive, send)
        else:
            raise NotImplementedError(
                f"Unknown scope type {scope['type']}"
            )
  3. Add state middleware:

    MIDDLEWARE = [
        # ...
        "django_asgi_lifespan.middleware.LifespanStateMiddleware",
        # ...
    ]
  4. Register async context manager:

    from contextlib import asynccontextmanager
    
    import httpx
    
    from django_asgi_lifespan.types import LifespanManager
    
    @asynccontextmanager
    async def httpx_lifespan_manager() -> LifespanManager:
        state = {
            "httpx_client": httpx.AsyncClient()
        }
    
        try:
            yield state
        finally:
            await state["httpx_client"].aclose()
    from django.apps import AppConfig
    
    from django_asgi_lifespan.register import register_lifespan_manager
    from .context import (
        httpx_lifespan_manager,
    )
    
    class ExampleAppConfig(AppConfig):
    
        def ready(self):
            register_lifespan_manager(
                context_manager=httpx_lifespan_manager
            )
  5. Use some resource (in this case the HTTPX client) in views.

    from http import HTTPStatus
    
    import httpx
    from django.http import HttpResponse
    
    async def example_view(request) -> HttpResponse:
        httpx_client: httpx.AsyncClient = request.state["httpx_client"]
    
        await httpx_client.head("https://www.example.com/")
    
        return HttpResponse(
            "OK",
            status=HTTPStatus.OK,
            content_type="text/plain; charset=utf-8",
        )
  6. Run uvicorn:

    uvicorn asgi:application --lifespan=on --port=8080