FastAPI 新手避坑指南

快速开发的同时，也要避开这些坑

FastAPI以其现代、高性能的特点深受开发者喜爱，特别是它的异步支持和自动API文档生成功能。但对于刚接触FastAPI的开发者来说，一些常见的陷阱可能会影响应用的性能和稳定性。本文将介绍FastAPI开发中的十大常见错误及如何避免它们。

1. 在异步环境中使用同步I/O操作

• 问题：在async def端点中调用requests.get()或time.sleep()等同步I/O操作，会导致整个事件循环阻塞。
• 影响：阻塞调用会暂停整个事件循环，影响其他请求的处理，大幅降低应用的并发能力。
• 解决方案：使用真正的异步库或将同步操作转移到线程池。

from fastapi import FastAPI
from fastapi.concurrency import run_in_threadpool
import httpx
import time

app = FastAPI()

@app.get("/good")
asyncdefgood_endpoint():
asyncwith httpx.AsyncClient(timeout=5) as client:
        response = await client.get("https://example.com")
return {"length": len(response.text)}

@app.get("/okish")
asyncdefokish_endpoint():
# 如果必须调用同步代码，使用线程池
return {"done": await run_in_threadpool(time.sleep, 1)}

2. 忽略Pydantic模型和response_model

• 问题：直接返回原始字典或接受任意JSON，放弃数据验证和类型安全。
• 影响：失去自动验证、准确的API文档和类型安全，容易产生运行时错误。
• 解决方案：明确定义请求和响应模型。

from pydantic import BaseModel, Field

classUserIn(BaseModel):
    email: str
    name: str = Field(min_length=1)

classUserOut(BaseModel):
    id: int
    email: str
    name: str

@app.post("/users", response_model=UserOut, status_code=201)
asyncdefcreate_user(payload: UserIn):
    user = {"id": 1, **payload.model_dump()}
return user

使用Pydantic模型可以在边界守护你的数据逻辑，保持核心业务代码的简洁性。

3. 数据库会话管理不当

• 问题：数据库连接堆积，事务长时间不关闭，出现"too many connections"错误。
• 影响：每个请求都需要一个短生命周期的会话，并确保最终能清理资源。
• 解决方案：使用带yield的依赖项管理会话生命周期。

from fastapi import Depends
from sqlalchemy.orm import Session

defget_db_session():
    db = SessionLocal()
try:
yield db
        db.commit()
except Exception:
        db.rollback()
raise
finally:
        db.close()

@app.get("/orders/{order_id}")
defget_order(order_id: int, db: Session = Depends(get_db_session)):
return db.get(Order, order_id)

这种模式确保即使在发生错误时，也能正确执行提交/回滚/关闭操作。

4. 每次请求都创建客户端/连接池

• 问题：在端点内部实例化httpx.AsyncClient()或创建数据库引擎。
• 影响：额外的握手开销，套接字浪费和内存抖动。
• 解决方案：使用应用生命周期管理共享资源。

from contextlib import asynccontextmanager
import httpx

@asynccontextmanager
asyncdefapp_lifespan(app: FastAPI):
# 启动时创建资源
    app.state.http_client = httpx.AsyncClient(timeout=5)
yield
# 关闭时清理资源
await app.state.http_client.aclose()

app = FastAPI(lifespan=app_lifespan)

@app.get("/health")
asyncdefhealth_check():
    response = await app.state.http_client.get("https://example.com/health")
return {"status": "healthy"if response.status_code == 200else"unhealthy"}

5. CORS配置错误

• 问题：同时设置allow_origins=["*"]和allow_credentials=True。
• 影响：浏览器会拒绝这种不安全的组合，导致CORS错误。
• 解决方案：明确指定允许的源。

from fastapi.middleware.cors import CORSMiddleware

allowed_origins = ["https://app.company.com", "https://staging.company.com"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=allowed_origins,
    allow_credentials=True,
    allow_methods=["GET", "POST", "PUT", "DELETE"],
    allow_headers=["*"],
)

CORS应该被视为安全边界，而不是事后考虑的内容。

6. 所有代码都在一个文件中

• 问题：处理器、模型和工具类都放在单个文件中，修改时容易引发不可预知的问题。
• 影响：耦合度高，测试和扩展困难。
• 解决方案：使用路由器和包组织代码。

