asgi-lifespan

Modular 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.
• 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, a LifespanNotSupported exception is raised. To silence this exception, use LifespanManager(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.

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.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.