启动和关闭#
The ASGI lifespan specification includes the ability for awaiting coroutines before the first byte is received and after the final byte is sent, through the startup
and shutdown
lifespan events. This is particularly useful for creating and destroying connection pools. Quart supports this via the decorators before_serving()
, after_serving()
, and while_serving()
which expects a function that returns a generator.
The decorated functions are all called within the app context, allowing current_app
and g
to be used.
警告
谨慎使用 g
,因为它会在启动后重置,即所有 before_serving
函数完成之后,以及 while serving 生成器中的初始 yield 之后。它仍然可以在此上下文中使用。如果您想创建在路由中使用的内容,请尝试将其存储在应用程序中。
要使用此功能,只需执行以下操作
@app.before_serving
async def create_db_pool():
app.db_pool = await ...
g.something = something
@app.before_serving
async def use_g():
g.something.do_something()
@app.while_serving
async def lifespan():
... # startup
yield
... # shutdown
@app.route("/")
async def index():
app.db_pool.execute(...)
# g.something is not available here
@app.after_serving
async def create_db_pool():
await app.db_pool.close()
测试#
Quart 的测试客户端基于请求生命周期,因此不会调用 before_serving
或 after_serving
函数,也不会推进 while_serving
生成器。相反,Quart 的测试应用程序可以被使用,例如
@pytest.fixture(name="app", scope="function")
async def _app():
app = create_app() # Initialize app
async with app.test_app() as test_app:
yield test_app
应用程序夹具可以像往常一样使用,因为 before_serving
和 after_serving
函数已被调用,并且 while_serving
生成器已被推进。
async def test_index(app):
test_client = app.test_client()
await test_client.get("/")
...