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.
LifespanManager.asyncio and trio.pip install 'asgi-lifespan==2.*'
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.
# example.py
from contextlib import asynccontextmanager
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, ...
@asynccontextmanager
async def lifespan(app):
print("Starting up!")
yield
print("Shutting down!")
app = Starlette(lifespan=lifespan)
async def main():
async with LifespanManager(app) as manager:
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!
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.
pip install asgi-lifespan httpx starlette pytest pytest-asyncio
# test_app.py
from contextlib import asynccontextmanager
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():
@asynccontextmanager
async def lifespan(app):
print("Starting up")
yield
print("Shutting down")
async def home(request):
return PlainTextResponse("Hello, world!")
app = Starlette(
routes=[Route("/", home)],
lifespan=lifespan,
)
async with LifespanManager(app) as manager:
print("We're in!")
yield manager.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")
$ 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 =======================
LifespanManager provisions a lifespan state which persists data from the lifespan cycle for use in request/response handling.
For your app to be aware of it, be sure to use manager.app instead of the app itself when inside the context manager.
For example if using HTTPX as an async test client:
async with LifespanManager(app) as manager:
async with httpx.AsyncClient(app=manager.app) as client:
...
LifespanManagerdef __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:
lifespan request to app in the background, then send the lifespan.startup event and wait for the application to send lifespan.startup.complete.lifespan.shutdown event and wait for the application to send lifespan.shutdown.complete.async with block, it bubbles up and no shutdown is performed.Example
async with LifespanManager(app) as manager:
# '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.Yields
manager (LifespanManager): the LifespanManager itself. In case you use lifespan state, use async with LifespanManager(app) as manager: ... then access manager.app to get a reference to the state-aware app.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:
send() before calling receive() for the first time.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.MIT
All notable changes to this project will be documented in this file.
The format is based on Keep a Changelog.
no_implicit_optional the default. (Pull #53 - Thanks @AllSeeingEyeTolledEweSew)5 - Production/Stable. (Pull #32)Lifespan and LifespanMiddleware. Please use Starlette's built-in lifespan capabilities instead. (Pull #27)sniffio for auto-detecting the async environment. (Pull #28)Lifespan to the lifespan module. (Pull #21)LifespanManager to drop dependency on asynccontextmanager on 3.6. (Pull #20)curio support. (Pull #18)anyio for asyncio and trio support. (Pull #18)py.typed is bundled with the package so that type checkers can detect type annotations. (Pull #16)LifespanManager (Pull #11):
send() before calling having called receive() at least once.LifespanManager. (Pull #10)LifespanManager for sending lifespan events into an ASGI app. (Pull #5)LifespanMiddleware, an ASGI middleware to add lifespan support to an ASGI app. (Pull #9)Lifespan, an ASGI app implementing the lifespan protocol with event handler registration support. (Pull #7)MANIFEST.in.FAQs
Programmatic startup/shutdown of ASGI apps.
We found that asgi-lifespan demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.
Security News
The Ultralytics' PyPI Package was compromised four times in one weekend through GitHub Actions cache poisoning and failure to rotate previously compromised API tokens.