florimondmanca / asgi-lifespan Goto Github PK
View Code? Open in Web Editor NEWProgrammatic startup/shutdown of ASGI apps.
Home Page: https://pypi.org/project/asgi-lifespan
License: MIT License
Programmatic startup/shutdown of ASGI apps.
Home Page: https://pypi.org/project/asgi-lifespan
License: MIT License
I'm using a stateful lifetime to share a database connection on my FastAPI app, which works when started with uvicorn
.
Since httpx
's AsyncClient
does not support lifetime management, they recommend using asgi-lifespan
.
Reference: https://www.python-httpx.org/async/#startupshutdown-of-asgi-apps
My lifetime looks like this.
class State(TypedDict):
db: Connection | Engine
@asynccontextmanager
async def lifespan(app: Starlette) -> AsyncGenerator[State, None]:
extra = app.__dict__["extra"]
config = extra["config"]
engine = create_engine(url=config.database_url, echo=config.debug)
connection: Connection | None = None
if config.debug:
db.Base.metadata.create_all(bind=engine)
connection = engine.connect()
yield {"db": connection or engine}
if connection is not None:
connection.close()
return
then my test looks like this
@pytest.mark.anyio
async def test_post_post(app: FastAPI):
print("Start test")
async with LifespanManager(app):
async with AsyncClient(app=app) as client:
response = await client.post("/post", json={"title": "test"})
assert response.status_code == 200
response_json = response.json()
assert response_json["title"] == "test"
# check it's inserted in database
state: State = cast(State, app.state)
with Session(state["db"]) as session:
db_post = (
session.query(db.Post).filter_by(id=response_json["uid"]).first()
)
assert db_post is not None
assert db_post.title == "test"
However I get
.venv\Lib\site-packages\asgi_lifespan\_manager.py:90: in __aenter__
await self.startup()
.venv\Lib\site-packages\asgi_lifespan\_manager.py:36: in startup
raise self._app_exception
.venv\Lib\site-packages\asgi_lifespan\_manager.py:64: in run_app
await self.app(scope, self.receive, self.send)
.venv\Lib\site-packages\fastapi\applications.py:273: in __call__
await super().__call__(scope, receive, send)
.venv\Lib\site-packages\starlette\applications.py:120: in __call__
await self.middleware_stack(scope, receive, send)
.venv\Lib\site-packages\starlette\middleware\errors.py:149: in __call__
await self.app(scope, receive, send)
.venv\Lib\site-packages\starlette\middleware\exceptions.py:55: in __call__
await self.app(scope, receive, send)
.venv\Lib\site-packages\fastapi\middleware\asyncexitstack.py:21: in __call__
raise e
.venv\Lib\site-packages\fastapi\middleware\asyncexitstack.py:18: in __call__
await self.app(scope, receive, send)
.venv\Lib\site-packages\starlette\routing.py:705: in __call__
await self.lifespan(scope, receive, send)
-----------------------------------------
self = <fastapi.routing.APIRouter object at 0x0000026E1B783ED0>
scope = {'app': <fastapi.applications.FastAPI object at 0x0000026E1B783F10>, 'fastapi_astack': <contextlib.AsyncExitStack object at 0x0000026E1CBFD890>, 'router': <fastapi.routing.APIRouter object at 0x0000026E1B783ED0>, 'type': 'lifespan'}
receive = <bound method LifespanManager.receive of <asgi_lifespan._manager.LifespanManager object at 0x0000026E1CBFD410>>
send = <bound method LifespanManager.send of <asgi_lifespan._manager.LifespanManager object at 0x0000026E1CBFD410>>
async def lifespan(self, scope: Scope, receive: Receive, send: Send) -> None:
"""
Handle ASGI lifespan messages, which allows us to manage application
startup and shutdown events.
"""
started = False
app: typing.Any = scope.get("app")
await receive()
try:
async with self.lifespan_context(app) as maybe_state:
if maybe_state is not None:
if "state" not in scope:
> raise RuntimeError(
'The server does not support "state" in the lifespan scope.'
)
E RuntimeError: The server does not support "state" in the lifespan scope.
.venv\Lib\site-packages\starlette\routing.py:678: RuntimeError
I reproduced the app example given in the README.txt and I got the following error:
❯ pytest -s test_app.py
========================================== test session starts ===========================================
platform linux -- Python 3.10.7, pytest-7.1.3, pluggy-1.0.0
rootdir: /home/daniel/coding/python/test-asgi-lifespan
plugins: anyio-3.6.1, asyncio-0.19.0
asyncio: mode=strict
collected 1 item
test_app.py Testing
F
================================================ FAILURES ================================================
_______________________________________________ test_home ________________________________________________
client = <async_generator object client at 0x7f65b037a9c0>
@pytest.mark.asyncio
async def test_home(client):
print("Testing")
> response = await client.get("/")
E AttributeError: 'async_generator' object has no attribute 'get'
test_app.py:43: AttributeError
======================================== short test summary info =========================================
FAILED test_app.py::test_home - AttributeError: 'async_generator' object has no attribute 'get'
=========================================== 1 failed in 0.06s ============================================
I run pytest -s test_app.py
on a new environment with the installed packages explained in the README. Here is my pip list
:
Package Version
-------------- ---------
anyio 3.6.1
asgi-lifespan 1.0.1
attrs 22.1.0
certifi 2022.9.14
h11 0.12.0
httpcore 0.15.0
httpx 0.23.0
idna 3.4
iniconfig 1.1.1
packaging 21.3
pip 22.2.2
pluggy 1.0.0
py 1.11.0
pyparsing 3.0.9
pytest 7.1.3
pytest-asyncio 0.19.0
rfc3986 1.5.0
setuptools 63.2.0
sniffio 1.3.0
starlette 0.20.4
tomli 2.0.1
Sounds like pytest
fixture is returning an async iterable instead the value in the yield
.
Could it be caused by some bug in some version of some package, or is a problem of the asgi-lifespan
? Or I mess up with my setting?
Currently, we allow and encourage:
lifespan = Lifespan()
@lifespan.on_event("startup")
async def startup(): ...
lifespan.add_event_handler("shutdown", lambda: ...)
In line with Starlette's approach in 0.13, we could encourage a declarative style instead:
async def startup(): ...
async def shutdown(): ...
lifespan = Lifespan(on_startup=[startup], on_shutdown=[shutdown])
This would need:
on_startup
and on_shutdown
parameters to Lifespan
to accept sequences of callables, instead of a single callable.@.on_event()
and .add_event_handler()
.Related to agronholm/anyio#74
from asgi_lifespan import LifespanManager, Lifespan
import pytest
@pytest.fixture
async def app():
app = Lifespan()
async with LifespanManager(app):
yield app
@pytest.mark.asyncio
async def test_something(app):
pass
Running pytest tests.py
fails with a KeyError
.
This is an issue with anyio
that is being resolved — tracking here for clarity.
Currently this package provides three components…
Lifespan
: an ASGI app that acts as a container of startup/shutdown event handlers.LifespanMiddleware
: routes lifespan
requests to a Lifespan
app, used to add lifespan support to any ASGI app.LifespanManager
: sends lifespan events into an ASGI app.LifespanManager
is obviously useful, as it's addressing a use case that's not addressed elsewhere — sending lifespan events into an app, with async support. (Starlette does this with its TestClient
, but it's sync only, and HTTPX does not send lifespan events when calling into an ASGI app.)
I'm less confident that Lifespan
and LifespanMiddleware
are worth keeping in here, though. Most ASGI frameworks provide a lifespan implementation anyway…
Eg. Starlette essentially has the exact same code than our own Lifespan
built into its router:
The super-light routing provided by our LifespanMiddleware
is built into the router as well:
So I'm wondering if using these two components on their own own is a use case that people encounter in the wild. I think it's not, and that maybe we should drop them and point people at using Starlette for handling lifespan on the application side.
This package was originally prompted by encode/httpx#350.
The use case was to test a Starlette app by making calls within a lifespan context, e.g.:
# asgi.py
from starlette.applications import Starlette
from starlette.responses import PlainTextResponse
app = Starlette()
@app.route("/")
async def home(request):
return PlainTextResponse("Hello, world!")
# tests.py
import httpx
import pytest
from asgi_lifespan import LifespanManager
from .asgi import app
@pytest.fixture
async def app():
async with LifespanManager(app):
yield
@pytest.fixture
async def client(app):
async with httpx.AsyncClient(app=app) as client:
yield client
@pytest.mark.asyncio
async def test_app(client):
r = await client.get("http://testserver")
assert r.status_code == 200
assert r.text == "Hello, world!"
We should add a section to the documentation on how to use asgi-lifespan
in this use case.
Prompted by https://gitter.im/encode/community?at=5f37fc0eba27767ee5f1d4ed
As per the ASGI spec, when an application shutdown fails, the app should send the lifespan.shutdown.failed
ASGI message.
Currently, LifespanManager
does not listen for this message on shutdown, but rather catches any exception raised during startup.
This works fine for Starlette apps (since Starlette does not obey the spec, and lets any shutdown exception bubble through instead of sending lifespan.shutdown.failed
), but for example I'd expect we have a small interop issue with Quart here (see source).
It's possible we might need to wait for this to be resolved in Starlette before fixing things, but not clear we can't support both use cases at the same time either.
Related issue in Uvicorn: encode/uvicorn#755
Currently how README is encouraging users to import from the top-level package. That’s good, but we could be enforcing it more, eg by switching to private module names. This would reduce the potential API surface, which is almost always a good thing (eg to allow us to more easily move things around if needed). Eg right now users could import our concurrency backends, but we really don’t want to expose those as they’re tightly tied to our specific needs in terms of concurrency.
So, we should switch to...
asgi_lifespan._middleware
asgi_lifespan._lifespan
asgi_lifespan._manager
asgi_lifespan._backends
And update imports __init__.py
.
It seem that setting context variables in on_startup
is not supported.
My naïve understanding is that there needs to be a new (copied) context in which all of on_startup
, asgi request and then on_shutdown
are ran.
Otherwise, the value that's been set is not accessible in the request handlers and/or the context var cannot be reset in the shutdown function using the token that was generated in the startup function.
Hello,
I think this is an awesome project. However, the strong dependency on starlette 0.13 makes my test hang forever (I haven't upgrade my code to 0.13 yet so I guess that's that...). Using starlette 0.12 doesn't cause an issue as far as I can tell with this plugin. So, I wondered about the rationale for such strict dependency?
As it stands, I had to drop the plugin when I wanted to migrate all my tests to it :(
Thanks,
Provides startup/shutdown event handler registration, and an implementation in reaction to lifespan
ASGI messages.
Proposed API:
from asgi_lifespan import Lifespan, LifespanMiddleware
# Standalone lifespan ASGI app.
lifespan = Lifespan()
# Decorator syntax:
@lifespan.on_event("startup")
async def startup():
pass
@lifespan.on_event("shutdown")
async def shutdown():
pass
# Imperative syntax:
async def extra_startup():
pass
lifespan.add_event_handler("startup", extra_startup)
# Wrapping an app:
app: "ASGIApp"
app = LifespanMiddleware(app, lifespan=lifespan)
I'm having trouble with tests. I attached the log of "make test" command. Could it be that pytest-4 is too old and it requires pytest-6 (upgrade is in progress on FreeBSD bugzilla)?
asgi-lifespan.log
Fires lifespan.startup
and lifespan.shutdown
events into an ASGI app.
Proposed API:
from asgi_lifespan import LifespanManager
async def main():
async with LifespanManager(app):
# Sent "lifespan" and "lifespan.startup" in the background
pass
# Sent "lifespan.shutdown"
The background task requires actual asynchronous I/O to be performed, and we shouldn't make too many assumptions on which environment users are running in.
Let's rely on anyio internally — it gives us asyncio + trio + curio support without any other extra dependencies.
anyio
was introduced to quickly get something running with support for both asyncio, trio and curio.
Curio isn't that well supported in the ASGI ecosystem anyway, so we may as well just drop it for now.
This leaves us with asyncio and trio.
We only use anyio
in LifespanManager
, and we use the following primitives:
Now, it's definitely a lot. But apart from task groups, all primitives are already present in both asyncio and trio, and we can easily build an abstraction layer on top of them, e.g. ConcurrencyBackend
, like what HTTPX does.
Asyncio doesn't have task groups, but the good news is we don't need to reimplement the whole thing. We only use a task group in LifespanManager
to run a coroutine in the background while we do stuff in the foreground. This can almost always be rewritten to use asyncio.gather()
. The subtlety here is that we yield
in the foreground. But there might be a way to rewrite the whole thing not to have the yield
in the way (and break the context-managed usage internally). I've got some insights on this based on encode/httpx#526.
So, hmm, we might look into this at some point? It's much easier because we don't actually perform any I/O (no pun intended), only coroutine-level concurrency.
A standalone ASGI app with event handler registration support.
Proposed API:
from asgi_lifespan import Lifespan
lifespan = Lifespan()
@lifespan.on_event("startup")
async def startup():
print("Starting up...")
@lifespan.on_event("shutdown")
async def shutdown():
print("Shutting down...")
def more_shutdown():
print("Bye!")
lifespan.add_event_handler("shutdown", more_shutdown)
As an ASGI app, it can be served using an ASGI server:
uvicorn app:lifespan
```console
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)
^CINFO: Shutting down
INFO: Waiting for application shutdown.
Shutting down...
Bye!
INFO: Finished server process [2407]
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.