FastAPI 基本路由
FastAPI 基本路由(Routing)完全指南
一、什么是路由?
路由(Route) = URL 路径 + HTTP 方法 → 函数处理
FastAPI 使用 装饰器 定义路由,简洁直观。
@app.get("/path") → GET 请求
@app.post("/path") → POST 请求
@app.put("/path") → PUT 请求
@app.delete("/path") → DELETE 请求
二、核心 HTTP 方法装饰器
| 装饰器 | 方法 | 常见用途 |
|---|---|---|
@app.get() | GET | 查询数据 |
@app.post() | POST | 创建数据 |
@app.put() | PUT | 整体更新 |
@app.patch() | PATCH | 部分更新 |
@app.delete() | DELETE | 删除数据 |
@app.options() | OPTIONS | 预检请求 |
@app.head() | HEAD | 仅返回头信息 |
三、基本路由示例(main.py)
from fastapi import FastAPI
app = FastAPI()
# 1. GET - 根路径
@app.get("/")
def read_root():
return {"message": "欢迎使用 FastAPI!"}
# 2. GET - 带路径参数
@app.get("/items/{item_id}")
def read_item(item_id: int):
return {"item_id": item_id}
# 3. POST - 接收 JSON 请求体
@app.post("/items/")
def create_item(name: str, price: float):
return {"name": name, "price": price}
# 4. PUT - 更新数据
@app.put("/items/{item_id}")
def update_item(item_id: int, name: str, price: float):
return {"item_id": item_id, "name": name, "price": price}
# 5. DELETE - 删除
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
return {"message": f"Item {item_id} 已删除"}
四、路径参数(Path Parameters)
1. 基本类型转换
@app.get("/users/{user_id}")
def get_user(user_id: int):
return {"user_id": user_id}
支持类型:int, float, bool, str, UUID, Path
2. 多个路径参数
@app.get("/files/{file_path:path}")
def read_file(file_path: str):
return {"file_path": file_path}
:path表示匹配/后的所有内容(如/home/user/docs/file.txt)
3. 枚举限制
from enum import Enum
class ModelName(str, Enum):
alexnet = "alexnet"
resnet = "resnet"
lenet = "lenet"
@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
if model_name is ModelName.alexnet:
return {"model_name": model_name, "message": "Deep Learning!"}
return {"model_name": model_name}
访问 /models/resnet → 有效
访问 /models/invalid → 422 错误
五、查询参数(Query Parameters)
@app.get("/items/")
def read_items(
skip: int = 0, # 默认值
limit: int = 10,
q: str | None = None, # 可选参数
active: bool = True # 布尔值自动转换
):
return {"skip": skip, "limit": limit, "q": q, "active": active}
访问:
http://localhost:8000/items/?skip=5&limit=10&q=book&active=false
必填查询参数(无默认值)
@app.get("/search/")
def search(q: str): # 必须提供
return {"query": q}
六、请求体(Request Body)+ Pydantic 模型
from pydantic import BaseModel
from typing import Optional
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
tax: float | None = None
@app.post("/items/")
def create_item(item: Item):
item_dict = item.dict()
if item.tax:
price_with_tax = item.price + item.tax
item_dict["price_with_tax"] = price_with_tax
return item_dict
请求示例:
{
"name": "Apple",
"price": 1.5,
"tax": 0.1
}
七、混合参数:路径 + 查询 + 请求体
@app.put("/items/{item_id}")
def update_item(
item_id: int,
item: Item, # 请求体
q: str | None = None # 查询参数
):
result = {"item_id": item_id, **item.dict()}
if q:
result["q"] = q
return result
八、响应模型(Response Model)
控制返回字段,自动过滤多余数据:
class UserOut(BaseModel):
id: int
username: str
email: str
@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int):
# 数据库返回更多字段,但只返回指定的
return {"id": user_id, "username": "john", "email": "john@example.com", "password": "secret"}
→ 响应中 不会包含 password
九、状态码(Status Code)
from fastapi import status
@app.post("/items/", status_code=status.HTTP_201_CREATED)
def create_item(item: Item):
return item
@app.delete("/items/{item_id}", status_code=204)
def delete_item(item_id: int):
return None # 204 无响应体
常用状态码:
| 代码 | 含义 |
|---|---|
| 200 | OK |
| 201 | Created |
| 204 | No Content |
| 400 | Bad Request |
| 404 | Not Found |
| 422 | Validation Error |
十、路由分组:APIRouter(推荐!)
避免 main.py 过长,按功能模块化。
routers/users.py
from fastapi import APIRouter
router = APIRouter(prefix="/users", tags=["用户"])
@router.get("/")
def get_users():
return [{"id": 1, "name": "Alice"}]
@router.post("/")
def create_user():
return {"message": "用户创建成功"}
main.py
from fastapi import FastAPI
from routers import users
app = FastAPI()
app.include_router(users.router)
# 访问:GET /users/ , POST /users/
十一、完整示例:用户管理路由
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
class User(BaseModel):
id: int
name: str
email: str
age: Optional[int] = None
fake_db: List[User] = []
@app.get("/users/", response_model=List[User])
def list_users(skip: int = 0, limit: int = 10):
return fake_db[skip:skip + limit]
@app.post("/users/", response_model=User, status_code=status.HTTP_201_CREATED)
def create_user(user: User):
if any(u.id == user.id for u in fake_db):
raise HTTPException(400, "用户 ID 已存在")
fake_db.append(user)
return user
@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int):
for user in fake_db:
if user.id == user_id:
return user
raise HTTPException(404, "用户不存在")
@app.put("/users/{user_id}", response_model=User)
def update_user(user_id: int, user_update: User):
for i, user in enumerate(fake_db):
if user.id == user_id:
fake_db[i] = user_update
return user_update
raise HTTPException(404, "用户不存在")
@app.delete("/users/{user_id}", status_code=204)
def delete_user(user_id: int):
for i, user in enumerate(fake_db):
if user.id == user_id:
fake_db.pop(i)
return
raise HTTPException(404, "用户不存在")
十二、运行与测试
uvicorn main:app --reload
访问:
- 文档:http://127.0.0.1:8000/docs
- 所有路由自动显示
总结:路由核心要点
| 特性 | 写法 |
|---|---|
| 路径参数 | /items/{item_id} + item_id: int |
| 查询参数 | def func(q: str = None) |
| 请求体 | item: Item |
| 响应模型 | response_model=UserOut |
| 状态码 | status_code=201 |
| 分组路由 | APIRouter(prefix="/api") |
下一步学习
| 主题 | 关键词 |
|---|---|
| 路由分组与模块化 | router |
| 依赖注入 | Depends |
| 认证路由 | auth |
| 文件上传路由 | upload |
现在就运行你的路由,打开 /docs 体验自动文档吧!
需要我生成一个 完整模块化路由项目模板(users/items/admin 分离)吗?
回复 路由模板 立刻获取!