启动和关闭#

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_servingafter_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_servingafter_serving 函数已被调用,并且 while_serving 生成器已被推进。

async def test_index(app):
    test_client = app.test_client()
    await test_client.get("/")
    ...