FastAPI響應(Response)
- 1、Response入門
- 2、Response基本操作
- 設置響應體(返回數據)
- 設置狀態碼
- 設置響應頭
- 設置 Cookies
- 3、響應模型 response_model
- 4、響應類型 response_class
- Response派生類
- 自定義response_class
在“FastAPI系列05:FastAPI請求(Request)”一節中我們詳細了解了FastAPI程序對請求參數的校驗和處理,在程序中這些處理后的數據將被送入業務邏輯單元,進行進一步的處理,最終向客戶端返回HTTP響應。本節我們通過FastAPI實現大文件斷點續傳等示例詳細討論FastAPI向客戶端返回HTTP響應的內容。
1、Response入門
在HTTP協議中,HTTP響應報文主要由三部分組成:
-
狀態行 (Status Line)
狀態行包含: HTTP版本號(如HTTP/1.1、HTTP/2。)、狀態碼(- 如 200、404、500,代表服務器對請求的處理結果。)、狀態描述( 如 OK、Not Found、Internal Server Error,是簡單的人類可讀的描述) -
響應頭部 (Response Headers)
一系列 鍵值對,告訴客戶端一些關于響應報文本身或者服務器的信息。常見的響應頭字段有:- Content-Type: 指定返回內容的類型,比如 text/html; charset=UTF-8
- Content-Length: 返回內容的長度(單位:字節)
- Server: 服務器軟件的信息
- Set-Cookie: 設置客戶端的 Cookie
- Cache-Control: 緩存策略
- Location: 重定向地址(配合 301/302 狀態碼)
-
響應主體 (Response Body)
主體部分是服務器真正返回給客戶端的數據內容。比如:- HTML 代碼
- 圖片
- JSON 格式的數據
- 二進制文件流
一個常見的HTTP響應報文大致如下:
HTTP/1.1 200 OK
Content-Type: text/html; charset=UTF-8
Content-Length: 137
Connection: keep-alive
Server: nginx/1.18.0<!DOCTYPE html>
<html>
<head>
<title>Hello</title>
</head>
<body>
<h1>Hello, World!</h1>
</body>
</html>
在FastAPI中可以使用Response
或其派生對象方便地設置狀態碼、響應頭以及響應主體。
2、Response基本操作
設置響應體(返回數據)
在FastAPI中,可以通過直接返回 dict / list(這時FastAPI 自動轉成 JSON)、返回 HTML、文件、流式響應或用 Response 或 StreamingResponse 等手動控制來設置響應體。
from fastapi import FastAPIapp = FastAPI()@app.get("/json")
def read_json():return {"message": "Hello, World"}
設置狀態碼
FastAPI中定義了status
枚舉,用于在程序中指定響應狀態碼。
from fastapi import FastAPI, statusapp = FastAPI()@app.post("/create", status_code=status.HTTP_201_CREATED)
def create_item():return {"msg": "Item created"}
設置響應頭
可以用 Response 或 JSONResponse 手動設置頭部,也可以在路由里加 response_model、response_class等參數。
from fastapi import FastAPI
from fastapi.responses import JSONResponseapp = FastAPI()@app.get("/custom-header")
def custom_header():content = {"message": "Custom header"}headers = {"X-Custom-Header": "FastAPI"}return JSONResponse(content=content, headers=headers)
設置 Cookies
使用 Response.set_cookie() 方法
from fastapi import FastAPI, Responseapp = FastAPI()@app.get("/set-cookie")
def set_cookie(response: Response):response.set_cookie(key="session_id", value="abc123")return {"message": "Cookie set"}
3、響應模型 response_model
在FastAPI中,response_model主要用來聲明和驗證返回數據格式。FastAPI 會根據定義的 response_model,完成:
- 自動校驗返回的數據是否符合
- 自動生成 OpenAPI 文檔(自動文檔超好看)
- 自動進行數據的序列化(比如只返回你指定的字段)
from fastapi import FastAPI
from pydantic import BaseModelapp = FastAPI()class UserOut(BaseModel):id: intname: str@app.get("/user", response_model=UserOut)
def get_user():# 返回一個字典,FastAPI會根據UserOut去校驗和生成響應return {"id": 1, "name": "Alice", "password": "123456"}
說明:
- password字段不會返回給前端!因為 UserOut 沒有定義 password 字段。
4、響應類型 response_class
在FastAPI中, response_class主要用來指定響應體的類型和格式。FastAPI 會根據response_class控制返回的內容格式,比如 JSON、HTML、純文本、文件、流式響應等等。
Response派生類
FastAPI提供了JSONResponse、HTMLResponse、PlainTextResponse、FileResponse、StreamingResponse、RedirectResponse等response_class用于定義響應體的類型和格式,它們均派生自 Response 類。
一般情況下,FastAPI默認使用JSONResponse,返回application/json 響應。下面示例中指定了response_class為HTMLResponse以返回 HTML 字符串。
from fastapi import FastAPI
from fastapi.responses import HTMLResponseapp = FastAPI()@app.get("/html", response_class=HTMLResponse)
def get_html():return "<h1>Hello, HTML</h1>"
自定義response_class
我們可以通過以下幾步自定義一個 Response 類:
- 繼承自 fastapi.Response 或 starlette.responses.Response
- 重寫 render() 方法,告訴 FastAPI:怎么把內容轉成字節流(bytes)
- 還可以自定義 media_type(Content-Type)
自定義一個返回 .csv 文件的 Response
from fastapi import FastAPI
from starlette.responses import Responseclass CSVResponse(Response):media_type = "text/csv"def render(self, content: str) -> bytes:# 直接把字符串編碼成bytes返回return content.encode("utf-8")app = FastAPI()@app.get("/csv", response_class=CSVResponse)
def get_csv():csv_content = "id,name\n1,Alice\n2,Bob"return csv_content
自定義一個加密后的 JSON Response
import json
from fastapi import FastAPI
from starlette.responses import Response
import base64class EncryptedJSONResponse(Response):media_type = "application/json"def render(self, content: dict) -> bytes:json_data = json.dumps(content)# 簡單加密:base64編碼(真實項目要用AES/自定義加密)encrypted = base64.b64encode(json_data.encode("utf-8"))return encryptedapp = FastAPI()@app.get("/encrypted", response_class=EncryptedJSONResponse)
def get_encrypted():return {"msg": "secret data"}
自定義一個超大文件分塊下載的 Response
from starlette.responses import StreamingResponseclass LargeFileResponse(StreamingResponse):def __init__(self, file_path: str, chunk_size: int = 1024 * 1024):generator = self.file_chunk_generator(file_path, chunk_size)super().__init__(generator, media_type="application/octet-stream")self.headers["Content-Disposition"] = f"attachment; filename={file_path.split('/')[-1]}"@staticmethoddef file_chunk_generator(file_path: str, chunk_size: int):with open(file_path, mode="rb") as file:while chunk := file.read(chunk_size):yield chunk@app.get("/download")
def download_big_file():return LargeFileResponse("bigfile.zip")
實現斷點續傳(支持 Range )的 StreamingResponse
HTTP協議有個標準:Range 請求頭。客戶端可以在請求頭里加Range: bytes=1000-
,意思是:“我只要從第1000個字節開始的數據,后面的。”
這樣可以實現大文件斷點續傳或音頻、視頻流式播放(在線播放時,只請求一部分)。
from fastapi import FastAPI, Request, HTTPException
from starlette.responses import StreamingResponse
import osapp = FastAPI()# 分塊讀取文件
def range_file_reader(file_path: str, start: int = 0, end: int = None, chunk_size: int = 1024 * 1024):with open(file_path, "rb") as f:f.seek(start)remaining = (end - start + 1) if end else Nonewhile True:read_size = chunk_size if not remaining else min(remaining, chunk_size)data = f.read(read_size)if not data:breakyield dataif remaining:remaining -= len(data)if remaining <= 0:break@app.get("/download")
async def download_file(request: Request):file_path = "bigfile.zip" # 換成你的大文件路徑if not os.path.exists(file_path):raise HTTPException(status_code=404, detail="File not found")file_size = os.path.getsize(file_path)range_header = request.headers.get("range")if range_header:# 解析 range頭bytes_unit, byte_range = range_header.split("=")start_str, end_str = byte_range.split("-")start = int(start_str) if start_str else 0end = int(end_str) if end_str else file_size - 1if start >= file_size:raise HTTPException(status_code=416, detail="Range Not Satisfiable")content_length = end - start + 1headers = {"Content-Range": f"bytes {start}-{end}/{file_size}","Accept-Ranges": "bytes","Content-Length": str(content_length),"Content-Type": "application/octet-stream","Content-Disposition": f"attachment; filename={os.path.basename(file_path)}",}return StreamingResponse(range_file_reader(file_path, start, end),status_code=206, # Partial Contentheaders=headers,)# 沒有Range頭,普通全量返回headers = {"Content-Length": str(file_size),"Content-Type": "application/octet-stream","Content-Disposition": f"attachment; filename={os.path.basename(file_path)}",}return StreamingResponse(range_file_reader(file_path),status_code=200,headers=headers,)
在此基礎上,還可以:
- 支持多段 Range(比如同時請求0-100, 200-300),但是這個場景很少,比較復雜
- 限制最大單次傳輸大小(保護服務器)
- 支持 gzip 壓縮返回(如果是文本文件)
- 加上異步讀取(aiofiles)提升 IO 性能
總之,通過自定義response_class可以實現非常多且實用的功能。