副标题: 90%的人不知道，FastAPI的性能比Flask快300倍
痛点：为什么你的Web API开发总是很慢？
2025年某团队用Flask开发API，响应时间平均500ms。问题出在哪？工程师没有使用现代异步框架。
真相：FastAPI基于Starlette和Pydantic，性能接近NodeJS和Go。
| 框架 | 性能 | 学习曲线 | 推荐度 |
|---|
| Flask | 1x | 低 | ⭐⭐⭐ |
| Django | 0.5x | 高 | ⭐⭐⭐ |
| FastAPI | 3-5x | 中 | ⭐⭐⭐⭐⭐ |
| Sanic | 4x | 中 | ⭐⭐⭐⭐ |
一、FastAPI基础
1.1 第一个应用
from fastapi import FastAPI
app = FastAPI(
title="我的API",
description="一个FastAPI示例项目",
version="1.0.0"
)
@app.get("/")
async def root():
return {"message": "Hello, World!"}
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
# 运行
uvicorn main:app --reload
访问自动文档
http://localhost:8000/docs # Swagger UI
http://localhost:8000/redoc # ReDoc
1.2 路径操作
from fastapi import FastAPI
app = FastAPI()
GET
@app.get("/items/{item_id}")
async def get_item(item_id: int):
return {"item_id": item_id}
POST
@app.post("/items/")
async def create_item(name: str, price: float):
return {"name": name, "price": price}
PUT
@app.put("/items/{item_id}")
async def update_item(item_id: int, name: str):
return {"item_id": item_id, "name": name}
DELETE
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
return {"item_id": item_id, "message": "Deleted"}
带查询参数
@app.get("/search")
async def search(q: str, limit: int = 10, skip: int = 0):
return {"q": q, "limit": limit, "skip": skip}
1.3 请求体
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
price: float
description: str | None = None
tags: list[str] = []
@app.post("/items/")
async def create_item(item: Item):
return item
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
return {"item_id": item_id, **item.model_dump()}
二、路径参数
2.1 基础路径参数
from fastapi import FastAPI
app = FastAPI()
@app.get("/users/{user_id}")
async def read_user(user_id: str):
return {"user_id": user_id}
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
2.2 路径参数验证
from fastapi import FastAPI, Path
from typing import Annotated
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(
item_id: Annotated[int, Path(title="物品ID", ge=0, le=1000)]
):
return {"item_id": item_id}
@app.get("/users/{user_id}")
async def read_user(
user_id: Annotated[str, Path(min_length=3, max_length=50, pattern=r"^[a-zA-Z0-9_-]+$")]
):
return {"user_id": user_id}
2.3 多个路径参数
@app.get("/users/{user_id}/items/{item_id}")
async def read_user_item(
user_id: str,
item_id: int,
q: str | None = None,
short: bool = False
):
item = {"item_id": item_id, "owner_id": user_id}
if q:
item.update({"q": q})
if short:
item = {"item_id": item_id, "owner_id": user_id}
return item
三、查询参数
3.1 基础查询参数
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/")
async def read_items(
q: str | None = None,
skip: int = 0,
limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
3.2 查询参数验证
from fastapi import FastAPI, Query
from typing import Annotated
app = FastAPI()
@app.get("/items/")
async def read_items(
q: Annotated[str | None, Query(min_length=3, max_length=50)] = None,
skip: Annotated[int, Query(ge=0)] = 0,
limit: Annotated[int, Query(ge=1, le=100)] = 100
):
return {"q": q, "skip": skip, "limit": limit}
3.3 列表查询参数
@app.get("/items/")
async def read_items(
q: list[str] | None = None,
skip: int = 0,
limit: int = 100
):
return {"q": q, "skip": skip, "limit": limit}
访问: /items/?q=foo&q=bar&q=baz
四、请求体与字段验证
4.1 Pydantic模型
from fastapi import FastAPI
from pydantic import BaseModel, Field, field_validator
from typing import Annotated
app = FastAPI()
class Item(BaseModel):
name: Annotated[str, Field(min_length=1, max_length=50)]
price: Annotated[float, Field(gt=0)]
description: str | None = Field(None, max_length=500)
tags: list[str] = Field(default_factory=list)
in_stock: bool = True
@field_validator('name')
@classmethod
def validate_name(cls, v: str) -> str:
if not v.strip():
raise ValueError('名称不能为空')
return v.strip()
@app.post("/items/")
async def create_item(item: Item):
return item
4.2 嵌套模型
class Address(BaseModel):
street: str
city: str
zip_code: str
class User(BaseModel):
username: str
email: str
address: Address | None = None
@app.post("/users/")
async def create_user(user: User):
return user
4.3 多个请求体
from fastapi import FastAPI, Body
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
price: float
@app.put("/items/{item_id}")
async def update_item(
item_id: int,
item: Item,
q: str | None = None,
importance: int = Body(1, ge=1, le=5),
include_in_memory: bool = Body(False)
):
return {
"item_id": item_id,
"item": item,
"q": q,
"importance": importance,
"include_in_memory": include_in_memory
}
五、响应模型
5.1 响应模型过滤
from fastapi import FastAPI
from pydantic import BaseModel, SecretStr
app = FastAPI()
class UserIn(BaseModel):
username: str
password: SecretStr
email: str
full_name: str | None = None
class UserOut(BaseModel):
username: str
email: str
full_name: str | None = None
@app.post("/users/", response_model=UserOut)
async def create_user(user: UserIn):
# password不会被返回
return user
5.2 响应状态码
from fastapi import FastAPI, status
app = FastAPI()
@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: dict):
return item
@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
return None
5.3 自定义响应
from fastapi import FastAPI
from fastapi.responses import JSONResponse, RedirectResponse
app = FastAPI()
@app.get("/redirect")
async def redirect_example():
return RedirectResponse(url="/items/")
@app.get("/custom")
async def custom_response():
return JSONResponse(
content={"message": "Custom response"},
status_code=200,
headers={"X-Custom-Header": "value"}
)
六、依赖注入
6.1 基础依赖
from fastapi import FastAPI, Depends
app = FastAPI()
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
return commons
@app.get("/users/")
async def read_users(commons: dict = Depends(common_parameters)):
return commons
6.2 类依赖
from fastapi import FastAPI, Depends, HTTPException
app = FastAPI()
class CommonQueryParams:
def __init__(self, q: str | None = None, skip: int = 0, limit: int = 100):
self.q = q
self.skip = skip
self.limit = limit
@app.get("/items/")
async def read_items(commons: CommonQueryParams = Depends()):
return {
"q": commons.q,
"skip": commons.skip,
"limit": commons.limit
}
6.3 认证依赖
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
app = FastAPI()
security = HTTPBearer()
async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
token = credentials.credentials
# 验证token
if token != "valid-token":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid token"
)
return {"user_id": "user123", "token": token}
@app.get("/users/me")
async def read_users_me(current_user: dict = Depends(get_current_user)):
return current_user
6.4 依赖树
async def query_dependency(q: str | None = None):
return {"q": q}
async def pagination_dependency(skip: int = 0, limit: int = 100):
return {"skip": skip, "limit": limit}
async def common_dependency(
q_data: dict = Depends(query_dependency),
page_data: dict = Depends(pagination_dependency)
):
return {q_data, page_data}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_dependency)):
return commons
七、错误处理
7.1 HTTPException
from fastapi import FastAPI, HTTPException, status
app = FastAPI()
items = {"foo": "bar"}
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Item {item_id} not found",
headers={"X-Error": "Item not found"}
)
return {"item": items[item_id]}
7.2 自定义异常处理器
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
app = FastAPI()
class ItemNotFoundError(Exception):
def __init__(self, item_id: str):
self.item_id = item_id
@app.exception_handler(ItemNotFoundError)
async def item_not_found_handler(request: Request, exc: ItemNotFoundError):
return JSONResponse(
status_code=404,
content={"message": f"Item {exc.item_id} not found"}
)
@app.get("/items/{item_id}")
async def read_item(item_id: str):
if item_id not in items:
raise ItemNotFoundError(item_id)
return {"item": items[item_id]}
7.3 全局异常处理
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={
"message": "Internal server error",
"detail": str(exc) if app.debug else None
}
)
八、中间件
8.1 基础中间件
from fastapi import FastAPI, Request
import time
app = FastAPI()
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
return response
8.2 CORS中间件
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
8.3 日志中间件
import logging
from fastapi import FastAPI, Request
logger = logging.getLogger("uvicorn")
@app.middleware("http")
async def log_requests(request: Request, call_next):
start_time = time.time()
response = await call_next(request)
process_time = time.time() - start_time
logger.info(
f"{request.method} {request.url.path} - {response.status_code} - {process_time:.3f}s"
)
return response
九、实战案例
9.1 完整API项目结构
my_api/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── config.py
│ ├── database.py
│ ├── models.py
│ ├── schemas.py
│ ├── api/
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── items.py
│ │ │ └── users.py
│ │ └── deps.py
│ └── core/
│ ├── __init__.py
│ ├── security.py
│ └── exceptions.py
├── tests/
│ ├── __init__.py
│ ├── test_items.py
│ └── test_users.py
├── .env
├── pyproject.toml
└── README.md
9.2 用户管理系统
# app/schemas.py
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime
from typing import Annotated
class UserBase(BaseModel):
email: EmailStr
username: Annotated[str, Field(min_length=3, max_length=50)]
class UserCreate(UserBase):
password: Annotated[str, Field(min_length=8)]
class User(UserBase):
id: int
is_active: bool = True
created_at: datetime
class ItemBase(BaseModel):
name: str
description: str | None = None
price: float
class ItemCreate(ItemBase):
pass
class Item(ItemBase):
id: int
owner_id: int
created_at: datetime
# app/api/v1/items.py
from fastapi import APIRouter, Depends, HTTPException, status
from typing import Annotated
from app.schemas import Item, ItemCreate
from app.api import deps
router = APIRouter()
@router.post("/", response_model=Item, status_code=status.HTTP_201_CREATED)
async def create_item(
item_in: ItemCreate,
current_user: dict = Depends(deps.get_current_user)
):
return item_in
@router.get("/", response_model=list[Item])
async def read_items(
skip: int = 0,
limit: int = 100,
current_user: dict = Depends(deps.get_current_user)
):
return []
@router.get("/{item_id}", response_model=Item)
async def read_item(
item_id: int,
current_user: dict = Depends(deps.get_current_user)
):
return {"id": item_id, "name": "test"}
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(
item_id: int,
current_user: dict = Depends(deps.get_current_user)
):
return None
# app/main.py
from fastapi import FastAPI
from app.api.v1 import items, users
from app.core.config import settings
app = FastAPI(
title=settings.APP_NAME,
version="1.0.0"
)
app.include_router(items.router, prefix="/api/v1/items", tags=["items"])
app.include_router(users.router, prefix="/api/v1/users", tags=["users"])
@app.get("/health")
async def health_check():
return {"status": "healthy"}
十、测试
10.1 单元测试
from fastapi.testclient import TestClient
from app.main import app
client = TestClient(app)
def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello, World!"}
def test_create_item():
response = client.post(
"/items/",
json={"name": "foo", "price": 10.5}
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "foo"
assert data["price"] == 10.5
def test_read_item_not_found():
response = client.get("/items/999")
assert response.status_code == 404
10.2 依赖覆盖测试
from fastapi import Depends
from app.main import app
from app.api import deps
def override_get_current_user():
return {"user_id": "test_user", "is_admin": True}
app.dependency_overrides[deps.get_current_user] = override_get_current_user
def test_admin_endpoint():
response = client.get("/admin/users")
assert response.status_code == 200
app.dependency_overrides.clear()
常见坑自查清单
| 坑 | 现象 | 自查方法 | 修复方案 |
|---|
| 类型错误 | 422错误 | 检查Pydantic模型 | 添加类型注解 |
| 依赖未注入 | None值 | 检查Depends使用 | 正确使用Depends |
| 异步阻塞 | 性能差 | 检查async/await | 使用async函数 |
| 循环依赖 | 导入错误 | 检查模块结构 | 重构代码结构 |
结语
关键洞察：
-
●FastAPI性能优秀，基于异步-
●Pydantic提供强大的数据验证-
●依赖注入系统灵活强大-
●自动生成API文档-
互动
-
1.你用FastAPI开发过什么项目？-
2.依赖注入好用吗？-
3.遇到过性能问题吗？-
版本: V1.0 | 2026-05-26 | Python Web开发系列
📚 推荐阅读
📝 摘要：今天深入学习静态代码分析技术，这是安全审计的核心技能。从 Python AST 模块到检测模式设计，收获满满！
发布于 202603
01-Python 环境搭建与第一个脚本
发布于 202603
【优化】Python代码优化与调试技巧
发布于 202603
KEYWORDS
IL, Python, AI, 函数, 循环
💡 如果你觉得这篇文章有帮助，请点个在看，分享给更多需要的人！
📝 关注我，获取更多实用干货～
🤝 有问题欢迎评论区留言交流！
10个月宝宝每天需要喝多少奶粉？
