限制請求速率
限制請求速率（rate limiting，也常叫「限流」）是規定「同一個來源在一段時間內最多能打幾次 API」。例如登入端點限制「每個 IP 每分鐘最多 5 次」，可以擋住暴力猜密碼、爬蟲狂刷，也避免單一使用者把伺服器資源吃光。
FastAPI 本身沒有內建限流功能。本頁示範兩種做法：用社群套件 SlowAPI（實務首選），以及自己用依賴（Depends）寫一個，幫助理解原理。
先搞懂三個問題
任何限流方案都在回答這三件事：
| 問題 | 說明 | 常見選擇 |
|---|---|---|
| 用什麼當 key | 「同一個來源」怎麼認 | IP、使用者 ID、API key |
| 規則是什麼 | 多久內幾次 | 5/minute、100/hour |
| 計數存在哪 | 次數記在哪裡 | 記憶體（單機）、Redis（多台／多 worker） |
超過上限時，標準做法是回 HTTP 429 Too Many Requests，並在 Retry-After header 告訴客戶端幾秒後可以再試。
計數演算法
- 固定視窗（fixed window）：每分鐘從 0 開始算。簡單省資源，但在視窗交界處可能短時間內放行兩倍請求（59 秒打 5 次、61 秒再打 5 次）。
- 滑動視窗（moving / sliding window）：永遠看「過去 60 秒」打了幾次，比較精準，代價是要記住每次請求的時間。
- 權杖桶（token bucket）：桶子以固定速度補充權杖，每次請求花掉一個，允許短暫爆量。Nginx、雲端 API Gateway 常用。
入門只要知道：要嚴格就用滑動視窗；要省資源就用固定視窗。
用 SlowAPI：每分鐘最多 5 次
SlowAPI 是把 Flask-Limiter 移植到 Starlette／FastAPI 的套件，底層用 limits 處理計數，是 FastAPI 社群最常見的選擇。
from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address
limiter = Limiter(key_func=get_remote_address) # (1)!
app = FastAPI()
app.state.limiter = limiter # (2)!
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) # (3)!
@app.get("/hello")
@limiter.limit("5/minute") # (4)!
async def hello(request: Request): # (5)!
return {"message": "hi"}
key_func決定「同一個來源」怎麼認。get_remote_address取客戶端 IP。- SlowAPI 從
app.state.limiter找到這個 limiter，名稱不能改。 - 超過上限時 SlowAPI 拋出
RateLimitExceeded，這個 handler 把它轉成 429 回應。 - 規則字串格式是「次數/時間單位」，例如
5/minute、100/hour、1000/day、10/second。也可寫5 per minute。 - 函式一定要有
request: Request參數，SlowAPI 要從它取 IP。少了它，程式啟動時就會報錯：No "request" or "websocket" argument on function。
裝飾器順序也要注意：@app.get(...) 在上，@limiter.limit(...) 在下（緊貼函式）。
同一個 IP 連打 6 次，第 6 次會得到：
同時套多條規則
規則可以疊加，全部都要滿足才放行：
@app.post("/login")
@limiter.limit("5/minute")
@limiter.limit("20/hour")
async def login(request: Request):
...
或寫成一個字串：@limiter.limit("5/minute;20/hour")。
回傳剩餘次數的 header
開啟 headers_enabled=True，回應會帶上 X-RateLimit-Limit、X-RateLimit-Remaining、X-RateLimit-Reset 與 Retry-After，前端可以據此顯示「還剩幾次」。
from fastapi import Response
limiter = Limiter(key_func=get_remote_address, headers_enabled=True)
@app.get("/hello")
@limiter.limit("5/minute")
async def hello(request: Request, response: Response): # 需要 response 參數
return {"message": "hi"}
開啟 header 時要加 response: Response
端點回傳 dict 時，SlowAPI 要透過 response 參數把 header 寫上去。少了它會在執行時報錯。
全站預設上限
不想每個端點都加裝飾器時，設定 default_limits 並掛上中介軟體，所有路由都會套用；個別端點再用 @limiter.limit 覆寫，或用 @limiter.exempt 排除（例如健康檢查）。
from slowapi.middleware import SlowAPIMiddleware
limiter = Limiter(key_func=get_remote_address, default_limits=["100/minute"])
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
app.add_middleware(SlowAPIMiddleware)
@app.get("/items") # 套用預設 100/minute
def list_items():
return []
@app.get("/health")
@limiter.exempt # 不限流
def health():
return {"ok": True}
依使用者或 API key 限流
用 IP 當 key 的問題是：公司、學校、行動網路常常很多人共用一個 IP。已登入的 API 通常改用使用者身分或 API key：
def user_or_ip(request: Request) -> str:
api_key = request.headers.get("X-API-Key")
if api_key:
return f"key:{api_key}"
return f"ip:{get_remote_address(request)}"
limiter = Limiter(key_func=user_or_ip)
key_func 回傳的字串就是計數用的 key。加上 key:、ip: 前綴，避免兩種來源撞名。
改用滑動視窗與 Redis
SlowAPI 預設用固定視窗、計數存在行程記憶體裡。這在正式環境有兩個問題：
- 用
fastapi run --workers 4或多台機器時，每個 worker 各算各的，實際上限會變成 5 × 4 = 20 次。 - 伺服器重啟，計數歸零。
改成共用 Redis 就能解決，並可順便換成滑動視窗：
limiter = Limiter(
key_func=get_remote_address,
storage_uri="redis://localhost:6379", # 需另外 uv add redis
strategy="moving-window", # 預設是 "fixed-window"
)
storage_uri 也支援 memcached://、mongodb:// 等，見 limits 儲存後端說明。
自己寫：用依賴實作每分鐘 5 次
不想加套件、或只是想理解原理，可以用一個可呼叫的類別當依賴（見 依賴注入）。下面用滑動視窗：每個 IP 一條佇列，記錄過去 60 秒內每次請求的時間。
import time
from collections import defaultdict, deque
from fastapi import Depends, FastAPI, HTTPException, Request, status
class RateLimiter:
def __init__(self, times: int, seconds: int):
self.times = times
self.seconds = seconds
self.hits: dict[str, deque[float]] = defaultdict(deque)
def __call__(self, request: Request) -> None:
key = request.client.host if request.client else "unknown"
now = time.monotonic()
window = self.hits[key]
# 丟掉已經超過視窗時間的紀錄
while window and now - window[0] >= self.seconds:
window.popleft()
if len(window) >= self.times:
retry_after = int(self.seconds - (now - window[0])) + 1
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="請求太頻繁，請稍後再試",
headers={"Retry-After": str(retry_after)},
)
window.append(now)
app = FastAPI()
five_per_minute = RateLimiter(times=5, seconds=60)
@app.get("/hello", dependencies=[Depends(five_per_minute)])
def hello():
return {"message": "hi"}
重點：
five_per_minute是同一個實例，所以計數會跨請求保留。若寫成Depends(RateLimiter(5, 60))放在每個端點，各端點會有各自獨立的計數，這通常也是你要的。- 要整組路由都限流，可掛在 router 上：
APIRouter(dependencies=[Depends(five_per_minute)])。 - 用
time.monotonic()而不是time.time()，系統校時不會讓計數亂掉。
這個版本只適合單一行程
計數存在記憶體，多 worker、多台機器或重啟都會失效；hits 也不會清掉沒再出現的 IP，長期執行會慢慢吃記憶體。學原理、內部工具可以用；正式環境請改用 SlowAPI + Redis，或交給閘道層處理。
部署時的注意事項
反向代理後面拿到的 IP 不對
放在 Nginx、Traefik 或雲端負載平衡器後面時，request.client.host 會變成代理的 IP，等於所有人共用一個額度，5 次很快就用完。
解法是讓 Uvicorn 信任代理傳來的 X-Forwarded-For（見 佈署）：
--forwarded-allow-ips 只填你自己的代理 IP。設成 * 而 API 又直接對外時，攻擊者可以偽造 X-Forwarded-For，每次換一個假 IP 就繞過限流。
也可以在閘道層限流
應用程式層限流的好處是能依「使用者」「方案等級」細分規則。但大量惡意流量最好在更前面擋掉，不要讓它進到 Python：
- Nginx：
limit_req模組，例如limit_req_zone $binary_remote_addr zone=api:10m rate=5r/m;。 - 雲端服務：Cloudflare、AWS API Gateway／WAF 等都有限流規則。
常見組合是：閘道層設寬鬆的全站上限擋濫用，FastAPI 層設細緻的業務規則（例如登入 5 次／分、免費方案 1000 次／天）。
測試限流
用 測試 介紹的 TestClient 連打 6 次，確認第 6 次是 429：
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_rate_limit():
for _ in range(5):
assert client.get("/hello").status_code == 200
response = client.get("/hello")
assert response.status_code == 429
計數會跨測試保留。若多個測試打同一端點，可在 fixture 裡呼叫 limiter.reset()（SlowAPI）清空，或在測試中把 app.state.limiter.enabled = False 關掉限流。