如果你在2026 年想用 Python 写高性能、现代、好维护的 API，FastAPI 几乎是唯一主流选择。我从零开始折腾了一圈，把踩过的坑和核心要点都记了下来，从环境搭建到生产部署，全是能直接上手的干货。
一、为什么是 FastAPI？
FastAPI 基于 Starlette（负责底层网络）和 Pydantic（负责数据校验），性能接近 Node.js 和 Go，开发效率比 Flask 高出一大截。最爽的是，你写完代码，Swagger 文档和 ReDoc 就自动生成了，连接口测试都不用离开浏览器。
二、五分钟跑通第一个 API
环境要求 Python 3.8+，强烈建议用虚拟环境隔离项目。
# 创建虚拟环境 python -m venv fastapi_env # 激活 (Linux/macOS) source fastapi_env/bin/activate # Windows: fastapi_env\Scripts\activate # 安装 FastAPI 和服务器 pip install fastapi uvicorn
新建 main.py，写入最简单的接口：
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "FastAPI"}
@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
启动服务：
uvicorn main:app --reload
浏览器访问 http://localhost:8000/docs，你会看到自动生成的 Swagger UI，直接在页面上测试 GET /items/5?q=abc 这种感觉太棒了。
三、核心概念：路径参数、查询参数与请求体
路径参数：URL 中的变量，比如 /users/123 里的 123。FastAPI 会根据类型声明自动解析和报错。
@app.get("/users/{user_id}")
async def read_user(user_id: int):
return {"user_id": user_id}
查询参数：问号后面的键值对，比如 /items?limit=10。非必填参数直接给默认值。
from fastapi import Query
@app.get("/items/")
async def read_items(
item_id: int = Query(..., description="商品ID"),
q: str = None,
limit: int = 10
):
return {"item_id": item_id, "q": q, "limit": limit}
请求体：POST/PUT 时放在 body 里的 JSON，用 Pydantic 模型声明结构。
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.post("/items/")
async def create_item(item: Item):
item_dict = item.dict()
if item.tax:
item_dict["price_with_tax"] = item.price + item.tax
return item_dict
四、Pydantic 模型与数据验证
Pydantic 是 FastAPI 的校验灵魂。除了基础类型，还能用 EmailStr、Field 加约束，甚至写自定义验证器。
from pydantic import BaseModel, EmailStr, Field, validator
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=50)
email: EmailStr
password: str = Field(..., min_length=8)
@validator('password')
def password_strength(cls, v):
if not any(c.isdigit() for c in v):
raise ValueError('密码必须包含数字')
return v
如果请求体不符合规则，FastAPI 会自动返回 422 错误和清晰的字段提示。
五、响应模型与状态码
用 response_model 过滤掉敏感字段（比如密码），还能保证输出格式符合文档。
class UserOut(BaseModel):
username: str
email: str
# 注意没有 password
@app.post("/users/", response_model=UserOut, status_code=201)
async def create_user(user: UserCreate):
# 假设保存到数据库，返回新用户信息
return {"username": user.username, "email": user.email}
六、依赖注入：写优雅代码的核心
依赖注入（DI）让你把共用的逻辑（数据库会话、认证、配置）抽出来，路由函数只关心业务。
from fastapi import Depends
def get_db():
db = create_connection()
try:
yield db
finally:
db.close()
@app.get("/items/")
def read_items(db = Depends(get_db)):
return db.query(Item).all()
你还可以做类级别的依赖，或者在整个路径上挂载依赖（比如全局校验 token）。
async def verify_token(x_token: str = Header(...)):
if x_token != "secret":
raise HTTPException(400, "无效Token")
@app.get("/secure/", dependencies=[Depends(verify_token)])
async def secure_endpoint():
return {"message": "验证通过"}
七、异步 vs 同步：到底用不用 async def？
如果你的路由里调用了异步数据库驱动、异步 HTTP 客户端（比如 httpx.AsyncClient），就用 async def + await。如果只是纯计算或同步 I/O，用普通 def 就行，FastAPI 会自动用线程池处理，不会阻塞事件循环。
import httpx
@app.get("/async-data")
async def get_async_data():
async with httpx.AsyncClient() as client:
resp = await client.get("https://api.example.com/data")
return resp.json()
注意：在 async 路由里直接调用同步阻塞函数（比如 time.sleep(5)）会卡死整个服务。此时要用 run_in_threadpool 把它丢到线程池。
from fastapi.concurrency import run_in_threadpool
@app.get("/compute")
async def heavy():
result = await run_in_threadpool(slow_sync_function)
return {"result": result}
八、数据库集成（异步 SQLAlchemy 2.0+）
生产环境强烈推荐异步驱动（如 asyncpg）+ SQLAlchemy 2.0 的异步特性。下面是一个标准模板：
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
Base = declarative_base()
# 依赖获取 session
async def get_db():
async with AsyncSessionLocal() as session:
yield session
@app.get("/users/")
async def get_users(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User))
return result.scalars().all()
如果你更喜欢 Tortoise-ORM 那种 Django 风格，也可以直接上 Tortoise，它对异步的支持非常友好。
九、安全认证：JWT + OAuth2 密码流
FastAPI 内置了 OAuth2PasswordBearer，配合 python-jose 和 passlib，半小时就能搭一个靠谱的登录体系。
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import jwt
from passlib.context import CryptContext
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_password(plain, hashed):
return pwd_context.verify(plain, hashed)
def get_password_hash(password):
return pwd_context.hash(password)
def create_access_token(data: dict):
to_encode = data.copy()
expire = datetime.utcnow() + timedelta(minutes=30)
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
# 验证用户名密码（从数据库取用户）
if not verify_password(form.password, fake_hashed):
raise HTTPException(401, "用户名或密码错误")
access_token = create_access_token(data={"sub": form.username})
return {"access_token": access_token, "token_type": "bearer"}
async def get_current_user(token: str = Depends(oauth2_scheme)):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload.get("sub")
except JWTError:
raise HTTPException(401, "无效token")
@app.get("/users/me")
async def me(current_user: str = Depends(get_current_user)):
return {"username": current_user}
十、WebSocket 实时通信
FastAPI 对 WebSocket 的支持也很简洁。写一个简单的聊天室：
from fastapi import WebSocket, WebSocketDisconnect
class ConnectionManager:
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: str):
for conn in self.active_connections:
await conn.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/chat")
async def chat(websocket: WebSocket):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"用户说: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
await manager.broadcast("用户离开了聊天室")
十一、后台任务（BackgroundTasks）
发送邮件、写日志这种不需要用户等待的操作，丢给 BackgroundTasks 最合适。
from fastapi import BackgroundTasks
def send_email(email: str, subject: str, body: str):
# 模拟耗时
print(f"发送邮件到 {email}")
@app.post("/users/")
async def create_user(user: UserCreate, background_tasks: BackgroundTasks):
user_id = save_to_db(user)
background_tasks.add_task(send_email, user.email, "欢迎", "感谢注册")
return {"user_id": user_id}
十二、中间件与 CORS
解决跨域问题只需要加一行 CORS 中间件：
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["https://your-frontend.com", "http://localhost:3000"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
你也可以写自定义中间件，比如记录每个请求的耗时：
@app.middleware("http")
async def log_request(request: Request, call_next):
start = time.time()
response = await call_next(request)
process = time.time() - start
response.headers["X-Process-Time"] = str(process)
return response
十三、错误处理与自定义异常
用 HTTPException 抛出标准错误，或者注册全局异常处理器统一返回格式。
from fastapi import HTTPException, status
@app.get("/users/{uid}")
async def get_user(uid: int):
user = await find_user(uid)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user
# 自定义业务异常
class BizException(Exception):
def __init__(self, code: int, msg: str):
self.code = code
self.msg = msg
@app.exception_handler(BizException)
async def biz_handler(request: Request, exc: BizException):
return JSONResponse(status_code=400, content={"code": exc.code, "msg": exc.msg})
十四、测试（TestClient）
FastAPI 自带 TestClient，基于 requests，不需要启动服务器就能测。
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_read_root():
resp = client.get("/")
assert resp.status_code == 200
assert resp.json() == {"Hello": "FastAPI"}
def test_create_item():
payload = {"name": "book", "price": 29.9}
resp = client.post("/items/", json=payload)
assert resp.status_code == 201
十五、项目结构最佳实践
当代码超过几百行，建议按领域拆分：
my_project/ ├── app/ │ ├── main.py │ ├── config.py │ ├── database.py │ ├── api/ │ │ └── v1/ │ │ ├── endpoints/ │ │ │ ├── users.py │ │ │ └── items.py │ │ └── router.py │ ├── models/ # Pydantic schemas │ ├── schemas/ # SQLAlchemy models │ ├── services/ # 业务逻辑层 │ └── dependencies.py ├── tests/ └── requirements.txt
这种结构让路由保持轻量，业务逻辑放在 service 层，便于复用和测试。
十六、部署与性能优化
开发时用 uvicorn main:app --reload，生产环境推荐 Gunicorn + Uvicorn workers：
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000 --preload
-w 4：worker 数一般等于 CPU 核心数。--preload：在 fork 子进程前加载代码，减少内存占用。- 异步 worker 可以处理大量并发连接，即使 worker 数量不多。
其他优化小贴士：
- 使用异步数据库驱动（asyncpg, aiomysql）。
- 对频繁查询的数据加 Redis 缓存。
- Pydantic v2 比 v1 快很多，保持升级。
- 生产环境关闭 debug，配置 CORS 白名单。
- 添加限流中间件（如 slowapi）防止滥用。
写在最后
FastAPI 的学习曲线其实很平滑，只要掌握类型提示和依赖注入，剩下的就是照着官方文档堆功能。这套笔记覆盖了我从零到上线的所有关键节点，建议你一边看一边敲代码，把每个代码段都跑通。当你看到自动生成的 Swagger 文档那一刻，会觉得这一切都值了。
如果遇到问题，优先查官方文档，里面几乎有所有场景的示例。也欢迎在评论区交流实际踩过的坑。