app/
 ├─ main.py          # 应用工厂、生命周期、中间件
 ├─ api/
 │   ├─ __init__.py
 │   ├─ users.py     # 用户相关路由
 │   └─ orders.py    # 订单相关路由
 ├─ models/          # Pydantic模型
 ├─ db/              # 数据库引擎、会话、模型
 └─ dependencies/    # 依赖项

在users.py中：

from fastapi import APIRouter

router = APIRouter(prefix="/users", tags=["users"])

@router.get("/{user_id}")
defget_user(user_id: int):
# 业务逻辑
return {"user_id": user_id}

在main.py中包含路由器：

from app.api import users, orders

app.include_router(users.router)
app.include_router(orders.router)

7. 直接返回ORM对象

• 问题：直接返回SQLAlchemy模型实例，序列化时会出现奇怪问题。
• 影响：ORM对象不是直接的JSON，可能泄露敏感字段或遇到递归问题。
• 解决方案：使用Pydantic模型并从属性序列化。

from pydantic import BaseModel, ConfigDict

classUserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    email: str
    name: str

@app.get("/users/{user_id}", response_model=UserResponse)
defget_user(user_id: int, db: Session = Depends(get_db_session)):
    user = db.get(User, user_id)
return user

明确指定要暴露的字段，不多不少刚刚好。

8. 弱认证和自制安全方案

• 问题：自己实现令牌逻辑，不验证权限范围，所有路由都是公开的。
• 影响：安全债务会快速累积。
• 解决方案：使用依赖项集中处理认证。

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

asyncdefget_current_user(token: str = Depends(oauth2_scheme)):
    user = verify_token(token)  # 你的令牌验证逻辑
ifnot user:
raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
return user

@app.get("/users/me")
asyncdefread_current_user(current_user: User = Depends(get_current_user)):
return {"id": current_user.id, "email": current_user.email}

使用依赖注入在路由或整个路由器级别强制执行认证。

9. 对外部服务调用缺少超时、重试和限流

• 问题：下游服务延迟会级联影响到你的API，线程堆积，延迟激增。
• 解决方案：结合连接池、超时和退避策略。

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=0.2, max=2))
asyncdeffetch_data(client: httpx.AsyncClient, url: str):
    response = await client.get(url, timeout=httpx.Timeout(2.0, connect=1.0))
    response.raise_for_status()
return response.json()

@app.get("/external-data")
asyncdefget_external_data(http_client: httpx.AsyncClient = Depends(get_http_client)):
    data = await fetch_data(http_client, "https://api.external.com/data")
return {"data": data}

有界的重试机制加上严格的超时设置可以防止一个依赖服务拖垮整个应用。

10. 忽视测试和错误处理

• 问题：使用cURL手动测试，祈祷一切正常。
• 影响：难以进行重构，错误信息不友好。
• 解决方案：使用TestClient、依赖覆盖和异常处理器。

from fastapi.testclient import TestClient

defoverride_get_current_user():
return FakeUser(id=1, email="[email protected]")

app.dependency_overrides[get_current_user] = override_get_current_user

client = TestClient(app)

deftest_read_current_user():
    response = client.get("/users/me")
assert response.status_code == 200
assert response.json()["email"] == "[email protected]"

添加自定义异常处理器，提供一致的错误响应：

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(ValueError)
asyncdefvalue_error_handler(request: Request, exc: ValueError):
return JSONResponse(
        status_code=400,
        content={"error": str(exc)},
    )

@app.exception_handler(UnicornException)
asyncdefunicorn_exception_handler(request: Request, exc: UnicornException):
return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )

测试为重构提供安全保障，异常处理器防止嘈杂的堆栈跟踪泄露给客户端。

快速检查清单

• 使用真正的异步库；将不可避免的同步工作卸载到线程池
• 使用Pydantic模型和response_model进行验证
• 通过yield依赖项管理数据库会话
• 在生命周期中创建共享的客户端/连接池
• 锁定CORS策略
• 使用路由器和包组织代码
• 使用from_attributes=True序列化DTO
• 通过依赖项强制执行认证
• 为外部调用添加超时/重试机制
• 编写测试和异常处理器

写在最后

FastAPI提供了极大的灵活性，关键在于如何正确利用这些特性。提前加强这十个方面的实践，可以避免昂贵的重构，保持低延迟预算。希望这些经验能帮助你构建更健壮、高效的FastAPI应用！

你是否有自己的FastAPI"救命"技巧？欢迎在评论区分享～