FastAPI 整合 MongoDB 資料函式庫 CRUD 操作實作

在現代 Web 開發中，API 伺服器與資料函式庫的互動至關重要。本文將示範如何使用 FastAPI 框架與 MongoDB 資料函式庫進行整合，並使用 Beanie ODM 函式庫簡化資料函式庫操作。我們將逐步講解如何定義 MongoDB 檔案模型、初始化資料函式庫連線，以及如何實作新增、讀取、更新和刪除等 CRUD 操作。透過 Beanie ODM，我們可以以更 Pythonic 的方式操作 MongoDB，並受益於其非同步的特性，提升應用程式的效能。文章中將提供完整的程式碼範例，方便讀者理解和實作。

資料函式庫建立與事件管理實作

在前面的章節中，我們已經成功建立了資料函式庫連線並定義了事件模型。現在，我們將進一步實作事件的新增、讀取、更新和刪除（CRUD）操作。

建立事件

首先，我們需要更新 routes/events.py 檔案，以包含 Event 模型類別和 get_session() 函式。這樣，我們就可以在路由中使用資料函式庫會話物件。

from fastapi import APIRouter, Depends, HTTPException, Request, status
from database.connection import get_session
from models.events import Event, EventUpdate

接下來，我們更新 create_event() 函式，以使用資料函式庫會話建立新的事件：

@event_router.post("/new")
async def create_event(new_event: Event, session=Depends(get_session)) -> dict:
    session.add(new_event)
    session.commit()
    session.refresh(new_event)
    return {"message": "Event created successfully"}

程式碼解析：

session=Depends(get_session)：這裡使用了 Depends 類別來進行依賴注入，確保在執行資料函式庫操作之前，會話物件已經被正確初始化。
session.add(new_event)：將新的事件物件新增到會話中。
session.commit()：提交會話中的變更，將資料寫入資料函式庫。
session.refresh(new_event)：重新整理事件物件，以取得資料函式庫中自動生成的 ID 等資訊。

測試事件建立

使用 curl 命令測試事件建立功能：

curl -X 'POST' \
'http://0.0.0.0:8080/event/new' \
-H 'accept: application/json' \
-H 'Content-Type: application/json' \
-d '{
"title": "FastAPI Book Launch",
"image": "fastapi-book.jpeg",
"description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",
"tags": ["python", "fastapi", "book", "launch"],
"location": "Google Meet"
}'

讀取事件

更新 retrieve_all_events() 和 retrieve_event() 函式，以從資料函式庫中讀取事件：

@event_router.get("/", response_model=List[Event])
async def retrieve_all_events(session=Depends(get_session)) -> List[Event]:
    statement = select(Event)
    events = session.exec(statement).all()
    return events

@event_router.get("/{id}", response_model=Event)
async def retrieve_event(id: int, session=Depends(get_session)) -> Event:
    event = session.get(Event, id)
    if event:
        return event
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event with supplied ID does not exist")

程式碼解析：

statement = select(Event)：使用 SQLAlchemy 的 select() 函式來查詢所有事件。
events = session.exec(statement).all()：執行查詢並取得所有事件。
event = session.get(Event, id)：根據 ID 取得單個事件。

更新事件

更新 update_event() 函式，以更新現有的事件：

@event_router.put("/edit/{id}", response_model=Event)
async def update_event(id: int, new_data: EventUpdate, session=Depends(get_session)) -> Event:
    event = session.get(Event, id)
    if event:
        event_data = new_data.dict(exclude_unset=True)
        for key, value in event_data.items():
            setattr(event, key, value)
        session.add(event)
        session.commit()
        session.refresh(event)
        return event
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event with supplied ID does not exist")

程式碼解析：

event_data = new_data.dict(exclude_unset=True)：將更新資料轉換為字典，並排除未設定的欄位。
setattr(event, key, value)：更新事件物件的屬性。

刪除事件

更新 delete_event() 函式，以刪除現有的事件：

@event_router.delete("/delete/{id}")
async def delete_event(id: int, session=Depends(get_session)) -> dict:
    event = session.get(Event, id)
    if event:
        session.delete(event)
        session.commit()
        return {"message": "Event deleted successfully"}
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Event with supplied ID does not exist")

程式碼解析：

session.delete(event)：刪除事件物件。
session.commit()：提交變更，將事件從資料函式庫中刪除。

本章節中，我們成功實作了事件的新增、讀取、更新和刪除操作，並使用 curl 命令進行測試。這些操作都與資料函式庫進行了互動，確保了資料的持久化儲存。

使用 Beanie 和 MongoDB 實作 CRUD 操作

在前面的章節中，我們已經成功地將 SQL 資料函式庫整合到我們的 FastAPI 應用程式中，並實作了 CRUD 操作。現在，我們將切換到 MongoDB 作為資料函式庫平台，並使用 Beanie 這個非同步的 Object Document Mapper (ODM) 函式庫來執行資料函式庫操作。

設定 MongoDB

首先，我們需要安裝 Beanie 函式庫。執行以下命令：

(venv)$ pip install beanie

檔案定義

在 MongoDB 中，資料儲存在檔案中，而不是表格中。檔案的定義方式與 Pydantic 模型類別似，但需要繼承 Beanie 的 Document 類別。以下是一個範例：

from beanie import Document

class Event(Document):
    name: str
    location: str

    class Settings:
        name = "events"

CRUD 操作方法

Beanie 提供了多種方法來執行 CRUD 操作：

.insert() 和 .create()：用於建立新的檔案記錄。
.find() 和 .get()：用於檢索檔案記錄。
.update() 和 .upsert()：用於更新檔案記錄。
.delete()：用於刪除檔案記錄。

以下是一些範例：

event = Event(name="Packt office launch", location="Hybrid")
await event.create()

