Reference

The following is a reference of modules and their public functions and classes.

Please note that this documentation is not versioned. The reference for the last successful build of the main branch is displayed.

asgi

Public function for ASGI Application with custom ASGI handler.

get_asgi_application()

The public interface to Django's custom ASGI support that supports the lifespan protocol.

This is modified copy of

https://github.com/django/django/blob/main/django/core/asgi.py

:return: An ASGI 3 callable.

Source code in django_asgi_lifespan/asgi.py

def get_asgi_application() -> LifespanASGIHandler:
    """
    The public interface to Django's custom ASGI support
    that supports the lifespan protocol.

    This is modified copy of:
        <https://github.com/django/django/blob/main/django/core/asgi.py>

    :return: An ASGI 3 callable.
    """
    django.setup(set_prefix=False)
    return LifespanASGIHandler()

dispatcher

errors

events

send_lifespan_signal_collecting_contexts(signal, scope) async

Dispatches the given signal. Returns a list of async context managers.

Source code in django_asgi_lifespan/events.py

async def send_lifespan_signal_collecting_contexts(
    signal: CompatAsyncSignal, scope: LifespanScope
) -> List[Callable[[], AbstractAsyncContextManager[State]]]:
    """
    Dispatches the given signal. Returns a list of async context managers.
    """
    logger.debug("Dispatching signal: %s", signal)

    if "state" not in scope:
        logger.warning("Missing state in scope. Cannot dispatch signal.")
        raise MissingScopeStateError("Missing state in scope. Cannot dispatch signal.")

    logger.debug(
        "Awaiting signal `%s` using compat `compat_asend_async_only` method.", signal
    )

    # List of tuple pairs [(receiver, response), ...].
    receiver_responses: List[
        Tuple[Any, Callable[[], AbstractAsyncContextManager[State]]]
    ] = await signal.compat_asend_async_only(
        LifespanSender, scope=scope, state=scope["state"]
    )
    context_managers = [context_manager for _, context_manager in receiver_responses]

    return context_managers

send_lifespan_signal_compat(*, signal, scope) async

Dispatches the given signal.

Source code in django_asgi_lifespan/events.py

async def send_lifespan_signal_compat(*, signal: Signal, scope: LifespanScope) -> None:
    """
    Dispatches the given signal.
    """
    logger.debug("Dispatching signal: %s", signal)

    if callable(getattr(signal, "asend", None)):
        logger.debug("Awaiting signal using native `asend` method: %s", signal)
        await signal.asend(sender=LifespanSender, scope=scope)
    else:
        logger.debug("Sending signal using synchronous `send` method: %s", signal)
        responses = signal.send(sender=LifespanSender, scope=scope)

        # Synchronous send method returns coroutine objects, that need to be awaited
        for _, response in responses:
            if not response:
                continue

            if inspect.isawaitable(response):
                await response
            else:
                response()

    logger.debug("Signal: %s dispatched", signal)

handler

Subclass of Django ASGIHandler with ASGI Lifespan support.

LifespanASGIHandler

Bases: ASGIHandler

A subclass of ASGIHandler that supports the ASGI Lifespan protocol.

Source code in django_asgi_lifespan/handler.py

class LifespanASGIHandler(ASGIHandler):
    """A subclass of ASGIHandler that supports the ASGI Lifespan protocol."""

    _lifespan_event_dispatcher: LifespanEventDispatcher

    def __init__(self):
        super().__init__()
        self._lifespan_event_dispatcher = LifespanEventDispatcher()

    async def __call__(
        self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
    ) -> None:
        """
        If scope type is lifespan, handle lifespan request.
        Otherwise, delegate to the superclass call method.

        The base Django `ASGIHandler` can only handle http scope.
        """

        if scope["type"] == "lifespan":
            await self._handle_lifespan(scope, receive, send)
        else:
            await super().__call__(scope, receive, send)

    async def _handle_lifespan(
        self,
        scope: LifespanScope,
        receive: ASGIReceiveCallable,
        send: ASGISendCallable,
    ) -> None:
        """Process lifespan request events."""

        while True:
            message: ASGIReceiveEvent = await receive()

            match message["type"]:
                case "lifespan.startup":
                    try:
                        await self._lifespan_event_dispatcher.startup(scope)
                    except Exception as exc:
                        await send(
                            LifespanStartupFailedEvent(
                                type="lifespan.startup.failed", message=str(exc)
                            )
                        )
                        raise
                    else:
                        await send(
                            LifespanStartupCompleteEvent(
                                type="lifespan.startup.complete"
                            )
                        )

                case "lifespan.shutdown":
                    try:
                        await self._lifespan_event_dispatcher.shutdown(scope)
                    except Exception as exc:
                        await send(
                            LifespanShutdownFailedEvent(
                                type="lifespan.shutdown.failed", message=str(exc)
                            )
                        )
                        raise
                    else:
                        await send(
                            LifespanShutdownCompleteEvent(
                                type="lifespan.shutdown.complete"
                            )
                        )
                        # The return statement is important here to break the while loop
                        # and prevent the function from processing any further messages
                        # after the shutdown event.
                        # Ref.:
                        # https://asgi.readthedocs.io/en/latest/specs/lifespan.html
                        return

                case _:
                    raise ValueError(
                        f"Unknown lifespan message type: {message['type']}"
                    )

__call__(scope, receive, send) async

If scope type is lifespan, handle lifespan request. Otherwise, delegate to the superclass call method.

The base Django ASGIHandler can only handle http scope.

Source code in django_asgi_lifespan/handler.py

async def __call__(
    self, scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable
) -> None:
    """
    If scope type is lifespan, handle lifespan request.
    Otherwise, delegate to the superclass call method.

    The base Django `ASGIHandler` can only handle http scope.
    """

    if scope["type"] == "lifespan":
        await self._handle_lifespan(scope, receive, send)
    else:
        await super().__call__(scope, receive, send)

middleware

register

register_lifespan_manager(*, context_manager)

Registers a context manager for lifecycle events

Source code in django_asgi_lifespan/register.py

def register_lifespan_manager(
    *, context_manager: Callable[[], AbstractAsyncContextManager[State]]
) -> None:
    """
    Registers a context manager for lifecycle events
    """
    wrapper = LifespanContextManagerSignalWrapper(context_manager)
    # weak=False is important here, otherwise the receiver will be garbage collected
    asgi_lifespan.connect(wrapper.receiver, sender=None, weak=False, dispatch_uid=None)

signals

ASGI lifespan Django signals for server startup and shutdown events.

types