生命周期和测试

有些代码需要在应用启动时执行一次,在应用关闭时清理。

例如:

  • 创建数据库连接池
  • 加载机器学习模型
  • 初始化缓存客户端
  • 关闭外部资源连接

FastAPI 当前推荐使用 lifespan 管理这些逻辑。

一、lifespan 基本写法

from contextlib import asynccontextmanager

from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("应用启动")
    app.state.cache = {}
    yield
    print("应用关闭")
    app.state.cache.clear()


app = FastAPI(lifespan=lifespan)


@app.get("/")
def root():
    return {"message": "Hello FastAPI"}

执行顺序:

启动应用

执行 yield 前面的代码

处理请求

应用关闭

执行 yield 后面的代码

旧教程里常见的 @app.on_event("startup")@app.on_event("shutdown") 已不再是推荐写法,新项目优先使用 lifespan

二、app.state

app.state 可以保存应用级别的共享对象。

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.items = {"hello": "world"}
    yield
    app.state.items.clear()

路由里读取:

from fastapi import Request


@app.get("/items/{key}")
def get_item(key: str, request: Request):
    return {"value": request.app.state.items.get(key)}

注意:app.state 适合放共享资源,不适合放每个用户自己的数据。

三、为什么要写测试

接口测试可以帮你确认:

  • 路由地址没有写错
  • 参数校验符合预期
  • 成功响应结构正确
  • 错误状态码正确

FastAPI 提供 TestClient,可以不用真正启动服务器就测试接口。

四、安装 pytest

pip install pytest

如果使用 fastapi[standard],接口测试需要的基础 HTTP 客户端依赖通常已经具备。

五、最简单的测试

main.py

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def root():
    return {"message": "Hello FastAPI"}

test_main.py

from fastapi.testclient import TestClient

from main import app

client = TestClient(app)


def test_root():
    response = client.get("/")

    assert response.status_code == 200
    assert response.json() == {"message": "Hello FastAPI"}

运行:

pytest

六、测试参数校验

@app.get("/users/{user_id}")
def get_user(user_id: int):
    return {"user_id": user_id}

测试:

def test_get_user():
    response = client.get("/users/1")

    assert response.status_code == 200
    assert response.json() == {"user_id": 1}


def test_get_user_with_invalid_id():
    response = client.get("/users/abc")

    assert response.status_code == 422

七、测试 POST 请求

from pydantic import BaseModel


class UserCreate(BaseModel):
    username: str
    age: int


@app.post("/users/")
def create_user(user: UserCreate):
    return {"id": 1, **user.model_dump()}

测试:

def test_create_user():
    response = client.post(
        "/users/",
        json={"username": "zhangsan", "age": 18},
    )

    assert response.status_code == 200
    assert response.json() == {
        "id": 1,
        "username": "zhangsan",
        "age": 18,
    }

八、测试 lifespan

如果要让测试触发生命周期,使用 with TestClient(app)

from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.testclient import TestClient


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.ready = True
    yield
    app.state.ready = False


app = FastAPI(lifespan=lifespan)


@app.get("/ready")
def ready(request: Request):
    return {"ready": request.app.state.ready}


def test_lifespan():
    with TestClient(app) as client:
        response = client.get("/ready")
        assert response.status_code == 200
        assert response.json() == {"ready": True}

九、测试文件放哪里

推荐结构:

fastapi-demo/
├── app/
│   ├── __init__.py
│   └── main.py
└── tests/
    ├── __init__.py
    └── test_main.py

运行:

pytest

初学阶段不用追求测试覆盖率很高,但至少给核心接口写几个成功和失败用例。