FastAPI 生命周期与依赖状态

示例说明

每个请求各自创建 HTTP 客户端会浪费连接池，全局裸建又没有关闭时机。lifespan 异步上下文管理器正好补齐生命周期：yield 之前是启动逻辑，之后是关闭逻辑，框架保证优雅停机时一定执行。资源挂到 app.state，再通过依赖注入函数按类型分发给路由——路由声明什么依赖就拿到什么资源。

技术要点

• lifespan 的 yield 前后对应启停
• app.state 挂载跨请求共享资源
• 依赖注入按类型分发资源

代码解读

lifespan 里创建带超时与连接上限的客户端及有界队列；get_client 与 get_queue 两个注入器从请求的应用状态取资源。价格路由注入客户端代理外部行情，告警路由注入队列做入队，满队时返回背压信号而不是静默丢弃。

使用建议

共享资源要考虑并发安全，队列消费者另行启动。优雅停机顺序：先停流量再关资源。客户端超时与连接数按下游容量规划。

代码示例

import asyncio
from contextlib import asynccontextmanager

import httpx
from fastapi import Depends, FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(
        timeout=httpx.Timeout(10.0),
        limits=httpx.Limits(max_connections=50),
    )
    app.state.queue = asyncio.Queue(maxsize=1000)
    print('resources ready')
    yield
    await app.state.http.aclose()
    print('resources released')


app = FastAPI(lifespan=lifespan)


async def get_client(request) -> httpx.AsyncClient:
    return request.app.state.http


async def get_queue(request) -> asyncio.Queue:
    return request.app.state.queue


@app.get('/prices/{symbol}')
async def price(symbol: str, client: httpx.AsyncClient = Depends(get_client)):
    resp = await client.get(f'https://api.example.com/p/{symbol}')
    return {'symbol': symbol, 'data': resp.json()}


@app.post('/alerts')
async def enqueue(payload: dict, queue: asyncio.Queue = Depends(get_queue)):
    if queue.full():
        return {'queued': False, 'reason': 'backpressure'}
    await queue.put(payload)
    return {'queued': True, 'depth': queue.qsize()}