asgi-lifespan2.1.0
Published
Programmatic startup/shutdown of ASGI apps.
pip install asgi-lifespan
Package Downloads
Authors
Project URLs
Requires Python
>=3.7
Dependencies
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
andtrio
. - Fully type-annotated.
- 100% test coverage.
Installation
pip install 'asgi-lifespan==2.*'
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 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!
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
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")
- 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 =======================
Accessing state
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:
...
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 toapp
in the background, then send thelifespan.startup
event and wait for the application to sendlifespan.startup.complete
. - On exit, send the
lifespan.shutdown
event and wait for the application to sendlifespan.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) 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. UseNone
for no timeout.shutdown_timeout
(Optional[float]
, defaults to 5): maximum number of seconds to wait for the application to shutdown. UseNone
for no timeout.
Yields
manager
(LifespanManager
): theLifespanManager
itself. In case you use lifespan state, useasync with LifespanManager(app) as manager: ...
then accessmanager.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 thelifespan.startup
ASGI 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 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.1.0 - 2023-03-28
Added
- Add support for lifespan state. (Pull #59)
2.0.0 - 2022-11-11
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 - 2020-06-08
Fixed
- Update development status to
5 - Production/Stable
. (Pull #32)
1.0.0 - 2020-02-02
Removed
- Drop
Lifespan
andLifespanMiddleware
. 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 - 2019-11-29
Changed
- Move
Lifespan
to thelifespan
module. (Pull #21) - Refactor
LifespanManager
to drop dependency onasynccontextmanager
on 3.6. (Pull #20)
0.5.0 - 2019-11-29
- 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 - 2019-10-06
Fixed
- Ensure
py.typed
is bundled with the package so that type checkers can detect type annotations. (Pull #16)
0.4.1 - 2019-09-29
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 calledreceive()
at least once.
0.4.0 - 2019-09-29
- Enter Alpha development status.
0.3.1 - 2019-09-29
Added
- Add configurable timeouts to
LifespanManager
. (Pull #10)
0.3.0 - 2019-09-29
Added
- Add
LifespanManager
for sending lifespan events into an ASGI app. (Pull #5)
0.2.0 - 2019-09-28
Added
- Add
LifespanMiddleware
, an ASGI middleware to add lifespan support to an ASGI app. (Pull #9)
0.1.0 - 2019-09-28
Added
- Add
Lifespan
, an ASGI app implementing the lifespan protocol with event handler registration support. (Pull #7)
0.0.2 - 2019-09-28
Fixed
- Installation from PyPI used to fail due to missing
MANIFEST.in
.
0.0.1 - 2019-09-28
Added
- Empty package.