Uvicorn Graceful Shutdown Workers

import asyncio
import contextlib

from fastapi import FastAPI

app = FastAPI()
shutdown_event = asyncio.Event()


@app.get('/health')
async def health():
    if shutdown_event.is_set():
        return {'status': 'draining'}
    return {'status': 'ready'}


@app.on_event('shutdown')
async def release_resources():
    print('draining in-flight requests')
    await asyncio.sleep(0.2)
    print('background loops cancelled')


async def background_sync():
    while not shutdown_event.is_set():
        await asyncio.sleep(5)
        print('sync tick')
    print('background loop exited cleanly')


@contextlib.asynccontextmanager
async def runner():
    task = asyncio.create_task(background_sync())
    try:
        yield
    finally:
        shutdown_event.set()
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task


if __name__ == '__main__':
    # 关键在于命令行参数：
    # uvicorn app:app --timeout-graceful-shutdown 20
    # 收到 SIGTERM 后：停止接新请求 → 排空在途请求 → 最多等 20 秒 → 强制退出
    import uvicorn

    uvicorn.run(
        'app:app',
        host='0.0.0.0',
        port=8000,
        timeout_graceful_shutdown=20,
        timeout_keep_alive=5,
    )