如何用 FastAPI 建立支援多租戶的產品與訂單 API
引言：為什麼多租戶設計在 SaaS 中如此重要？
在現代軟體服務 (SaaS) 的開發中，多租戶 (Multi-Tenancy) 架構已成為主流。它允許單一應用程式實例服務多個客戶 (租戶)，每個客戶的數據彼此隔離且安全。這種模式能有效降低營運成本，簡化部署，並提高資源利用率。
然而，實作多租戶並確保數據的嚴格隔離，是許多工程師面臨的挑戰。特別是在 API 設計中，如何安全地識別租戶身份並自動過濾數據，是至關重要的環節。
本教學文章將帶領熟悉 Python 的工程師，運用高效能的 FastAPI 框架，結合 JWT (JSON Web Token) 認證機制和 SQLAlchemy ORM，一步步建立一個支援多租戶的產品與訂單 API。我們將重點強調 JWT 多租戶設計的細節，以及如何在程式碼層面強制執行資料隔離。
您可以透過以下 GitHub 連結檢閱本專案的原始碼：https://github.com/BpsEason/agribridge-saas.git
在開始實作之前，讓我們快速回顧幾個核心概念：
FastAPI：一個現代、快速 (高效能) 的 Web 框架，用於建構基於標準 Python 型別提示的 API。它自動生成 OpenAPI (Swagger) 和 ReDoc 文檔，大大提升開發效率。
JWT (JSON Web Token)：一種輕量級、自包含的訊息格式，用於在各方之間安全地傳輸資訊。JWT 通過數位簽章來驗證資訊的完整性，常被用於身份認證和授權。透過在 JWT 的 Payload 中嵌入
tenant_id，我們可以在每次請求中安全地傳遞租戶身份。多租戶 (Multi-Tenancy)：
資料層面：本教學將採用最常見且成本效益最高的共享資料庫/共享 Schema (Shared Database/Shared Schema) 策略。這意味著所有租戶的數據都存在於同一個資料庫和相同的表格中，但通過一個
discriminator欄位（即tenant_id）來區分數據歸屬。應用層面：關鍵在於應用程式邏輯必須確保所有數據操作（讀、寫、更新、刪除）都自動帶上當前租戶的 ID，防止數據洩漏或混淆。
實作步驟逐條展開
我們將以循序漸進的方式，建立一個支援多租戶的 FastAPI 應用程式。
步驟 1：專案初始化與依賴安裝
首先，建立專案目錄並設定虛擬環境，然後安裝必要的 Python 函式庫。
# 建立專案目錄
mkdir fastapi-multitenant-api
cd fastapi-multitenant-api
# 創建並激活虛擬環境 (強烈建議，保持環境整潔)
python -m venv venv
source venv/bin/activate # 適用於 macOS/Linux
# .\venv\Scripts\activate # 適用於 Windows PowerShell
# 安裝 FastAPI 及其所有依賴 (包括 uvicorn)、SQLAlchemy、PyJWT 和 PyMySQL
pip install "fastapi[all]" uvicorn SQLAlchemy PyJWT python-multipart pymysql
步驟 2：資料庫模型設計 (SQLAlchemy ORM)
這是多租戶設計的基礎。我們在每個需要隔離的資料表（products 和 orders）中加入一個 tenant_id 欄位。這個欄位將作為外鍵，連結到 tenants 表，明確指出每條數據屬於哪個租戶。
建立 models.py 檔案：
# models.py
from sqlalchemy import Column, Integer, String, Float, ForeignKey, text
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
# 宣告 ORM 模型的基類
Base = declarative_base()
class Tenant(Base):
"""
租戶（組織）模型。
這是所有多租戶數據的根，每個租戶都有唯一的 ID。
"""
__tablename__ = "tenants"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(255), unique=True, index=True, nullable=False)
# 定義與產品和訂單的一對多關係，並設定級聯刪除行為
products = relationship("Product", back_populates="tenant", cascade="all, delete-orphan")
orders = relationship("Order", back_populates="tenant", cascade="all, delete-orphan")
class User(Base):
"""
使用者模型，每個使用者都屬於一個特定的租戶。
hashed_password 欄位應儲存密碼的雜湊值，而非明文。
"""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(255), unique=True, index=True, nullable=False)
hashed_password = Column(String(255), nullable=False) # 在實際應用中應使用 bcrypt 等雜湊演算法
# 關鍵：將使用者與租戶連結起來，這決定了使用者所屬的數據範圍
tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False)
class Product(Base):
"""
產品模型，**包含 tenant_id 欄位以實現數據隔離**。
所有對產品的查詢和修改都必須基於此 tenant_id。
"""
__tablename__ = "products"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(255), index=True, nullable=False)
price = Column(Float, nullable=False)
# 關鍵：多租戶資料隔離欄位，作為外鍵並建立索引以優化查詢
tenant_id = Column(Integer, ForeignKey("tenants.id"), index=True, nullable=False)
# 定義與租戶的關係
tenant = relationship("Tenant", back_populates="products")
orders = relationship("Order", back_populates="product") # 與訂單的一對多關係
class Order(Base):
"""
訂單模型，**同樣包含 tenant_id 欄位**。
每個訂單都屬於特定租戶的特定產品。
"""
__tablename__ = "orders"
id = Column(Integer, primary_key=True, index=True)
quantity = Column(Integer, nullable=False)
total_price = Column(Float, nullable=False)
# 關鍵：多租戶資料隔離欄位，作為外鍵並建立索引以優化查詢
tenant_id = Column(Integer, ForeignKey("tenants.id"), index=True, nullable=False)
product_id = Column(Integer, ForeignKey("products.id"), nullable=False)
# 定義與租戶和產品的關係
tenant = relationship("Tenant", back_populates="orders")
product = relationship("Product", back_populates="orders")
設計重點：
每個需要隔離的資料表 (
Product,Order) 都必須包含tenant_id欄位作為外鍵。在 SQLAlchemy 模型中，我們將
tenant_id設為index=True，這對於基於tenant_id進行的查詢性能至關重要。
User模型也包含tenant_id，將使用者與其所屬的租戶綁定。
步驟 3：資料庫連接設定
我們需要一個檔案來定義資料庫的連接引擎和會話工廠。
建立 database.py 檔案：
# database.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
# 設定 MySQL 資料庫連接字串
# **請根據您的實際資料庫配置修改此處**
# 範例：mysql+pymysql://使用者名稱:密碼@主機名稱:埠號/資料庫名稱
SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:your_mysql_root_password@localhost:3306/multitenant_db"
# 創建 SQLAlchemy 引擎
engine = create_engine(SQLALCHEMY_DATABASE_URL)
# 創建一個 SessionLocal 類別，用於創建資料庫會話
# autocommit=False: 不會自動提交事務
# autoflush=False: 不會自動將物件同步到資料庫
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
"""
FastAPI 依賴注入：獲取資料庫會話。
此函數將在每個需要資料庫連線的 API 請求中被調用。
它確保會話在使用後被正確關閉。
"""
db = SessionLocal()
try:
yield db # 使用 yield 讓 FastAPI 管理會話的生命週期
finally:
db.close() # 確保資料庫會話被關閉
設計重點：
get_db()函數作為 FastAPI 的依賴注入，為每個請求提供一個獨立的資料庫會話，並確保會話在請求結束後被正確關閉。
步驟 4：JWT 認證與多租戶依賴注入
這一層是實現多租戶安全的核心。我們將建立一個 FastAPI Depends 依賴，它負責：
從 HTTP 請求的
Authorization頭中提取 JWT。驗證 JWT 的有效性。
從 JWT 的 Payload 中提取當前使用者的
username和最重要的tenant_id。將經認證的使用者物件（包含其
tenant_id）注入到需要保護的 API 路徑中。
建立 auth.py 檔案：
# auth.py
from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import Depends, HTTPException, status, Header
from jose import jwt, JWTError
from pydantic import BaseModel
from sqlalchemy.orm import Session
from . import models, database # 導入 models 和 database 模組
# JWT 相關配置
# **請務必在生產環境中將此密鑰替換為強密鑰！**
# 建議使用環境變數或專用的配置管理工具來管理。
SECRET_KEY = "your-very-secure-secret-key-that-should-be-random-and-long"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
class TokenData(BaseModel):
"""用於儲存從 JWT Payload 中解析出的數據。"""
username: str | None = None
tenant_id: int | None = None
def create_access_token(data: dict):
"""
建立一個 JWT Token，並將額外的數據（例如 tenant_id）嵌入其中。
"""
to_encode = data.copy()
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire}) # 將過期時間加入 payload
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
def verify_jwt_token(token: str) -> TokenData:
"""
驗證 JWT Token 的有效性，並嘗試解析其 Payload。
如果 Token 無效或過期，將拋出 HTTPException。
"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="無法驗證憑證",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
tenant_id: int = payload.get("tenant_id") # 從 Token 中提取 tenant_id
if username is None or tenant_id is None:
raise credentials_exception
token_data = TokenData(username=username, tenant_id=tenant_id)
except JWTError:
raise credentials_exception # 如果 JWT 解析失敗或簽名無效
except Exception as e:
# 捕獲其他可能的錯誤，例如過期
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail=f"Token 錯誤: {e}",
headers={"WWW-Authenticate": "Bearer"},
)
return token_data
def get_current_user(
token_header: Annotated[str, Header(alias="Authorization")],
db: Session = Depends(database.get_db)
) -> models.User:
"""
FastAPI 依賴注入：驗證 Token，並從資料庫載入使用者對象。
這個依賴將在所有受保護的 API 路徑中使用。
"""
# 從 "Bearer <token>" 格式中提取實際的 Token 字串
if not token_header or not token_header.startswith("Bearer "):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="無效的認證頭，請使用 'Bearer <token>' 格式",
headers={"WWW-Authenticate": "Bearer"},
)
token = token_header.replace("Bearer ", "")
# 驗證 JWT Token 並獲取其內含的數據
token_data = verify_jwt_token(token)
# 從資料庫中查找對應的使用者
user = db.query(models.User).filter(models.User.username == token_data.username).first()
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="使用者未找到")
# **關鍵**：驗證 Token 中的 tenant_id 與資料庫中的使用者所屬的 tenant_id 是否一致
# 這一步驟是為了防範某些極端情況下的 Token 篡改或數據不一致
if user.tenant_id != token_data.tenant_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Token 中的租戶 ID 與使用者綁定的租戶 ID 不符，請重新登入。"
)
return user
設計重點：
create_access_token函式將tenant_id作為 JWT Payload 的一部分，確保每個 Token 都帶有租戶身份資訊。
get_current_user依賴是核心。它不僅驗證 Token，還從 Token 中提取tenant_id，並將完整的User物件注入到 API 路由的參數中。這確保了後續的數據操作能夠利用這個tenant_id。額外加入了 Token 中的
tenant_id與資料庫中使用者tenant_id的比對，增加安全性。
步驟 5：Pydantic Schemas
Pydantic 模型用於定義 API 請求和響應的數據結構，確保數據的類型安全和有效性。
建立 schemas.py 檔案：
# schemas.py
from pydantic import BaseModel
from typing import Optional # 導入 Optional 類型
class UserLogin(BaseModel):
"""用於使用者登入請求的 Pydantic 模型。"""
username: str
password: str
class ProductBase(BaseModel):
"""產品的基礎模型，包含共同屬性。"""
name: str
price: float
class ProductCreate(ProductBase):
"""用於創建新產品的請求模型。"""
pass
class Product(ProductBase):
"""
用於產品響應的 Pydantic 模型，包含資料庫生成的 ID 和 tenant_id。
Config.from_attributes = True 用於兼容 SQLAlchemy ORM 對象。
"""
id: int
tenant_id: int
class Config:
from_attributes = True
class OrderBase(BaseModel):
"""訂單的基礎模型。"""
product_id: int
quantity: int
class OrderCreate(OrderBase):
"""用於創建新訂單的請求模型。"""
pass
class Order(OrderBase):
"""
用於訂單響應的 Pydantic 模型，包含資料庫生成的 ID 和 tenant_id。
"""
id: int
total_price: float # 訂單總價通常由後端根據產品價格和數量計算
tenant_id: int
class Config:
from_attributes = True
步驟 6：實作多租戶 API 路由
現在，我們將建立實際的 API 路徑。在每個受保護的 API 邏輯中，我們都會利用 get_current_user 依賴注入，並使用其注入的 user 物件來自動過濾數據或強制設定數據的 tenant_id。
建立 routes/ 目錄並在其中建立 auth_routes.py, product_routes.py, order_routes.py。
routes/auth_routes.py (登入路由)：
# routes/auth_routes.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from .. import models, database, schemas # 導入 models, database, schemas
from ..auth import create_access_token # 導入 JWT token 創建函數
router = APIRouter(tags=["Auth"])
@router.post("/login")
def login(user_credentials: schemas.UserLogin, db: Session = Depends(database.get_db)):
"""
使用者登入端點。
驗證使用者憑證，並在成功後返回包含租戶 ID 的 JWT 存取令牌。
"""
# 在實際應用中，這裡應該有更安全的密碼驗證邏輯，例如 bcrypt
user = db.query(models.User).filter(
models.User.username == user_credentials.username,
models.User.hashed_password == user_credentials.password # **注意：實際中應該是密碼雜湊比對**
).first()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="使用者名稱或密碼不正確",
headers={"WWW-Authenticate": "Bearer"},
)
# **關鍵：在 JWT Token 的 Payload 中包含使用者的 tenant_id**
# 這使得我們可以在後續的請求中直接從 Token 獲取租戶身份，而無需再次查詢資料庫。
access_token = create_access_token(data={"sub": user.username, "tenant_id": user.tenant_id})
return {"access_token": access_token, "token_type": "bearer"}
routes/product_routes.py (產品路由)：
# routes/product_routes.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from .. import models, database, schemas # 導入 models, database, schemas
from ..auth import get_current_user # 導入多租戶認證依賴
router = APIRouter(
prefix="/products",
tags=["Products"],
# **將 get_current_user 依賴應用於此路由器下的所有路徑**
# 這意味著所有 /products/* 的請求都將被 JWT 認證，並自動注入 current_user 物件。
dependencies=[Depends(get_current_user)]
)
@router.get("/", response_model=List[schemas.Product])
def get_products(
current_user: models.User = Depends(get_current_user), # 獲取當前使用者對象，其中包含 tenant_id
db: Session = Depends(database.get_db)
):
"""
獲取當前租戶的所有產品。
**重要的資料隔離點：只會返回屬於該租戶的產品。**
"""
products = db.query(models.Product).filter(
models.Product.tenant_id == current_user.tenant_id # **使用 tenant_id 進行過濾**
).all()
return products
@router.post("/", response_model=schemas.Product, status_code=status.HTTP_201_CREATED)
def create_product(
product: schemas.ProductCreate,
current_user: models.User = Depends(get_current_user), # 獲取當前使用者對象
db: Session = Depends(database.get_db)
):
"""
為當前租戶創建一個新產品。
**重要的資料隔離點：新產品的 tenant_id 會被強制設為當前使用者的 tenant_id。**
"""
# 創建新的產品實例，並將其 tenant_id 設為當前使用者的 tenant_id
new_product = models.Product(
**product.dict(),
tenant_id=current_user.tenant_id # **強制設定 tenant_id，防止使用者惡意指定其他租戶的 ID**
)
db.add(new_product)
db.commit() # 提交事務以將新產品寫入資料庫
db.refresh(new_product) # 刷新物件以獲取資料庫生成的 ID
return new_product
# 範例：根據 ID 獲取產品 (也需過濾 tenant_id)
@router.get("/{product_id}", response_model=schemas.Product)
def get_product_by_id(
product_id: int,
current_user: models.User = Depends(get_current_user),
db: Session = Depends(database.get_db)
):
"""
根據 ID 獲取特定產品。
**只會返回屬於當前租戶且 ID 匹配的產品。**
"""
product = db.query(models.Product).filter(
models.Product.id == product_id,
models.Product.tenant_id == current_user.tenant_id # **再次過濾 tenant_id**
).first()
if not product:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="產品未找到或不屬於您的租戶")
return product
routes/order_routes.py (訂單路由)：
# routes/order_routes.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List
from .. import models, database, schemas # 導入 models, database, schemas
from ..auth import get_current_user # 導入多租戶認證依賴
router = APIRouter(
prefix="/orders",
tags=["Orders"],
# **將 get_current_user 依賴應用於此路由器下的所有路徑**
dependencies=[Depends(get_current_user)]
)
@router.get("/", response_model=List[schemas.Order])
def get_orders(
current_user: models.User = Depends(get_current_user), # 獲取當前使用者對象
db: Session = Depends(database.get_db)
):
"""
獲取當前租戶的所有訂單。
**只會返回屬於該租戶的訂單。**
"""
orders = db.query(models.Order).filter(
models.Order.tenant_id == current_user.tenant_id # **使用 tenant_id 進行過濾**
).all()
return orders
@router.post("/", response_model=schemas.Order, status_code=status.HTTP_201_CREATED)
def create_order(
order: schemas.OrderCreate,
current_user: models.User = Depends(get_current_user), # 獲取當前使用者對象
db: Session = Depends(database.get_db)
):
"""
為當前租戶創建一個新訂單。
**確保訂單中的產品也屬於當前租戶，並強制設定訂單的 tenant_id。**
"""
# 首先，驗證訂單中的產品是否存在，並且該產品必須屬於當前使用者的租戶
product = db.query(models.Product).filter(
models.Product.id == order.product_id,
models.Product.tenant_id == current_user.tenant_id # **確保產品與租戶一致**
).first()
if not product:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="產品未找到或不屬於您的租戶")
# 計算訂單總價
total_price = product.price * order.quantity
# 創建新的訂單實例，並將其 tenant_id 設為當前使用者的 tenant_id
new_order = models.Order(
product_id=order.product_id,
quantity=order.quantity,
total_price=total_price,
tenant_id=current_user.tenant_id # **強制設定 tenant_id**
)
db.add(new_order)
db.commit() # 提交事務
db.refresh(new_order) # 刷新物件以獲取資料庫生成的 ID 和其他預設值
return new_order
程式碼重點：
每個路由的
APIRouter都使用dependencies=[Depends(get_current_user)]，這是一個非常方便的方式，可以將 JWT 認證應用於路由器下的所有路徑。在
GET請求中，我們使用filter(models.Table.tenant_id == current_user.tenant_id)來嚴格過濾資料庫查詢結果。這是實現資料隔離的關鍵一步。在
POST請求中，我們強制將新建立的物件的tenant_id設為當前使用者的tenant_id，而不是依賴前端傳入的值。這可以防止惡意的跨租戶數據注入或數據歸屬錯誤。創建訂單時，會先驗證所選產品是否存在且屬於當前租戶，進一步加強數據隔離和業務邏輯的正確性。
步驟 7：整合 FastAPI 主應用
最後，將所有組件整合到 FastAPI 的主應用程式中。我們還將添加一個啟動事件，用於在應用程式啟動時自動建立資料庫表和一些測試數據。
建立 main.py 檔案：
# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy_utils import database_exists, create_database # 用於檢查和創建資料庫
from . import models, database # 導入資料庫模型和連接設定
from .routes import auth_routes, product_routes, order_routes # 導入定義好的路由
app = FastAPI(
title="多租戶產品與訂單 API",
description="一個支援多租戶、基於 JWT 認證的產品與訂單管理 API。強調資料隔離。",
version="1.0.0"
)
# 包含所有子路由
app.include_router(auth_routes.router, prefix="/api/v1")
app.include_router(product_routes.router, prefix="/api/v1")
app.include_router(order_routes.router, prefix="/api/v1")
@app.on_event("startup")
def create_database_and_seed_data():
"""
應用程式啟動事件：
1. 檢查資料庫是否存在，如果不存在則創建。
2. 建立所有在 models.py 中定義的資料庫表格。
3. 如果資料庫中沒有租戶和使用者，則插入一些初始的測試數據。
"""
# 檢查資料庫是否存在，如果不存在則創建
if not database_exists(database.engine.url):
print(f"資料庫 '{database.engine.url.database}' 不存在，正在創建...")
create_database(database.engine.url)
print("資料庫創建成功。")
else:
print(f"資料庫 '{database.engine.url.database}' 已存在。")
# 建立所有在 models.py 中定義的表格
print("正在建立資料庫表格...")
models.Base.metadata.create_all(bind=database.engine)
print("資料庫表格建立完成。")
# 插入初始測試數據（如果資料庫為空）
db_session = database.SessionLocal()
try:
# 檢查是否有任何租戶存在
if not db_session.query(models.Tenant).count():
print("正在插入初始租戶和使用者數據...")
# 創建兩個租戶
tenant1 = models.Tenant(name="租戶 A")
tenant2 = models.Tenant(name="租戶 B")
db_session.add_all([tenant1, tenant2])
db_session.commit() # 提交租戶以獲取 ID
db_session.refresh(tenant1)
db_session.refresh(tenant2)
# 建立使用者並與租戶關聯
# **注意：這裡的密碼是明文，實際應用中請使用雜湊化密碼**
user1 = models.User(username="user_a", hashed_password="password_a", tenant_id=tenant1.id)
user2 = models.User(username="user_b", hashed_password="password_b", tenant_id=tenant2.id)
db_session.add_all([user1, user2])
db_session.commit()
print("初始租戶和使用者數據插入完成。")
else:
print("資料庫中已存在租戶，跳過初始數據插入。")
except Exception as e:
print(f"初始化資料庫數據時發生錯誤: {e}")
# 在生產環境中，這裡應該記錄日誌而不是直接列印
finally:
db_session.close() # 確保資料庫會話被關閉
# 應用程式的根路徑
@app.get("/")
def read_root():
"""根路徑，返回歡迎訊息。"""
return {"message": "歡迎來到多租戶 API 範例！"}
# 健康檢查端點
@app.get("/health")
def health_check(db: Session = Depends(database.get_db)):
"""
健康檢查端點，驗證資料庫連接是否正常。
"""
try:
db.execute(text("SELECT 1")) # 簡單的資料庫查詢以驗證連接
return {"status": "ok", "database_connection": True}
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"資料庫連接失敗: {e}"
)
重要提示：
在
database.py中，請務必將SQLALCHEMY_DATABASE_URL修改為您實際的 MySQL 連接字串（例如，替換your_mysql_root_password和multitenant_db）。在
auth.py中，SECRET_KEY必須替換為一個安全且足夠長的隨機字串。
main.py中的初始使用者密碼password_a和password_b僅用於演示。在實際生產環境中，絕不能以明文儲存密碼，必須使用bcrypt或argon2等雜湊演算法進行雜湊處理。
測試與驗證：確保數據隔離
現在，您的多租戶 FastAPI 應用程式已經準備就緒。我們可以運行它並驗證數據隔離是否正確工作。
運行應用程式：
打開終端機，進入專案根目錄 (fastapi-multitenant-api/)，然後執行：
Bashuvicorn main:app --reloadFastAPI 將在
http://127.0.0.1:8000啟動。訪問 API 文檔：
打開瀏覽器，導航到 http://127.0.0.1:8000/docs。您會看到自動生成的 Swagger UI 文檔，您可以在這裡直接測試 API。
測試流程：
3.1 租戶 A 的使用者 (
user_a) 登入並獲取 Token在 Swagger UI 中找到
POST /api/v1/login。點擊 "Try it out"，填寫
username: user_a,password: password_a。點擊 "Execute"。
複製響應中的
access_token(不包含 "Bearer ")。這就是user_a的身份憑證。
3.2 租戶 A (
user_a) 創建產品在 Swagger UI 中找到
POST /api/v1/products。點擊 "Authorize" (通常在頁面頂部或路徑旁邊)，輸入
Bearer <user_a 的 access_token>。點擊 "Try it out"，填寫產品資訊：
name: 筆記型電腦,price: 1200.0。點擊 "Execute"。
驗證響應狀態碼為
201 Created，並且響應數據中tenant_id欄位的值為user_a所屬的租戶 ID (通常是 1)。
3.3 租戶 A (
user_a) 獲取產品清單在 Swagger UI 中找到
GET /api/v1/products。確保已設定
user_a的 Authorization (Bearer Token)。點擊 "Try it out" -> "Execute"。
驗證響應中只包含您剛才創建的產品。
3.4 租戶 B 的使用者 (
user_b) 登入並獲取 Token重複步驟 3.1，但使用
username: user_b,password: password_b。複製
user_b的access_token。
3.5 租戶 B (
user_b) 嘗試獲取產品清單在 Swagger UI 中找到
GET /api/v1/products。重要：點擊 "Authorize"，清除或替換為
user_b的access_token。點擊 "Try it out" -> "Execute"。
驗證結果：您應該會看到一個空列表
[]。這是因為user_b的租戶目前沒有任何產品。這證明了數據隔離的有效性。
3.6 租戶 B (
user_b) 創建產品在 Swagger UI 中找到
POST /api/v1/products。確保已設定
user_b的 Authorization (Bearer Token)。點擊 "Try it out"，填寫產品資訊：
name: 智慧型手機,price: 800.0。點擊 "Execute"。
驗證響應狀態碼為
201 Created，且tenant_id為user_b所屬的租戶 ID (通常是 2)。
3.7 租戶 A (
user_a) 再次獲取產品清單重要：點擊 "Authorize"，再次替換為
user_a的access_token。在 Swagger UI 中找到
GET /api/v1/products。點擊 "Try it out" -> "Execute"。
驗證結果：您應該仍然只看到
user_a之前創建的產品 (筆記型電腦)，而不會看到user_b創建的智慧型手機。
這個測試流程明確地展示了多租戶數據隔離的成功實作。每個使用者都只能看到和操作他們自己租戶下的數據。
應用情境與延伸想法
這種多租戶設計模式不僅限於產品和訂單管理，還可以應用於：
使用者管理：每個租戶可以有自己的使用者群，並獨立管理。
數據分析與報告：為每個租戶生成獨立的數據報告，而無需在應用層面進行複雜的數據切分。
設定管理：允許每個租戶擁有獨立的應用程式設定。
文件儲存：文件儲存路徑或命名規則可包含
tenant_id。
延伸想法：
更強大的身份驗證：將密碼雜湊化 (例如使用
passlib和bcrypt)，並引入 OAuth2 / OpenID Connect 流程。權限管理 (RBAC)：除了
tenant_id，可以在 JWT Payload 中加入roles或permissions，實現更細粒度的權限控制 (例如，只有特定角色才能創建產品)。租戶生命週期管理：增加租戶的創建、啟用、停用、刪除功能，以及租戶資料的備份和遷移策略。
動態資料庫路由：對於極大規模的多租戶應用，可以考慮根據
tenant_id將請求路由到不同的資料庫實例或 Schema，這會更加複雜但能提供更高的擴展性。前端整合：在前端應用程式中，登入成功後儲存 JWT，並在每次 API 請求時自動附加
Authorization頭。
結語與個人觀點
使用 FastAPI 結合 JWT 實現多租戶 API 是一個非常優雅且高效的解決方案。FastAPI 的依賴注入系統 (Depends) 極大地簡化了認證和授權邏輯的整合，使開發者能夠專注於業務邏輯，同時確保了強大的數據隔離。
這種架構不僅能提升 SaaS 應用程式的安全性，還能顯著降低營運成本和開發複雜性，為未來的擴展打下堅實的基礎。作為 Python 工程師，掌握這種設計模式將使您在構建現代化、可擴展的 Web 服務時如虎添翼。
沒有留言:
張貼留言