FastAPI 请求与响应(Request & Response)完全指南
一、核心概念
名称 说明 Request 客户端发送给服务器的数据(路径、查询、请求体、头、文件等) Response 服务器返回给客户端的数据(JSON、HTML、文件、状态码、头等) FastAPI 自动处理 你只关心 业务逻辑 ,FastAPI 负责解析与序列化
二、请求(Request)类型
类型 获取方式 说明 路径参数 item_id: int/items/{item_id}查询参数 q: str?q=hello请求体 item: ItemJSON POST 请求头 user_agent: str = Header()User-AgentCookie session: str = Cookie()浏览器 Cookie 文件上传 file: UploadFilemultipart/form-data原始请求 request: Request完整 Starlette 请求对象
三、响应(Response)类型
类型 返回方式 说明 JSON(默认) return {"key": "value"}自动序列化 自定义 JSON JSONResponse控制头、状态码 HTML HTMLResponse返回网页 纯文本 PlainTextResponsereturn "hello"文件 FileResponse下载文件 流式 StreamingResponse大文件、视频流 重定向 RedirectResponse跳转页面 自定义状态码 status_code=201201 Created
四、完整示例:各种请求与响应
main.py
from fastapi import FastAPI, Header, Cookie, Form, UploadFile, File, HTTPException, status, Request
from fastapi.responses import JSONResponse, HTMLResponse, PlainTextResponse, FileResponse, StreamingResponse, RedirectResponse
from pydantic import BaseModel, EmailStr
from typing import Optional, List
import os
import shutil
app = FastAPI(title="请求与响应大全")
# ==================== 1. 路径参数 ====================
@app.get("/items/{item_id}")
def read_item(item_id: int):
return {"item_id": item_id}
# ==================== 2. 查询参数 ====================
@app.get("/search/")
def search(q: str, skip: int = 0, limit: int = 10):
return {"query": q, "skip": skip, "limit": limit}
# ==================== 3. 请求体(Pydantic) ====================
class UserCreate(BaseModel):
username: str
email: EmailStr
age: int
bio: Optional[str] = None
@app.post("/users/", response_model=UserCreate, status_code=status.HTTP_201_CREATED)
def create_user(user: UserCreate):
return user # 自动转 JSON
# ==================== 4. 请求头 ====================
@app.get("/headers/")
def get_headers(user_agent: Optional[str] = Header(None), x_token: str = Header(...)):
return {"User-Agent": user_agent, "X-Token": x_token}
# ==================== 5. Cookie ====================
@app.get("/cookie/")
def read_cookie(session_id: Optional[str] = Cookie(None)):
return {"session_id": session_id}
# ==================== 6. 表单数据 ====================
@app.post("/login/")
def login(username: str = Form(...), password: str = Form(...)):
return {"username": username, "logged_in": True}
# ==================== 7. 文件上传 ====================
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
contents = await file.read()
return {"filename": file.filename, "size": len(contents)}
# 多文件上传
@app.post("/upload-multiple/")
async def upload_files(files: List[UploadFile] = File(...)):
return {"filenames": [f.filename for f in files]}
# ==================== 8. 原始 Request 对象 ====================
@app.post("/raw/")
async def raw_request(request: Request):
body = await request.body()
return {"method": request.method, "url": str(request.url), "body": body.decode()}
# ==================== 9. 各种响应类型 ====================
# 9.1 自定义 JSON 响应
@app.get("/custom-json/")
def custom_json():
return JSONResponse(
content={"message": "自定义 JSON"},
status_code=200,
headers={"X-Custom": "value"}
)
# 9.2 HTML 响应
@app.get("/html/", response_class=HTMLResponse)
def get_html():
return """
<html>
<head><title>FastAPI</title></head>
<body><h1>Hello from FastAPI!</h1></body>
</html>
"""
# 9.3 纯文本
@app.get("/text/", response_class=PlainTextResponse)
def get_text():
return "This is plain text"
# 9.4 文件下载
@app.get("/download/")
def download_file():
file_path = "example.txt"
with open(file_path, "w") as f:
f.write("这是一个下载文件")
return FileResponse(file_path, filename="下载.txt")
# 9.5 流式响应(大文件)
def generate_large_data():
for i in range(1000):
yield f"Line {i}\n"
@app.get("/stream/")
def stream_data():
return StreamingResponse(generate_large_data(), media_type="text/plain")
# 9.6 重定向
@app.get("/redirect/")
def redirect():
return RedirectResponse("/docs")
# ==================== 10. 响应模型(过滤字段) ====================
class UserInDB(BaseModel):
id: int
username: str
email: str
password_hash: str # 敏感字段
class UserOut(BaseModel):
id: int
username: str
email: str
@app.get("/users/{user_id}", response_model=UserOut)
def get_user(user_id: int):
# 模拟数据库
return UserInDB(
id=user_id,
username="john",
email="john@example.com",
password_hash="secret123"
)
# password_hash 不会返回!
五、运行与测试
uvicorn main:app --reload
测试方式:
功能 访问地址 文档 http://127.0.0.1:8000/docs 路径参数 GET /items/42查询参数 GET /search/?q=book创建用户 POST /users/ + JSON文件上传 POST /upload/ + 文件下载文件 GET /download/
六、高级技巧
1. 自定义响应头
from fastapi import Response
@app.get("/with-header/")
def with_header(response: Response):
response.headers["X-App-Version"] = "1.0.0"
return {"status": "ok"}
2. 压缩响应(Gzip)
from fastapi import FastAPI
from starlette.middleware.gzip import GZipMiddleware
app = FastAPI()
app.add_middleware(GZipMiddleware, minimum_size=1000)
3. 缓存控制
from fastapi.responses import Response
@app.get("/cached/")
def cached_data(response: Response):
response.headers["Cache-Control"] = "max-age=3600"
return {"data": "cached for 1 hour"}
七、常见响应状态码
from fastapi import status
# 201 Created
status_code=status.HTTP_201_CREATED
# 204 No Content
status_code=204
# 400 Bad Request
raise HTTPException(status_code=400, detail="错误请求")
# 422 Validation Error(自动)
八、请求体 + 查询 + 路径混合
class ItemUpdate(BaseModel):
name: Optional[str] = None
price: Optional[float] = None
@app.patch("/items/{item_id}")
def partial_update(
item_id: int,
item: ItemUpdate,
q: Optional[str] = None
):
return {"item_id": item_id, "update": item.dict(exclude_unset=True), "q": q}
九、文件上传完整示例(保存到磁盘)
@app.post("/upload-save/")
async def upload_and_save(file: UploadFile = File(...)):
save_path = f"uploads/{file.filename}"
os.makedirs("uploads", exist_ok=True)
with open(save_path, "wb") as f:
shutil.copyfileobj(file.file, f)
return {"saved": save_path}
十、响应模型排除敏感字段
class UserPrivate(BaseModel):
password: str
class UserPublic(BaseModel):
username: str
email: str
@app.get("/me", response_model=UserPublic, response_model_exclude_unset=True)
def read_current_user():
return {
"username": "alice",
"email": "alice@example.com",
"password": "secret" # 不会返回
}
总结:请求与响应速查表
类型 写法 路径参数 item_id: int查询参数 q: str = None请求体 user: UserCreate请求头 token: str = Header()文件 file: UploadFileJSON 响应 return {...}HTML 响应 response_class=HTMLResponse文件下载 FileResponse(path)状态码 status_code=201响应模型 response_model=UserOut
下一步学习
主题 关键词 表单 + 文件混合 form file大文件分片上传 chunkWebSocket websocket背景任务 background自定义中间件 middleware
现在就打开 http://127.0.0.1:8000/docs,体验各种请求与响应吧!
需要我生成一个 完整请求响应项目模板 (包含登录、上传、下载、流式)吗? 回复 请求模板 立刻获取!