FastAPI 是一個現代、高效的 Python Web 框架，以其效能和易用性而聞名。構建 Web 應用程式時，設計清晰的路由系統和穩定的資料函式庫連線至關重要。本文將逐步講解如何在 FastAPI 應用程式中建立路由，處理使用者驗證和事件管理，並整合 SQLModel 和 MongoDB 資料函式庫，實作資料的持久化儲存和操作。從路由的規劃、資料模型的定義，到資料函式庫的連線與 CRUD 操作，文章提供完整的程式碼範例和詳細的解析，協助讀者掌握 FastAPI 開發的精髓。透過整合 SQLModel 的 ORM 功能，簡化資料函式庫操作流程，並探討如何使用 Beanie 與 MongoDB 互動，提供更靈活的資料儲存方案。
在FastAPI應用程式中建立路由系統
在建立應用程式的過程中，下一步是設定API的路由系統。我們將設計用於事件和使用者的路由系統。使用者路由將包括登入、登出和註冊路由。經過驗證的使用者將能夠存取用於建立、更新和刪除事件的路由，而公眾則可以在事件建立後檢視它。下圖顯示了兩個路由之間的關係：
使用者路由
現在我們已經清楚瞭解了圖5.2中要實作的路由，接下來我們將從定義users.py中的使用者路由開始。讓我們看看具體步驟：
- 首先定義一個基本的註冊路由：
from fastapi import APIRouter, HTTPException, status
from models.user import User, UserSignIn
user_router = APIRouter(tags=["User"])
users = {}
@user_router.post("/signup")
async def sign_new_user(data: NewUser) -> dict:
if data.email in users:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="提供的使用者名稱已存在"
)
users[data.email] = data
return {
"message": "使用者註冊成功！"
}
內容解密：
此註冊路由使用了應用程式內建的資料函式庫（在第6章《連線到資料函式庫》中，我們將引入真正的資料函式庫）。該路由在新增新使用者之前檢查資料函式庫中是否已存在具有相同電子郵件地址的使用者。
- 接下來實作登入路由：
@user_router.post("/signin")
async def sign_user_in(user: UserSignIn) -> dict:
if user.email not in users:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="使用者不存在"
)
if users[user.email].password != user.password:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="提供的憑證錯誤"
)
return {
"message": "使用者登入成功"
}
內容解密：
在此路由中，首先檢查資料函式庫中是否存在該使用者，如果不存在則引發異常。如果使用者存在，則應用程式會檢查密碼是否匹配，然後傳回成功訊息或異常。
- 現在我們已經定義了使用者操作的路由，接下來在
main.py中註冊它們並啟動應用程式。首先匯入所需的函式庫和使用者路由定義：
from fastapi import FastAPI
from routes.user import user_router
import uvicorn
- 然後，建立一個FastAPI例項並註冊路由和應用程式：
app = FastAPI()
app.include_router(user_router, prefix="/user")
if __name__ == "__main__":
uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)
內容解密：
在此程式碼區塊中，我們建立了一個FastAPI例項並註冊了路由。然後使用uvicorn.run()方法在8080埠啟動應用程式，並將熱過載設定為True。
事件路由
有了使用者路由之後，下一步是實作事件操作的路由。讓我們看看具體步驟：
- 首先匯入所需的相依性和定義事件路由器：
from fastapi import APIRouter, Body, HTTPException, status
from models.events import Event
from typing import List
event_router = APIRouter(tags=["Events"])
events = []
- 接下來定義用於檢索資料函式庫中所有事件和與提供的ID相匹配的事件的路由：
@event_router.get("/", response_model=List[Event])
async def retrieve_all_events() -> List[Event]:
return events
@event_router.get("/{id}", response_model=Event)
async def retrieve_event(id: int) -> Event:
for event in events:
if event.id == id:
return event
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="提供的ID的事件不存在"
)
內容解密：
在第二個路由中，當提供的ID的事件不存在時，我們引發了一個HTTP_404_NOT_FOUND異常。
此圖示顯示了使用者和事件操作的關係： 此圖示清晰地展示了 FastAPI 應用程式中的不同路由之間的互動關係，有助於理解整個系統的架構。
在FastAPI應用程式中建立事件路由與資料函式庫連線
事件路由的實作
在前一章中，我們學習瞭如何架構FastAPI應用程式並實作了基本的路由和模型。在本章中，我們將進一步實作事件路由，包括建立事件、刪除單一事件和刪除所有事件。
事件路由程式碼實作
@event_router.post("/new")
async def create_event(body: Event = Body(...)) -> dict:
events.append(body)
return {
"message": "事件建立成功"
}
@event_router.delete("/{id}")
async def delete_event(id: int) -> dict:
for event in events:
if event.id == id:
events.remove(event)
return {
"message": "事件刪除成功"
}
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="提供的ID事件不存在"
)
@event_router.delete("/")
async def delete_all_events() -> dict:
events.clear()
return {
"message": "所有事件刪除成功"
}
程式碼解析
create_event函式：接收一個Event物件並將其新增到events列表中。- 輸入引數：
body（型別為Event，透過Body(...)註解） - 輸出結果：一個字典，包含成功訊息
- 輸入引數：
delete_event函式：根據提供的ID刪除對應的事件。- 輸入引數：
id（型別為整數） - 輸出結果：一個字典，包含成功訊息；若事件不存在，則丟擲
HTTPException（404狀態碼）
- 輸入引數：
delete_all_events函式：刪除所有事件。- 輸出結果：一個字典，包含成功訊息
測試事件路由
透過curl命令測試各個路由：
- GET路由：取得所有事件（初始為空列表）
- POST路由：建立一個新事件
- GET路由（帶ID）：取得特定ID的事件
- DELETE路由：刪除特定ID的事件
- DELETE路由（無ID）：刪除所有事件
連線到資料函式庫
在前面的章節中，我們使用應用程式內部的資料函式庫來儲存事件。然而，這種方式在應用程式重啟或當機時會導致資料丟失。因此，我們需要將應用程式遷移到使用正式的資料函式庫。
設定SQLModel
首先，安裝SQLModel函式庫：
pip install sqlmodel
SQLModel中的表格與操作
SQLModel是一個結合了Pydantic和SQLAlchemy的函式庫，使得定義模型變得更加容易。
表格概念
- 表格：儲存資料的基本單位，由列（欄位）和行（記錄）組成。
遷移到MongoDB資料函式庫
除了SQL資料函式庫外，我們還將探討如何使用Beanie連線到MongoDB資料函式庫。
設定MongoDB
- 安裝MongoDB並啟動服務。
- 安裝Beanie函式庫：
pip install beanie
使用Beanie進行CRUD操作
Beanie是一個非同步的MongoDB ODM（Object Document Mapper），使得在FastAPI應用程式中使用MongoDB變得更加方便。
使用 SQLModel 建立資料函式庫與執行 CRUD 操作
在現代的 Web 應用程式開發中，與資料函式庫的互動是不可或缺的一部分。SQLModel 是一個結合了 SQLAlchemy 和 Pydantic 的函式庫，讓開發者能夠更輕鬆地與資料函式庫進行互動。本文將介紹如何使用 SQLModel 建立資料函式庫、定義資料表模型、以及執行 CRUD（建立、讀取、更新、刪除）操作。
定義資料表模型
在使用 SQLModel 建立資料表之前，我們需要定義一個資料表模型類別。這個類別需要繼承自 SQLModel，並且設定 table=True 以表示這是一個資料表模型。
from sqlmodel import SQLModel, Field
from typing import Optional, List
class Event(SQLModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
title: str
image: str
description: str
location: str
tags: List[str]
內容解密：
Event類別代表了一個名為Event的資料表。id欄位是主鍵，用於唯一識別每筆資料。- 其他欄位如
title、image、description、location和tags代表了資料表的各個欄位。 tags欄位使用了List[str]，表示它可以儲存多個字串值。
建立資料函式庫連線
要與資料函式庫互動，我們需要建立一個資料函式庫連線。SQLModel 使用 SQLAlchemy 的引擎來管理資料函式庫連線。
from sqlmodel import create_engine
database_file = "planner.db"
database_connection_string = f"sqlite:///{database_file}"
engine = create_engine(database_connection_string, echo=True)
內容解密：
create_engine函式用於建立一個 SQLAlchemy 引擎例項。database_connection_string指定了資料函式庫的連線字串，這裡使用的是 SQLite 資料函式庫。echo=True表示當執行 SQL 命令時，會將命令列印到控制檯。
建立資料表
定義好資料表模型後，我們需要建立資料表。
SQLModel.metadata.create_all(engine)
內容解密：
SQLModel.metadata.create_all(engine)會根據已定義的資料表模型，在資料函式庫中建立相應的資料表。
執行 CRUD 操作
建立（Create）
要插入資料到資料表中，我們需要建立一個資料表模型的例項，然後使用 Session 物件將其新增到資料函式庫。
from sqlmodel import Session
new_event = Event(title="Book Launch", image="src/fastapi.png", description="The book launch event will be held at Packt HQ, Packt city", location="Google Meet", tags=["packt", "book"])
with Session(engine) as session:
session.add(new_event)
session.commit()
內容解密：
- 建立了一個
Event的例項new_event。 - 使用
Session物件將new_event新增到資料函式庫中，並提交事務。
讀取（Read）、更新（Update）和刪除（Delete）
讀取、更新和刪除操作同樣是透過 Session 物件來執行。具體的實作細節可以參考 SQLModel 的官方檔案。
連線應用程式與資料函式庫
為了讓 FastAPI 應用程式能夠與資料函式庫互動，我們需要在應用程式啟動時建立資料函式庫連線。
from fastapi import FastAPI
from database.connection import conn
app = FastAPI()
@app.on_event("startup")
def on_startup():
conn()
內容解密：
on_startup函式在應用程式啟動時被呼叫，用於建立資料函式庫連線。