Lifespan protocol support for ASGI apps and libraries.
Project description
asgi-lifespan
Modular components for adding lifespan protocol support to ASGI apps and libraries.
Contents
Features
- Create a lifespan-capable ASGI app with event handler registration support using
Lifespan
. - Add lifespan support to an ASGI app using
LifespanMiddleware
. - Send lifespan events to an ASGI app (e.g. for testing) using
LifespanManager
. (TODO) - Support for asyncio, trio and curio (provided by anyio).
- Fully type-annotated.
- 100% test coverage.
Installation
pip install asgi-lifespan
Usage
Adding lifespan to ASGI apps
from asgi_lifespan import Lifespan, LifespanMiddleware
# 'Lifespan' is a standalone ASGI app.
# It implements the lifespan protocol,
# and allows registering lifespan event handlers.
lifespan = Lifespan()
@lifespan.on_event("startup")
async def startup():
print("Starting up...")
@lifespan.on_event("shutdown")
async def shutdown():
print("Shutting down...")
# Sync event handlers and an imperative syntax are supported too.
def more_shutdown():
print("Bye!")
lifespan.add_event_handler("shutdown", more_shutdown)
# Example ASGI app. We're using a "Hello, world" application here,
# but any ASGI-compliant callable will do.
async def app(scope, receive, send):
assert scope["type"] == "http"
output = b"Hello, World!"
headers = [
(b"content-type", "text/plain"),
(b"content-length", str(len(output)))
]
await send(
{"type": "http.response.start", "status": 200, "headers": headers}
)
await send({"type": "http.response.body", "body": output})
# 'LifespanMiddleware' returns an ASGI app.
# It forwards lifespan requests to 'lifespan',
# and anything else goes to 'app'.
app = LifespanMiddleware(app, lifespan=lifespan)
Save this script as app.py
. You can serve this application with an ASGI server such as uvicorn:
uvicorn app:app
You should get the following output:
INFO: Started server process [2407]
INFO: Waiting for application startup.
Starting up...
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
Stop the server using Ctrl+C
, and you should get the following output:
INFO: Shutting down
INFO: Waiting for application shutdown.
Shutting down...
Bye!
INFO: Finished server process [2407]
Sending lifespan events
To programmatically send ASGI lifespan events to an ASGI app, use LifespanManager
. This is particularly useful for testing and/or making requests using an ASGI-capable HTTP client such as HTTPX.
from asgi_lifespan import Lifespan, LifespanManager
# Example lifespan-capable ASGI app.
# (Doesn't need to be a `Lifespan` instance.
# Any ASGI app implementing the lifespan protocol will do.)
app = Lifespan()
@app.on_event("startup")
async def startup():
print("Starting up...")
@app.on_event("shutdown")
async def shutdown():
print("Shutting down...")
async def main():
async with LifespanManager(app):
print("We're in!")
# Maybe make some requests to 'app'
# using an ASGI-capable test client here?
Note: if
LifespanManager
detects that the lifespan protocol isn't supported, aLifespanNotSupported
exception is raised. To silence this exception, useLifespanManager(app, ignore_unsupported=True)
.
Save this script as main.py
. You can run it with any of the supported async libraries:
# Add one of these at the bottom of 'main.py'.
import asyncio
asyncio.run(main())
import trio
trio.run(main)
import curio
curio.run(main)
Run $ python main.py
in your terminal, and you should get the following output:
Starting up...
We're in!
Shutting down...
API Reference
Lifespan
def __init__(self, on_startup: Callable = None, on_shutdown: Callable = None)
A standalone ASGI app that implements the lifespan protocol and supports registering event handlers.
Example
lifespan = Lifespan()
Parameters
on_startup
(Callable
): an optional initial startup event handler.on_shutdown
(Callable
): an optional initial shutdown event handler.
add_event_handler
def add_event_handler(self, event_type: str, func: Callable[[], None]) -> None
Register a callback to be called when the application starts up or shuts down.
Imperative version of .on_event()
.
Example
async def on_startup():
...
lifespan.add_event_handler("startup", on_startup)
Parameters
event_type
(str
): one of"startup"
or"shutdown"
.func
(Callable
): a callback. Can be sync or async.
on_event
def on_event(self, event_type: str) -> Callable[[], None]
Register a callback to be called when the application starts up or shuts down.
Decorator version of .add_event_handler()
.
Example
@lifespan.on_event("startup")
async def on_startup():
...
Parameters
event_type
(str
): one of"startup"
or"shutdown"
.
__call__
async def __call__(self, scope: dict, receive: Callable, send: Callable) -> None
ASGI 3 implementation.
LifespanMiddleware
def __init__(self, app: Callable, lifespan: Callable)
An ASGI middleware that forwards "lifespan" requests to lifespan
and anything else to app
.
Example
app = LifespanMiddleware(app, lifespan=lifespan)
This is roughly equivalent to:
default = app
async def app(scope, receive, send):
if scope["type"] == "lifespan":
await lifespan(scope, receive, send)
else:
await default(scope, receive, send)
Parameters
app
(Callable
): an ASGI application to be wrapped by the middleware.lifespan
(Callable
): an ASGI application.- Can be a
Lifespan
instance, but that is not mandatory. - This will only be given
"lifespan"
ASGI scope types, so it is safe (and recommended) to useassert scope["type"] == "lifespan"
in custom implementations.
- Can be a
__call__
async def __call__(self, scope: dict, receive: Callable, send: Callable) -> None
ASGI 3 implementation.
License
MIT
Changelog
All notable changes to this project will be documented in this file.
The format is based on Keep a Changelog.
Unreleased
0.2.0 (September 28, 2019)
Added
- Add
LifespanMiddleware
, an ASGI middleware to add lifespan support to an ASGI app. (Pull #9)
0.1.0 (September 28, 2019)
Added
- Add
Lifespan
, an ASGI app implementing the lifespan protocol with event handler registration support. (Pull #7)
0.0.2 (September 28, 2019)
Fixed
- Installation from PyPI used to fail due to missing
MANIFEST.in
.
0.0.1 (September 28, 2019)
Added
- Empty package.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.