FastAPI 请求和响应

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-Agent
Cookiesession: str = Cookie()浏览器 Cookie
文件上传file: UploadFilemultipart/form-data
原始请求request: Request完整 Starlette 请求对象

三、响应(Response)类型

类型返回方式说明
JSON(默认)return {"key": "value"}自动序列化
自定义 JSONJSONResponse控制头、状态码
HTMLHTMLResponse返回网页
纯文本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: UploadFile
JSON 响应return {...}
HTML 响应response_class=HTMLResponse
文件下载FileResponse(path)
状态码status_code=201
响应模型response_model=UserOut

下一步学习

主题关键词
表单 + 文件混合form file
大文件分片上传chunk
WebSocketwebsocket
背景任务background
自定义中间件middleware

现在就打开 http://127.0.0.1:8000/docs,体验各种请求与响应吧!

需要我生成一个 完整请求响应项目模板(包含登录、上传、下载、流式)吗?
回复 请求模板 立刻获取!

类似文章

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注