FastAPI 資料庫整合 — NoSQL（MongoDB / Beanie）

---

簡介

在現代 Web 開發中，NoSQL 資料庫因其彈性、水平擴展性以及對非結構化資料的良好支援，越來越受到開發者青睞。MongoDB 作為最流行的文件導向資料庫之一，提供了類似 JSON 的儲存格式，讓前端與後端資料的對應更為自然。

而 FastAPI 作為一個高效、型別安全且支援自動產生 OpenAPI 文件的 Python Web 框架，與 MongoDB 的結合也變得相當簡潔。若再搭配 Beanie（一個基於 Motor 的 ODM），我們可以在 FastAPI 中以 Pythonic 的方式操作 MongoDB，享受到資料模型、驗證與非同步 I/O 的完整支援。

本篇文章將從基礎概念切入，帶領讀者一步步完成 FastAPI + MongoDB + Beanie 的整合，並提供實作範例、常見陷阱與最佳實踐，讓你能在實務專案中快速上手。

---

核心概念

1. 為什麼選擇 MongoDB + Beanie？

項目	MongoDB	Beanie
資料模型	文件（Document）結構靈活，可隨時新增欄位	以 Pydantic/ Motor 為基礎的 ODM，支援型別提示與驗證
非同步支援	原生支援 (via Motor)	完全非同步，與 FastAPI 事件迴圈相容
開發效率	需要自行撰寫 CRUD	Beanie 提供 .save()、.find() 等高階 API
資料驗證	依賴程式自行檢查	內建 Pydantic 驗證，減少錯誤

結論：若你已經在使用 FastAPI，想要快速導入 NoSQL，Beanie 是最省事且最符合 Python 生態的選擇。

2. 安裝必要套件

pip install fastapi[all] motor beanie

• fastapi[all]：包含 uvicorn、pydantic 等常用依賴。
• motor：MongoDB 的非同步驅動。
• beanie：基於 Motor 的 ODM。

3. 建立資料模型

Beanie 的模型必須繼承自 Document，同時利用 Pydantic 進行欄位驗證。

# models.py
from beanie import Document, Link
from pydantic import BaseModel, Field, EmailStr
from typing import List, Optional
from datetime import datetime

class User(Document):
    """使用者資料模型"""
    email: EmailStr = Field(..., description="使用者唯一 Email")
    name: str = Field(..., max_length=50, description="使用者名稱")
    joined_at: datetime = Field(default_factory=datetime.utcnow)

    class Settings:
        name = "users"          # MongoDB collection 名稱
        indexes = [
            "email",             # 建立 email 唯一索引
        ]

class Post(Document):
    """部落格文章模型，示範關聯 (Link)"""
    title: str = Field(..., max_length=200)
    content: str = Field(..., description="Markdown 內容")
    author: Link[User]          # 使用 Link 形成外鍵關聯
    tags: List[str] = Field(default_factory=list)
    created_at: datetime = Field(default_factory=datetime.utcnow)

    class Settings:
        name = "posts"
        indexes = [
            [("author", 1), ("created_at", -1)],  # 複合索引
        ]

重點：Link[User] 讓我們在 Post 中直接引用 User，Beanie 會在查詢時自動解析關聯。

4. 初始化 Beanie 與 MongoDB 連線

在 FastAPI 啟動時，我們要先建立 MongoDB 連線，並讓 Beanie 載入模型。

# main.py
import motor.motor_asyncio
from beanie import init_beanie
from fastapi import FastAPI
from models import User, Post

app = FastAPI(title="FastAPI + MongoDB (Beanie) Demo")

@app.on_event("startup")
async def on_startup():
    """FastAPI 啟動時初始化 MongoDB 連線與 Beanie"""
    client = motor.motor_asyncio.AsyncIOMotorClient(
        "mongodb://localhost:27017",  # 依需求調整 URI
        uuidRepresentation="standard"
    )
    await init_beanie(
        database=client.my_fastapi_db,   # 指定資料庫名稱
        document_models=[User, Post]     # 載入模型
    )

小技巧：若在 Docker 或雲端環境，請將 URI 抽離至環境變數，避免硬編碼。

5. 基本 CRUD 範例

以下示範 建立、讀取、更新、刪除 四大操作，全部使用非同步函式，保持與 FastAPI 的一致性。

5.1 建立使用者

# routers/user.py
from fastapi import APIRouter, HTTPException, status
from models import User

router = APIRouter(prefix="/users", tags=["users"])

@router.post("/", response_model=User, status_code=status.HTTP_201_CREATED)
async def create_user(user: User):
    """建立新使用者，若 email 重複則拋出 409 Conflict"""
    existing = await User.find_one(User.email == user.email)
    if existing:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail="Email already registered"
        )
    await user.insert()   # Beanie 內建的 insert
    return user

5.2 取得使用者列表（分頁）

@router.get("/", response_model=List[User])
async def list_users(skip: int = 0, limit: int = 10):
    """支援簡易分頁的使用者列表"""
    users = await User.find().skip(skip).limit(limit).to_list()
    return users

5.3 更新使用者名稱

@router.patch("/{user_id}", response_model=User)
async def update_user_name(user_id: str, name: str):
    """僅更新 name 欄位，使用 .replace() 會覆寫整筆文件"""
    user = await User.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    user.name = name
    await user.save()   # 只更新變更的欄位
    return user

5.4 刪除使用者