event = await Event.get("74478287284ff")
update_query = {"$set": {"location": "virtual"}}
await event.update(update_query)

event = await Event.get("74478287284ff")
await event.delete()

初始化資料函式庫

要初始化資料函式庫，我們需要在 database 資料夾中建立一個 connection.py 檔案，並新增以下內容：

from beanie import init_beanie
from motor.motor_asyncio import AsyncIOMotorClient
from typing import Optional
from pydantic import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: Optional[str] = None

    async def initialize_database(self):
        client = AsyncIOMotorClient(self.DATABASE_URL)
        await init_beanie(
            database=client.get_default_database(),
            document_models=[]
        )

    class Config:
        env_file = ".env"

更新模型檔案

我們需要更新 models 資料夾中的模型檔案，以包含 MongoDB 檔案的定義。例如，在 models/events.py 中：

from beanie import Document
from typing import Optional, List

class Event(Document):
    title: str
    image: str
    description: str
    tags: List[str]
    location: str

    class Config:
        schema_extra = {
            "example": {
                "title": "FastAPI Book Launch",
                "image": "https://linktomyimage.com/image.png",
                "description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",
                "tags": ["python", "fastapi", "book", "launch"],
                "location": "Google Meet"
            }
        }

    class Settings:
        name = "events"

建立 Pydantic 模型

我們需要建立一個 Pydantic 模型來處理更新操作。例如，在 models/events.py 中：

class EventUpdate(BaseModel):
    title: Optional[str]
    image: Optional[str]
    description: Optional[str]
    tags: Optional[List[str]]
    location: Optional[str]

    class Config:
        schema_extra = {
            "example": {
                "title": "FastAPI Book Launch",
                "image": "https://linktomyimage.com/image.png",
                "description": "We will be discussing the contents of the FastAPI book in this event. Ensure to come with your own copy to win gifts!",
                "tags": ["python", "fastapi", "book", "launch"],
                "location": "Google Meet"
            }
        }

MongoDB 資料函式庫初始化與 CRUD 操作實作

在開發根據 FastAPI 的應用程式時，與 MongoDB 資料函式庫的整合是至關重要的環節。本文將詳細介紹如何初始化 MongoDB 資料函式庫，並實作基本的 CRUD（建立、讀取、更新、刪除）操作。

定義檔案模型

首先，我們需要定義 MongoDB 中的檔案模型。以下是一個範例，展示瞭如何使用 Beanie ODM 來定義 User 和 Event 檔案模型。

from beanie import Document, Link
from pydantic import EmailStr
from typing import Optional, List

class User(Document):
    email: EmailStr
    password: str
    events: Optional[List[Link["Event"]]]

    class Settings:
        name = "users"

    class Config:
        schema_extra = {
            "example": {
                "email": "fastapi@packt.com",
                "password": "strong!!!",
                "events": [],
            }
        }

class UserSignIn(BaseModel):
    email: EmailStr
    password: str

內容解密：

User 檔案模型定義了使用者的電子郵件、密碼以及與 Event 檔案的關聯。
email 欄位使用了 EmailStr 型別驗證，確保輸入的電子郵件格式正確。
events 欄位是一個可選的列表，用於儲存與使用者相關聯的事件 ID。

初始化資料函式庫

接下來，我們需要在 connection.py 中初始化 MongoDB 資料函式庫。

from models.users import User
from models.events import Event
from beanie import init_beanie

async def initialize_database(self):
    client = AsyncIOMotorClient(self.DATABASE_URL)
    await init_beanie(
        database=client.get_default_database(),
        document_models=[Event, User]
    )

內容解密：

initialize_database 方法負責初始化 Beanie ODM 與 MongoDB 的連線。
document_models 引數指定了需要被 Beanie 管理的檔案模型。

CRUD 操作實作

為了簡化 CRUD 操作，我們建立了一個 Database 類別，該類別接受一個檔案模型作為引數。

class Database:
    def __init__(self, model):
        self.model = model

    async def save(self, document) -> None:
        await document.create()
        return

    async def get(self, id: PydanticObjectId) -> Any:
        doc = await self.model.get(id)
        if doc:
            return doc
        return False

    async def get_all(self) -> List[Any]:
        docs = await self.model.find_all().to_list()
        return docs

    async def update(self, id: PydanticObjectId, body: BaseModel) -> Any:
        # 更新邏輯實作
        pass

    async def delete(self, id: PydanticObjectId) -> bool:
        # 刪除邏輯實作
        pass

內容解密：

Database 類別封裝了基本的 CRUD 操作，使得對不同檔案模型的資料函式庫操作變得更加統一和方便。
每個方法都針對特定的資料函式庫操作，例如 save 用於新增檔案，get 用於檢索單個檔案等。

路由更新

最後，我們需要更新路由處理函式，以使用 Database 類別進行 CRUD 操作。

event_database = Database(Event)

@event_router.get("/", response_model=List[Event])
async def retrieve_all_events() -> List[Event]:
    events = await event_database.get_all()
    return events

@event_router.post("/new")
async def create_event(body: Event) -> dict:
    await event_database.save(body)
    return {"message": "Event created successfully"}

內容解密：

路由處理函式現在透過 event_database 例項呼叫相應的 CRUD 方法。
這種設計使得業務邏輯與資料函式庫操作解耦，提高了程式碼的可維護性。

透過上述步驟，我們成功地在 FastAPI 應用程式中整合了 MongoDB 資料函式庫，並實作了基本的 CRUD 操作。這種架構設計不僅簡化了資料函式庫操作，也提高了程式碼的可讀性和可擴充套件性。

玄貓 BlackCat

技術愛好者，專注於分享程式開發、雲端技術與 AI 應用的心得體會。