FastAPI 連接 NoSQL 資料庫 (MongoDB)

MongoDB 是一種非常流行的 NoSQL 資料庫，它的文件結構 (BSON) 與 JSON 非常貼合，因此在 Web 開發中非常受歡迎。

為了發揮 FastAPI 的非同步效能，我們推薦使用 Motor，這是 MongoDB 官方提供的 Python 非同步驅動程式。

安裝 Motor

pip install motor

資料庫連線

通常我們會將資料庫連線放在 lifespan 或 startup event 中。

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()

@app.on_event("startup")
async def startup_db_client():
    app.mongodb_client = AsyncIOMotorClient("mongodb://localhost:27017")
    app.mongodb = app.mongodb_client["my_database"]

@app.on_event("shutdown")
async def shutdown_db_client():
    app.mongodb_client.close()

定義 Pydantic Model with ObjectId

MongoDB 使用 _id (ObjectId) 作為主鍵，但 ObjectId 不是標準 JSON 格式，這可能會在 Pydantic 驗證或 JSON 序列化時出問題。

我們需要稍微客製化一下 Pydantic Model：

from typing import Optional, List
from pydantic import BaseModel, Field, BeforeValidator
from typing_extensions import Annotated

# 定義一個可以處理 ObjectId 的型別
PyObjectId = Annotated[str, BeforeValidator(str)]

class StudentModel(BaseModel):
    # 將 MongoDB 的 _id 對應到 id 欄位
    id: PyObjectId | None = Field(alias="_id", default=None)
    name: str = Field(...)
    email: str = Field(...)
    course: str = Field(...)
    gpa: float = Field(..., le=4.0)

    model_config = {
        "populate_by_name": True,
        "arbitrary_types_allowed": True,
    }

實作 CRUD

from fastapi import Body, HTTPException, status
from fastapi.encoders import jsonable_encoder

@app.post("/students/", response_description="Add new student", response_model=StudentModel)
async def create_student(student: StudentModel = Body(...)):
    student = jsonable_encoder(student)
    new_student = await app.mongodb["students"].insert_one(student)
    created_student = await app.mongodb["students"].find_one({"_id": new_student.inserted_id})
    return created_student

@app.get("/students/", response_description="List all students", response_model=List[StudentModel])
async def list_students(limit: int = 1000):
    students = await app.mongodb["students"].find().to_list(limit)
    return students

ODM 工具 (Beanie)

雖然直接用 Motor 很靈活，但寫起來比較繁瑣。如果你喜歡類似 ORM 的開發體驗，推薦使用 Beanie。

Beanie 是一個基於 Motor 和 Pydantic 的 ODM (Object Document Mapper)。

使用 Beanie 初始化

from beanie import init_beanie, Document

class Student(Document):
    name: str
    gpa: float

@app.on_event("startup")
async def startup_event():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    await init_beanie(database=client.db_name, document_models=[Student])

使用 Beanie 操作資料

# 新增
student = Student(name="John", gpa=3.8)
await student.insert()

# 查詢
students = await Student.find(Student.gpa > 3.0).to_list()

總結

1. 使用 Motor 進行非同步 MongoDB 操作。
2. 處理 ObjectId 需要一些額外的 Pydantic 設定。
3. 推薦使用 Beanie ODM，它能讓 MongoDB 的操作像操作 Python 物件一樣簡單，完美整合 Pydantic。