@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(user_id: str):
    """刪除使用者，同時也會刪除其相關文章（示範 cascade）"""
    user = await User.get(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    await user.delete()
    # 若想同步刪除關聯文章，可自行實作 cascade
    return None

5.5 建立文章並關聯使用者

# routers/post.py
from fastapi import APIRouter, HTTPException, status
from models import Post, User
from beanie import PydanticObjectId

router = APIRouter(prefix="/posts", tags=["posts"])

@router.post("/", response_model=Post, status_code=status.HTTP_201_CREATED)
async def create_post(
    title: str,
    content: str,
    author_id: PydanticObjectId,
    tags: List[str] = None,
):
    """建立新文章，author 以 Link 方式關聯 User"""
    author = await User.get(author_id)
    if not author:
        raise HTTPException(status_code=404, detail="Author not found")
    post = Post(
        title=title,
        content=content,
        author=author,
        tags=tags or []
    )
    await post.insert()
    return post

5.6 取得文章並自動展開作者資訊

@router.get("/{post_id}", response_model=Post)
async def get_post(post_id: PydanticObjectId):
    """使用 .fetch_link() 展開 author 資訊"""
    post = await Post.get(post_id, fetch_links=True)
    if not post:
        raise HTTPException(status_code=404, detail="Post not found")
    return post

說明：fetch_links=True 會在一次查詢中把 author 的完整文件一起帶回，減少 N+1 查詢問題。

6. 進階查詢與聚合

Beanie 同時支援 MongoDB 聚合管線，讓你能在資料庫層面完成複雜的統計與分組。

@router.get("/stats/tags")
async def tag_statistics():
    """統計所有文章的標籤出現次數"""
    pipeline = [
        {"$unwind": "$tags"},
        {"$group": {"_id": "$tags", "count": {"$sum": 1}}},
        {"$sort": {"count": -1}}
    ]
    result = await Post.get_motor_collection().aggregate(pipeline).to_list()
    # 轉換為友善格式
    return {doc["_id"]: doc["count"] for doc in result}

---

常見陷阱與最佳實踐

陷阱	說明	解決方案
未使用非同步 I/O	直接使用 pymongo（同步）會阻塞事件迴圈，降低效能。	必須使用 motor（Beanie 已內建）並確保所有 DB 操作都是 await。
索引遺漏	大量查詢時若無適當索引，會導致全表掃描（Full Collection Scan）。	在模型 Settings.indexes 中定義常用欄位索引，部署前使用 mongosh 確認索引建立成功。
資料驗證不完整	只依賴前端驗證，後端仍可能收到不合法資料。	利用 Pydantic 的 Field、validator 完整校驗，並在模型層面設定 unique、max_length 等限制。
ObjectId 轉型錯誤	直接把字串傳入 User.get() 會拋出 InvalidId。	使用 PydanticObjectId 或 ObjectId 先行轉型，或在路由參數上加 type: PydanticObjectId。
關聯刪除未處理	刪除使用者後，相關文章仍留存造成孤兒資料。	實作 cascade delete（在 User.delete() 前先刪除或更新相關 Post），或使用 MongoDB TTL。

最佳實踐

1. 統一環境變數：將 MongoDB URI、資料庫名稱、連線池大小等抽離至 .env，使用 python-dotenv 載入。
2. 分層架構：將路由、服務（service layer）與資料模型分離，讓測試與維護更容易。
3. 加入測試：使用 pytest-asyncio 撰寫非同步單元測試，確保 CRUD 邏輯正確。
4. 日誌與監控：結合 structlog 或 loguru 記錄 DB 操作，配合 MongoDB Atlas 的監控儀表板偵測慢查詢。
5. 安全性：避免在 API 中直接回傳完整的 MongoDB _id，可自行封裝 DTO，或使用 BaseModel 的 Config 來隱藏欄位。

---

實際應用場景

場景	為什麼選擇 MongoDB + Beanie
部落格或內容平台	文章與評論屬於文件型資料，結構多變且常有標籤、分類等陣列欄位，MongoDB 的文件模型最為合適。
即時聊天系統	訊息以 JSON 形式存儲，且需要高寫入吞吐量與水平擴展，MongoDB 的分片（sharding）可輕鬆應對。
使用者設定與偏好	每位使用者的設定欄位可能不盡相同，使用文件儲存可以避免大量空欄位的浪費。
分析儀表板	結合聚合管線快速產生統計圖表，無需額外 ETL 流程。
原型開發	只要有 models.py 即可快速產出 CRUD API，縮短開發週期。

案例：某電商平台在商品評論功能上，採用了 FastAPI + Beanie，將每筆評論存為獨立文件，並以 product_id 建立索引。透過聚合管線即時算出平均星級與最近 10 筆評論，大幅提升前端渲染速度與使用者體驗。

---

總結

• MongoDB 為文件導向的 NoSQL 資料庫，提供彈性與水平擴展的特性；Beanie 則是與 FastAPI 完美配合的非同步 ODM，讓資料模型、驗證與 CRUD 操作變得簡潔且型別安全。
• 本文從安裝、模型定義、資料庫初始化，到完整的 CRUD 範例與聚合查詢，示範了如何在 FastAPI 專案中快速整合 MongoDB。
• 為避免常見陷阱，我們建議在開發階段即建立索引、使用非同步 I/O、妥善驗證資料，並依照最佳實踐將設定、路由與服務層分離。
• 最後，透過實務場景的說明，你可以更清楚地判斷何時適合使用 MongoDB + Beanie，並在專案中發揮其高效與彈性的優勢。

祝你在 FastAPI 與 NoSQL 的旅程中寫出 高效、可維護、易擴充 的 API！ 🚀