ASGI¶
Given that monkay is used from a bunch of libraries which hook into ASGI lifespans, we have also some ASGI helpers.
CMToASGIMiddleware¶
Transforms a ContextManager (async/sync) to an ASGI-Middleware. This is quite handy if you want to manipulate the ContextVars
and inject e.g. a database but only for a part of the ASGI monolith.
You can also provide an async/sync Callable, which receives the scope and is expected to return a ContextManager (async/sync).
Note
When a ContextManager is also callable, it is still treated as ContextManager (neccessary for compatibility). If you want it to be called,
use lambda scope: callable_cm(scope).
Edgy Example
Here we wrap an app with edgy.
from contextlib import contextmanager
from monkay.asgi import CMToASGIMiddleware
import edgy
def wrap_app(app):
registry = edgy.Registry(...)
instance = edgy.Instance(registry=registry)
@contextmanager
def returns_cm(scope):
with edgy.monkay.with_instance(instance):
yield
app = registry.asgi(CMToASGIMiddleware(app, cm=returns_cm))
instance = edgy.Instance(registry=registry, app=app)
Why is asgi(...) not enough?
We only manipulate the lifespan protocol with asgi, but don't set the instance.
Note
return_cm is a callable which returns a contextmanager. This is a way to generate a contextmanager every request.
This is only one example. We can do much more by injecting ContextManager as middleware or generating them. No need to program boilerplate code anymore for this.
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.