FastAPI实战：从踩坑到开发一个高性能API服务
去年我接手一个项目，Flask写的文件上传服务，并发一上来就卡死。同事说要不试试FastAPI？我心想，不就是个Python框架嘛，跟Flask能差多少——结果连踩三个坑才明白，这玩意儿的设计哲学完全不同。
性能是真香。同样的逻辑，QPS从200飙到1200。，异步原生支持就是好使。
*图1：FastAPI与Flask性能对比（基于我的测试数据）*
第一步：项目结构别乱搭
我一开始也是奔着写路由就去了，结果项目越写越大，连自己都找不到代码在哪儿。后来踩的坑多了，发现这样搞最省心：
“
myapi/
├── app/
│ ├── __init__.py
│ ├── main.py # 入口
│ ├── routers/ # 路由模块
│ │ ├── __init__.py
│ │ ├── users.py
│ │ └── items.py
│ ├── models/ # 数据模型
│ │ ├── __init__.py
│ │ └── sche
<
p>mas.py
│ ├── services/ # 业务逻辑
│ │ ├── __init__.py
│ │ └── user_service.py
│ └── dependencies/ # 依赖注入
│ └── __init__.py
├── requirements.txt
└── main.py # 启动脚本
`
为啥这么写？FastAPI的依赖注入系统（Dependency Injection）特别适合模块化。路由、业务、数据模型分开，后面加功能就跟搭积木一样——一开始我嫌麻烦没这么干，后来重构时真服了。
第一个坑：异步函数别乱用
看这段代码，我刚开始就是这么写的，结果性能还不如同步：
`python
错误示范：假异步
from fastapi import FastAPI
import asyncio
import time
app = FastAPI()
>
@app.get("/slow")
async def slow_endpoint():
time.sleep(5) # 阻塞操作！
return {"message": "done"}
`
这个写法导致事件循环直接卡住，等于白搭。正确写法：
`python
正确示范：真异步
from fastapi import FastAPI
import asyncio
app = FastAPI()
@app.get("/fast")
async def fast_endpoint():
await asyncio.sleep(5) # 非阻塞
return {"message": "done"}
`
或者，如果你的业务逻辑是CPU密集型的（比如图片处理），就老老实实用同步函数：
`python`
@app.get("/sync")
def sync_endpoint():
# 这里做CPU密集操作，FastAPI会自动用线程池处理
result = heavy_computation()
return {"result": result}
一开始搞错了，后来发现同步函数反而更快——因为线程池不会阻塞事件循环。
第二步：Pydantic模型是核心
FastAPI最大的亮点就是基于Pydantic的请求验证和文档生成。刚开始我写API时，参数校验全靠手写if判断，代码又长又丑。后来发现Pydantic真香：
`python
from pydantic import BaseModel, Field
from typing import Optional
from datetime import datetime
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50, description="用户名，3-50字符")
email: str = Field(..., pattern=r"^[\w\.-]+@[\w\.-]+\.\w+$", description="邮箱格式")
age: Optional[int] = Field(None, ge=0, le=150, description="年龄，0-150")
created_at: datetime = Field(default_factory=datetime.now)
在路由里直接使用
@app.post("/users/", response_model=UserCreate)
async def create_user(user: UserCreate):
# 自动验证，自动转换类型
return user
`
注意那个response_model=UserCreate，这是另一个坑。官方文档这段写得我看了三遍才明白——response_model不是让你返回同类型数据，而是控制API文档和输出过滤。比如你数据库里有密码字段，但不想返回给前端，就在response_model里不定义这个字段。
*图2：Pydantic模型验证错误示例*
第三个坑：依赖注入别滥用
依赖注入（Dependency Injection）是FastAPI的杀手锏，但用不好就变成灾难。比如这个：
`python
糟糕的设计：每个路由都重复写数据库session
from fastapi import Depends
from sqlalchemy.orm import Session
def get_db():
db = Session()
try:
yield db
finally:
db.close()
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: Session = Depends(get_db)):
# 业务逻辑
pass
@app.post("/users/")
async def create_user(user: UserCreate, db: Session = Depends(get_db)):
# 业务逻辑
pass
`
这看起来没问题，但如果有10个路由，就得重复写10次Depends(get_db)。更优雅的做法是用全局依赖：
`python
from fastapi import FastAPI, Depends
from typing import Annotated
db_dependency = Annotated[Session, Depends(get_db)]
app = FastAPI(dependencies=[Depends(get_db)])
@app.get("/users/{user_id}")
async def get_user(user_id: int, db: db_dependency):
pass
@app.post("/users/")
async def create_user(user: UserCreate, db: db_dependency):
pass
`
Annotated类型提示是Python 3.9+的特性，配合FastAPI的依赖注入，代码简洁到飞起。我团队里5个人，一开始大家都嫌麻烦不用，后来统一改成这个写法，代码量直接砍了一半。
性能优化：从3.2秒到0.8秒
实战中我遇到一个场景：用户请求一个列表，需要从数据库查数据，再调用两个外部API增强数据。原始代码：
`python`
@app.get("/items/")
async def get_items():
items = await db.query(Item).all()
for item in items:
item.extra1 = await call_external_api1(item.id)
item.extra2 = await call_external_api2(item.id)
return items
性能测试：100个请求，平均耗时3.2秒。原因：for循环里串行等待API响应。
优化后：
`python
import asyncio
@app.get("/items/")
async def get_items():
items = await db.query(Item).all()
# 并发调用外部API
tasks = []
for item in items:
tasks.append(
asyncio.gather(
call_external_api1(item.id),
call_external_api2(item.id)
)
)
results = await asyncio.gather(*tasks)
for item, (extra1, extra2) in zip(items, results):
item.extra1 = extra1
item.extra2 = extra2
return items
`
结果：平均耗时0.8秒，性能提升4倍。优化后同事还问我是不是换了服务器——其实就是把for循环改成了asyncio.gather。
*图3：性能优化前后的耗时对比柱状图*
还有一个技巧：中间件别乱写
很多人写中间件时，喜欢在里面做耗时操作，结果整个API都变慢了。记住：中间件是同步执行的，除非你用了BaseHTTPMiddleware。
`python
from fastapi import FastAPI, Request
from starlette.middleware.base import BaseHTTPMiddleware
import time
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start = time.time()
response = await call_next(request)
duration = time.time() - start
print(f"Request {request.url.path} took {duration:.2f}s")
return response
app = FastAPI()
app.add_middleware(TimingMiddleware)
“
这个中间件打印每个请求的耗时，调试时特别有用。但别在里面加数据库查询，我试过，每次请求多花200毫秒，真服了。
对了，还有个事儿——如果你想用ORM，推荐SQLAlchemy 2.0+配合FastAPI，异步支持做得不错。但别用Django ORM，兼容性太差了。