ASGI

Given that monkay is used from a bunch of libraries which hook into ASGI lifespans, we have also some ASGI helpers.

Lifespan

Wraps an asgi application as AsyncContextManager and run the lifespan protocol. You can optionally provide an timeout parameter. This can be handy for spurious hangups when used for testing. You can also use __aenter__() and __aexit__() manually for e.g. server implementations of lifespan.

Simple cli usage

from monkay.asgi import Lifespan

asgi_app = ...


async def cli_code():
    async with Lifespan(asgi_app) as app:  # noqa: F841
        # do something
        # e.g. app.list_routes()
        ...

Testing

from collections.abc import Awaitable, Callable, MutableMapping
from contextlib import AsyncExitStack
from typing import Any

from monkay.asgi import Lifespan, LifespanHook


async def stub_raise(
    scope: MutableMapping[str, Any],
    receive: Callable[[], Awaitable[MutableMapping[str, Any]]],
    send: Callable[[MutableMapping[str, Any]], Awaitable[None]],
) -> None:
    raise Exception("Should not be reached")


async def setup() -> AsyncExitStack:
    stack = AsyncExitStack()

    # do something
    async def cleanup_async(): ...

    stack.push_async_callback(cleanup_async)

    # do something else
    def cleanup_sync(): ...

    stack.callback(cleanup_sync)

    return stack


async def test_asgi_hook():
    hook_to_test = LifespanHook(LifespanHook(stub_raise, do_forward=False), setup=setup)
    async with Lifespan(hook_to_test, timeout=30):
        pass

ASGI Server

If you want to add asgi lifespan support to an ASGI server you can do as well:

from monkay.asgi import ASGIApp, Lifespan


class Server:
    def __init__(self, app: ASGIApp) -> None:
        self.app = app
        self.lifespan = Lifespan(app)

    async def startup(self) -> None:
        ...
        await self.lifespan.__aenter__()

    async def shutdown(self) -> None:
        ...
        await self.lifespan.__aexit__()

LifespanHook

This is the reverse part to Lifespan.

You have a library with setup/shutdown routines and want to integrate it with lifespan in an ASGI webserver? Or you have django, ... which still doesn't support lifespan?

This middleware is your life-saver.

For hooking simply provide a setup async callable which returns an AsyncExitStack (contextlib) for cleaning up. LifespanHook has an endpoint mode, so that lifespan events are not forwarded. This is required for e.g. django, which still doesn't support lifespans.

Example library integration

from contextlib import AsyncExitStack

from monkay.asgi import ASGIApp, LifespanHook

esmerald_app: ASGIApp = ...  #  type: ignore
django_app: ASGIApp = ...  #  type: ignore


async def setup() -> AsyncExitStack:
    stack = AsyncExitStack()

    # do something
    async def cleanup_async(): ...

    stack.push_async_callback(cleanup_async)

    # do something else
    def cleanup_sync(): ...

    stack.callback(cleanup_sync)

    return stack


# for frameworks supporting lifespan
app = LifespanHook(esmerald_app, setup=setup)
# for django or for testing
# asgi_app = LifespanHook(django_app, setup=setup, do_forward=False)

Example django

Django hasn't lifespan support yet. To use it with lifespan servers (and middleware) we can do something like this:

from monkay.asgi import ASGIApp, LifespanHook

django_app: ASGIApp = ...  # type: ignore

# for django
app = LifespanHook(django_app, do_forward=False)

Example testing

You need a quick endpoint for lifespan? Here it is.

from collections.abc import Awaitable, Callable, MutableMapping
from contextlib import AsyncExitStack
from typing import Any

from monkay.asgi import Lifespan, LifespanHook


async def stub_raise(
    scope: MutableMapping[str, Any],
    receive: Callable[[], Awaitable[MutableMapping[str, Any]]],
    send: Callable[[MutableMapping[str, Any]], Awaitable[None]],
) -> None:
    raise Exception("Should not be reached")


async def setup() -> AsyncExitStack:
    stack = AsyncExitStack()

    # do something
    async def cleanup_async(): ...

    stack.push_async_callback(cleanup_async)

    # do something else
    def cleanup_sync(): ...

    stack.callback(cleanup_sync)

    return stack


async def test_asgi_hook():
    hook_to_test = LifespanHook(LifespanHook(stub_raise, do_forward=False), setup=setup)
    async with Lifespan(hook_to_test, timeout=30):
        pass

Forwarded attributes feature of LifespanHook

Access on attributes which doesn't exist on LifespanHook are forwarded to the wrapped app (callable which can also be something like an Lilya or Esmerald instance). This allows users to access methods on it without unwrapping. Setting and deleting however doesn't work this way. To unwrap to the native instance use the __wrapped__ attribute.

The yielded app of Lifespan is not wrapped and can be natively used.