Lifespan protocol support for ASGI apps and libraries.
Project description
asgi-lifespan
Modular ASGI components for adding lifespan protocol support to ASGI apps and libraries.
Features
- Create a lifespan-capable ASGI app with event handler registration support using
Lifespan
. (TODO) - Add lifespan support to an ASGI app using
LifespanMiddleware
. (TODO) - 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 support to an ASGI app
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...
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.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.