asgi-lifespan

Programmatically send startup/shutdown lifespan events into ASGI applications. When used in combination with an ASGI-capable HTTP client such as HTTPX, this allows mocking or testing ASGI applications without having to spin up an ASGI server.

Features

• Send lifespan events to an ASGI app using LifespanManager.
• Support for asyncio and trio.
• Fully type-annotated.
• 100% test coverage.

Installation

pip install 'asgi-lifespan==1.*'

Usage

asgi-lifespan provides a LifespanManager to programmatically send ASGI lifespan events into an ASGI app. This can be used to programmatically startup/shutdown an ASGI app without having to spin up an ASGI server.

LifespanManager can run on either asyncio or trio, and will auto-detect the async library in use.

Basic usage

# example.py
from asgi_lifespan import LifespanManager
from starlette.applications import Starlette

# Example lifespan-capable ASGI app. Any ASGI app that supports
# the lifespan protocol will do, e.g. FastAPI, Quart, Responder, ...
app = Starlette(
    on_startup=[lambda: print("Starting up!")],
    on_shutdown=[lambda: print("Shutting down!")],
)

async def main():
    async with LifespanManager(app):
        print("We're in!")

# On asyncio:
import asyncio; asyncio.run(main())

# On trio:
# import trio; trio.run(main)

Output:

$ python example.py
Starting up!
We're in!
Shutting down!

Sending lifespan events for testing

The example below demonstrates how to use asgi-lifespan in conjunction with HTTPX and pytest in order to send test requests into an ASGI app.

• Install dependencies:

pip install asgi-lifespan httpx starlette pytest pytest-asyncio

• Test script:

# test_app.py
import httpx
import pytest
import pytest_asyncio
from asgi_lifespan import LifespanManager
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route


@pytest_asyncio.fixture
async def app():
    async def startup():
        print("Starting up")

    async def shutdown():
        print("Shutting down")

    async def home(request):
        return PlainTextResponse("Hello, world!")

    app = Starlette(
        routes=[Route("/", home)],
        on_startup=[startup],
        on_shutdown=[shutdown]
    )

    async with LifespanManager(app):
        print("We're in!")
        yield app


@pytest_asyncio.fixture
async def client(app):
    async with httpx.AsyncClient(app=app, base_url="http://app.io") as client:
        print("Client is ready")
        yield client


@pytest.mark.asyncio
async def test_home(client):
    print("Testing")
    response = await client.get("/")
    assert response.status_code == 200
    assert response.text == "Hello, world!"
    print("OK")

• Run the test suite:

$ pytest -s test_app.py
======================= test session starts =======================

test_app.py Starting up
We're in!
Client is ready
Testing
OK
.Shutting down

======================= 1 passed in 0.88s =======================

API Reference

LifespanManager

def __init__(
    self,
    app: Callable,
    startup_timeout: Optional[float] = 5,
    shutdown_timeout: Optional[float] = 5,
)

An asynchronous context manager that starts up an ASGI app on enter and shuts it down on exit.

More precisely:

• On enter, start a lifespan request to app in the background, then send the lifespan.startup event and wait for the application to send lifespan.startup.complete.
• On exit, send the lifespan.shutdown event and wait for the application to send lifespan.shutdown.complete.
• If an exception occurs during startup, shutdown, or in the body of the async with block, it bubbles up and no shutdown is performed.

Example

async with LifespanManager(app):
    # 'app' was started up.
    ...

# 'app' was shut down.

Parameters

• app (Callable): an ASGI application.
• startup_timeout (Optional[float], defaults to 5): maximum number of seconds to wait for the application to startup. Use None for no timeout.
• shutdown_timeout (Optional[float], defaults to 5): maximum number of seconds to wait for the application to shutdown. Use None for no timeout.

Raises

• LifespanNotSupported: if the application does not seem to support the lifespan protocol. Based on the rationale that if the app supported the lifespan protocol then it would successfully receive the lifespan.startup ASGI event, unsupported lifespan protocol is detected in two situations:
  • The application called send() before calling receive() for the first time.
  • The application raised an exception during startup before making its first call to receive(). For example, this may be because the application failed on a statement such as assert scope["type"] == "http".
• TimeoutError: if startup or shutdown timed out.
• Exception: any exception raised by the application (during startup, shutdown, or within the async with body) that does not indicate it does not support the lifespan protocol.

License

MIT

Changelog

All notable changes to this project will be documented in this file.

The format is based on Keep a Changelog.

2.0.0 (November 11, 2022)

Removed

• Drop support for Python 3.6. (Pull #55)

Added

• Add official support for Python 3.11. (Pull #55)
• Add official support for Python 3.9 and 3.10. (Pull #46 - Thanks @euri10)

Fixed

• Ensure compatibility with mypy 0.990+, which made no_implicit_optional the default. (Pull #53 - Thanks @AllSeeingEyeTolledEweSew)

1.0.1 (June 8, 2020)

Fixed

• Update development status to 5 - Production/Stable. (Pull #32)

1.0.0 (February 2, 2020)

Removed

• Drop Lifespan and LifespanMiddleware. Please use Starlette's built-in lifespan capabilities instead. (Pull #27)

Fixed

• Use sniffio for auto-detecting the async environment. (Pull #28)
• Enforce 100% test coverage on CI. (Pull #29)

Changed

• Enforce importing from the top-level package by switching to private internal modules. (Pull #26)

0.6.0 (November 29, 2019)

Changed

• Move Lifespan to the lifespan module. (Pull #21)
• Refactor LifespanManager to drop dependency on asynccontextmanager on 3.6. (Pull #20)

0.5.0 (November 29, 2019)

• Enter Beta development status.

Removed

• Remove curio support. (Pull #18)

Added

• Ship binary distributions (wheels) alongside source distributions.

Changed

• Use custom concurrency backends instead of anyio for asyncio and trio support. (Pull #18)

0.4.2 (October 6, 2019)

Fixed

• Ensure py.typed is bundled with the package so that type checkers can detect type annotations. (Pull #16)

0.4.1 (September 29, 2019)

Fixed

• Improve error handling in LifespanManager (Pull #11):
  • Exceptions raised in the context manager body or during shutdown are now properly propagated.
  • Unsupported lifespan is now also detected when the app calls send() before calling having called receive() at least once.

0.4.0 (September 29, 2019)

• Enter Alpha development status.

0.3.1 (September 29, 2019)

Added

• Add configurable timeouts to LifespanManager. (Pull #10)

0.3.0 (September 29, 2019)

Added

• Add LifespanManager for sending lifespan events into an ASGI app. (Pull #5)

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.