中文
devexamples

FastAPI Lifespan Dependency State

Manage shared resources with FastAPI lifespan context, wiring async clients into dependency injection with clean startup and shutdown.

PythonFastapiConfig Management

Code

snippet
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()}

Related Examples