FastAPI:高性能的Web框架
高性能:基于Starlette和Pydantic,性能媲美NodeJS和开发效率:自动生成交互式文档,减少开发时间类型安全:基于Python类型提示,减少约40%的错误异步支持:原生支持异步编程,适合高并发场景标准化:基于OpenAPI和JSON Schema标准。
·
🚀 快速开始:你的第一个FastAPI应用
1. 环境搭建
# 创建项目目录
mkdir fastapi-demo && cd fastapi-demo
# 创建虚拟环境(推荐)
python -m venv venv
# 激活虚拟环境
# Windows: .\venv\Scripts\activate
# macOS/Linux: source venv/bin/activate
# 安装FastAPI及Uvicorn服务器
pip install fastapi uvicorn[standard]
2. 创建第一个应用
创建 main.py 文件:
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Hello World"}
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
3. 启动服务
# 方式一:使用uvicorn命令行
uvicorn main:app --reload
# 方式二:在代码中启动
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
4. 访问应用
打开浏览器访问:
- API响应:http://127.0.0.1:8000/
- 交互式文档:http://127.0.0.1:8000/docs (Swagger UI)
- 可选文档:http://127.0.0.1:8000/redoc (ReDoc)
📊 核心概念详解
路径操作装饰器
FastAPI使用装饰器来定义路径操作,支持所有HTTP方法
from fastapi import FastAPI
app = FastAPI()
# GET请求
@app.get("/")
async def read_root():
return {"Hello": "World"}
# POST请求
@app.post("/items/")
async def create_item(item: Item):
return item
# PUT请求
@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
return {"item_id": item_id, **item.model_dump()}
# DELETE请求
@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
return {"message": "Item deleted"}
路径参数
路径参数是URL路径的一部分,使用花括号{}声明
from fastapi import FastAPI
app = FastAPI()
# 基本路径参数
@app.get("/users/{user_id}")
async def read_user(user_id: int):
return {"user_id": user_id}
# 枚举类型路径参数
from enum import Enum
class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
lenet = "lenet"
@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):
if model_name == ModelName.alexnet:
return {"model": model_name, "message": "Deep Learning FTW!"}
return {"model": model_name}
查询参数
查询参数是URL中?后面的键值对,使用默认值表示可选csdn.net:
from fastapi import FastAPI
app = FastAPI()
# 必填查询参数
@app.get("/items/")
async def read_item(name: str):
return {"name": name}
# 可选查询参数
from typing import Optional
@app.get("/items/")
async def read_item(name: Optional[str] = None):
return {"name": name}
# 带类型转换的查询参数
@app.get("/items/")
async def read_item(skip: int = 0, limit: int = 10):
return {"skip": skip, "limit": limit}
请求体
请求体通常使用Pydantic模型定义,FastAPI自动验证
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional
app = FastAPI()
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: Optional[float] = None
@app.post("/items/")
async def create_item(item: Item):
item_dict = item.model_dump()
if item.tax:
price_with_tax = item.price + item.tax
item_dict.update({"price_with_tax": price_with_tax})
return item_dict
🛠️ 实战应用:机器学习推理API
以下是一个完整的机器学习模型推理API示例
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np
from sklearn.datasets import load_iris
# 加载模型
model = joblib.load("iris_model.pkl")
iris = load_iris()
# 定义输入数据模型
class IrisInput(BaseModel):
sepal_length: float
sepal_width: float
petal_length: float
petal_width: float
# 定义输出预测模型
class IrisPrediction(BaseModel):
predicted_class: str
confidence: float
app = FastAPI()
@app.post("/predict", response_model=IrisPrediction)
async def predict_species(input_data: IrisInput):
# 转换数据为numpy数组
features = np.array([[input_data.sepal_length,
input_data.sepal_width,
input_data.petal_length,
input_data.petal_width]])
# 进行预测
prediction = model.predict(features)[0]
probabilities = model.predict_proba(features)[0]
# 获取预测结果
predicted_class = iris.target_names[prediction]
confidence = float(probabilities[prediction])
return {
"predicted_class": predicted_class,
"confidence": confidence
}
📝 自动文档生成
FastAPI自动生成交互式API文档,这是它的核心优势之一
Swagger UI (http://127.0.0.1:8000/docs)
- 可视化API文档
- 直接在浏览器中测试API
- 显示所有端点、参数和响应
ReDoc (http://127.0.0.1:8000/redoc)
- 美观的文档界面
- 适合团队协作和分享
OpenAPI规范 (http://127.0.0.1:8000/openapi.json)
- 标准化的API描述文件
- 可用于生成客户端代码
🔧 高级功能
依赖注入系统
FastAPI的依赖注入系统非常强大,可复用逻辑
from fastapi import Depends, FastAPI
app = FastAPI()
# 依赖项函数
def common_parameters(q: str = 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
中间件
中间件可以在请求处理前/后执行自定义逻辑
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
后台任务
异步执行耗时任务,不阻塞响应
from fastapi import BackgroundTasks, FastAPI
app = FastAPI()
def write_log(message: str):
# 模拟耗时写入操作
time.sleep(5)
with open("log.txt", mode="a") as log:
log.write(message + "\n")
@app.post("/send-notification/{email}")
async def send_notification(
email: str,
background_tasks: BackgroundTasks
):
# 添加后台任务
background_tasks.add_task(write_log, f"Notification sent to {email}")
return {"message": "Notification sent in the background"}
📊 FastAPI vs 其他框架
| 特性 | FastAPI | Flask | Django |
|---|---|---|---|
| 性能 | 极高(异步支持)csdn.net | 中等 | 较低 |
| 开发效率 | 高(自动文档)bookstack.cn | 中 | 高 |
| 学习曲线 | 中等 | 低 | 高 |
| 异步支持 | 原生支持csdn.net | 需要扩展 | 需要扩展 |
| 自动文档 | 是bookstack.cn | 否 | 是 |
| 类型安全 | 强(基于类型提示) | 弱 | 中 |
| 适用场景 | 高性能API服务 | 小型Web应用 | 全功能Web应用 |
🚨 生产环境最佳实践
1. 安全配置
from fastapi import FastAPI, HTTPException
from fastapi.security import HTTPBasic, HTTPBasicCredentials
app = FastAPI()
security = HTTPBasic()
def get_current_username(credentials: HTTPBasicCredentials = Depends(security)):
correct_username = "admin"
correct_password = "secret"
if credentials.username != correct_username or credentials.password != correct_password:
raise HTTPException(
status_code=401,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Basic"},
)
return credentials.username
@app.get("/users/me")
def read_users_me(username: str = Depends(get_current_username)):
return {"username": username}
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=["*"],
)
3. 错误处理
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
app = FastAPI()
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
return JSONResponse(
status_code=422,
content={"message": "Validation error", "details": exc.errors()},
)
@app.get("/items/{item_id}")
async def read_item(item_id: int):
if item_id == 0:
raise HTTPException(status_code=404, detail="Item not found")
return {"item_id": item_id}
📈 部署选项
Docker部署
# Dockerfile
FROM python:3.9
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
Gunicorn + Uvicorn
# 安装依赖
pip install gunicorn uvicorn[standard]
# 启动(多worker进程)
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker -b 0.0.0.0:8000
💎 总结
FastAPI的核心优势在于:
- 高性能:基于Starlette和Pydantic,性能媲美NodeJS和
- 开发效率:自动生成交互式文档,减少开发时间
- 类型安全:基于Python类型提示,减少约40%的错误
- 异步支持:原生支持异步编程,适合高并发场景
- 标准化:基于OpenAPI和JSON Schema标准
更多推荐
所有评论(0)