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
asyncioandtrio. - 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
from asgi_lifespan import LifespanManager
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
from starlette.routing import Route
@pytest.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.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
lifespanrequest toappin the background, then send thelifespan.startupevent and wait for the application to sendlifespan.startup.complete. - On exit, send the
lifespan.shutdownevent and wait for the application to sendlifespan.shutdown.complete. - If an exception occurs during startup, shutdown, or in the body of the
async withblock, 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. UseNonefor no timeout.shutdown_timeout(Optional[float], defaults to 5): maximum number of seconds to wait for the application to shutdown. UseNonefor 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 thelifespan.startupASGI event, unsupported lifespan protocol is detected in two situations:- The application called
send()before callingreceive()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 asassert scope["type"] == "http".
- The application called
TimeoutError: if startup or shutdown timed out.Exception: any exception raised by the application (during startup, shutdown, or within theasync withbody) that does not indicate it does not support the lifespan protocol.
License
MIT
![[feature request] support ContextVars](https://avatars.githubusercontent.com/u/662249?v=4)
1 Feb 15, 2022
3 Feb 11, 2022
0 Jul 25, 2022
307 Jan 07, 2023
1 Jan 18, 2022
19 Oct 14, 2022
114 Jul 21, 2022
25 Dec 05, 2022
17 Jun 27, 2022
85 Dec 07, 2022
4 Feb 03, 2022
1 Nov 18, 2021
35k Dec 31, 2022
8 Feb 27, 2022
1.3k Jan 06, 2023
84 Dec 31, 2022
261 Jan 03, 2023
3 Jun 23, 2022
5 Oct 01, 2022
504 Jan 04, 2023