FastAPI 框架以其非同步特性及高效能著稱，搭配資料函式庫使用時更能顯現其優勢。本文將探討如何結合 SQLAlchemy Core 和 PyMongo 進行資料函式庫操作，並以非同步方式提升 API 效能。同時，也將示範如何正確處理資料函式庫交易，確保資料一致性，並提供錯誤處理策略，提升應用程式穩定性。此外，文章還會比較 PyMongo 和 Motor 驅動程式在同步與非同步操作上的差異，幫助開發者根據專案需求選擇合適的工具。
from fastapi import FastAPI
import books.router, albums.router
app = FastAPI()
app.include_router(books.router.books)
app.include_router(albums.router.albums)
@app.get("/")
async def root():
return {"message": "首頁"}
詳細解析：FastAPI中的路徑操作與資料函式庫互動
在前面的章節中，我們已經瞭解瞭如何使用SQLAlchemy Core來與關係型資料函式庫進行互動，以及如何使用PyMongo來與MongoDB進行互動。本文將探討如何在FastAPI中實作路徑操作與資料函式庫互動的最佳實踐。
路徑操作的非同步處理
FastAPI的一大特色是其對非同步操作的原生支援。在處理路徑操作時，非同步函式可以提高應用的效能和回應速度。下面的範例展示瞭如何在FastAPI中使用非同步路徑操作函式來與資料函式庫互動：
@app.get("/books/{id}")
async def get_book(id: int, db=Depends(get_db)):
query = booklist.select().where(booklist.c.id == id)
return await db.fetch_one(query)
在這個範例中，get_book函式是一個非同步函式，它使用await關鍵字等待資料函式庫查詢的結果。這種模式確保了在等待資料函式庫操作的同時，其他請求可以被處理，從而提高了應用的整體效能。
資料函式庫交易的正確處理
在進行多步資料函式庫操作時，正確處理交易（transaction）是非常重要的。交易可以確保一系列操作的原子性，即要麼全部成功，要麼全部失敗，從而保持資料的一致性。
async def create_book(db, book: Book):
try:
async with db.transaction():
query = booklist.insert().values(id=book.id, title=book.title, author=book.author, price=book.price, publisher=book.publisher)
await db.execute(query)
# 其他相關操作
return "Book created successfully"
except Exception as e:
return f"Error: {e}"
在這個範例中，使用了async with db.transaction():來確保操作的原子性。如果任何一步操作失敗，交易將被回復，從而保持資料的一致性。
錯誤處理的最佳實踐
在與資料函式庫互動時，錯誤處理是非常重要的。正確地捕捉和處理錯誤，可以提高應用的穩定性和使用者經驗。
@app.get("/books/{id}")
async def get_book(id: int, db=Depends(get_db)):
try:
query = booklist.select().where(booklist.c.id == id)
result = await db.fetch_one(query)
if result is None:
raise HTTPException(status_code=404, detail="Book not found")
return result
except Exception as e:
raise HTTPException(status_code=500, detail=f"Internal Server Error: {e}")
在這個範例中，我們捕捉了可能的異常，並根據異常型別傳回適當的HTTP狀態碼和錯誤訊息。這樣不僅提高了應用的健壯性，也為客戶端提供了有用的錯誤訊息。
在FastAPI中使用MongoDB：非同步與同步驅動程式的比較
在現代Web開發中，資料函式庫的選擇對於應用程式的效能和可擴充套件性至關重要。MongoDB作為一種流行的NoSQL資料函式庫，因其靈活性和高效能而被廣泛使用。在本章中，我們將探討如何在FastAPI應用程式中使用MongoDB，並比較使用PyMongo和Motor驅動程式的差異。
PyMongo：同步操作
PyMongo是MongoDB官方提供的Python驅動程式，用於與MongoDB資料函式庫進行互動。儘管它提供了豐富的功能，但其操作是同步的，這意味著在進行資料函式庫操作時，應用程式的其他部分將被阻塞，直到操作完成。
連線到MongoDB
首先，我們需要建立一個函式來取得MongoDB的Collection物件。這可以透過以下程式碼實作：
from pymongo import MongoClient
def get_collection():
client = MongoClient()
DB = "mydata"
coll = "books"
bc = client[DB][coll]
yield bc
實作CRUD操作
使用PyMongo，我們可以輕鬆實作CRUD（建立、讀取、更新、刪除）操作。以下是一些範例：
from fastapi import FastAPI, Depends
from pydantic import BaseModel
app = FastAPI()
class Book(BaseModel):
bookID: int
title: str
author: str
price: int
publisher: str
@app.post("/books")
def add_book(b1: Book, bc = Depends(get_collection)):
result = bc.insert_one(b1.dict())
return "Book added successfully"
@app.get("/books", response_model=List[Book])
def get_books(bc = Depends(get_collection)):
books = list(bc.find())
return books
@app.get("/books/{id}", response_model=Book)
def get_book(id: int, bc = Depends(get_collection)):
b1 = bc.find_one({"bookID": id})
return b1
內容解密：
get_collection函式負責傳回一個MongoDB的Collection物件，供其他函式使用。add_book函式透過PyMongo的insert_one方法將新的檔案插入到集合中。get_books函式使用find方法檢索集合中的所有檔案，並將其轉換為Pydantic模型的列表。get_book函式根據提供的bookID檢索單個檔案。
Motor：非同步操作
Motor是MongoDB官方提供的另一個Python驅動程式，專為非同步操作設計。它根據PyMongo構建，但提供了非同步的API，使得在FastAPI等非同步框架中使用MongoDB變得更加高效。
連線到MongoDB
使用Motor連線到MongoDB的程式碼如下：
import motor.motor_asyncio
def get_collection():
client = motor.motor_asyncio.AsyncIOMotorClient()
DB = "mydb"
coll = "books"
bc = client[DB][coll]
yield bc
實作非同步CRUD操作
以下是使用Motor實作的非同步CRUD操作的範例：
@app.post("/books")
async def add_book(b1: Book, bc = Depends(get_collection)):
result = await bc.insert_one(b1.dict())
return "Book added successfully"
@app.get("/books", response_model=List[Book])
async def get_books(bc = Depends(get_collection)):
books = await bc.find().to_list(1000)
return books
@app.get("/books/{id}", response_model=Book)
async def get_book(id: int, bc = Depends(get_collection)):
b1 = await bc.find_one({"bookID": id})
return b1
內容解密：
- 使用Motor時，需要在CRUD操作函式前加上
async關鍵字，使其成為非同步函式。 - 在非同步函式中，使用
await關鍵字等待非同步操作的完成，如insert_one、find和find_one。 find方法傳回一個查詢結果集，需要呼叫to_list方法來具體執行查詢並取得結果列表。
隨著Web開發技術的不斷進步，非同步操作和高效能資料庫存取將變得越來越重要。未來，我們可以預見更多的應用將採用非同步框架和NoSQL資料函式庫，如FastAPI和MongoDB。掌握這些技術，將使開發者能夠更好地應對日益增長的效能需求，構建出更快速、更可靠的Web應用。
參考資料
透過結合FastAPI和MongoDB，開發者可以建立出高效、可擴充套件且易於維護的Web應用。無論是使用PyMongo還是Motor，關鍵在於根據具體需求選擇最合適的工具和技術，以實作最佳的效能和開發效率。
第7章 更龐大的應用程式架構
在前面的章節中，我們已經瞭解FastAPI網頁應用程式通常被包含在單一的Python指令碼（通常命名為main.py）中。所有的路徑操作路由、對應的操作函式、模型以及所需的匯入等都被放置在同一個程式碼檔案中。
然而，當應用程式涉及處理多個資源時，情況將變得更加複雜。以一個電子商務應用程式為例，需要對書籍和音樂專輯進行CRUD操作。如果將這兩種產品的所有HTTP操作及其模型都放在同一個檔案中，顯然不是組織應用程式碼的理想方式。
在本章中，我們將學習如何處理具有多個API的更大規模的應用程式。本章將涵蓋以下主題：
- 單一檔案應用程式
- APIRouter
- 掛載應用程式
- 依賴項
- 中介軟體
- CORS
單一檔案應用程式
首先，讓我們從定義一個單一指令碼中的書籍和專輯的所有路徑操作開始。應用程式的結構如清單7-1所示。
清單7-1. 具有兩個API的應用程式
from fastapi import FastAPI
from pydantic import BaseModel
class Book(BaseModel):
id: int
class Album(BaseModel):
id: int
app = FastAPI()
@app.get("/")
async def root():
return {"message": "首頁"}
# 書籍API的路由
@app.get("/books")
async def get_books():
return "透過"
@app.get("/books/{id}")
async def get_book(id: int):
return "透過"
@app.post("/books")
async def add_book(b1: Book):
return "透過"
@app.put("/books/{id}")
async def update_book(id: int):
return "透過"
@app.delete("/books/{id}")
async def delete_book(id: int):
return "透過"
# 專輯API的路由
@app.get("/albums")
async def get_albums():
return "透過"
@app.get("/albums/{id}")
async def get_album(id: int):
return "透過"
@app.post("/albums")
async def add_album(a1: Album):
return "透過"
@app.put("/albums/{id}")
async def update_album(id: int):
return "透過"
@app.delete("/albums/{id}")
async def delete_album(id: int):
return "透過"
內容解密：
此範例展示了一個簡單的FastAPI應用程式，包含兩個資源：書籍和專輯。每個資源都有基本的CRUD操作。然而，隨著模型欄位結構和函式處理邏輯的擴充套件，這種程式碼結構將變得冗長且難以維護。
APIRouter
FastAPI中的APIRouter類別允許我們將路由分組到不同的檔案結構中，使其更易於管理。在大型應用程式中，所有的路由被分成小的APIRouter單元。這些單元隨後被包含在主應用程式中。
需要注意的是，這些較小的單元並不是獨立的應用程式。它們可以被視為大型應用程式的一部分，是迷你FastAPI應用程式。所有APIRouter中的路由都將成為主應用程式檔案的一部分。
讓我們實作這個概念，並重新組織前面的應用程式碼。與書籍資源相關的路由和模型被儲存在books.py檔案中。同樣地，專輯路由和模型被放在albums.py指令碼中。
首先，宣告一個APIRouter類別的物件（位於fastapi模組中），而不是我們通常使用的FastAPI類別（清單7-2）。設定prefix屬性為“/books”，並定義一個在檔案中出現的標籤。
清單7-2. APIRouter類別
from fastapi import APIRouter
books = APIRouter(prefix="/books", tags=["books"])
內容解密：
這裡建立了一個名為books的APIRouter例項，並設定了字首和標籤。這使得所有與書籍相關的路由都將以/books為字首，並且在Swagger檔案中被歸類別在“books”標籤下。
將路由分離到不同的檔案
將書籍相關的路由和模型放在books.py檔案中，如清單7-3所示。
清單7-3. books路由器
from fastapi import APIRouter
from pydantic import BaseModel
books = APIRouter(prefix="/books", tags=["books"])
class Book(BaseModel):
id: int
@books.get("/")
async def get_books():
return "透過"
@books.get("/{id}")
async def get_book(id: int):
return "透過"
@books.post("/")
async def add_book(b1: Book):
return "透過"
@books.put("/{id}")
async def update_book(id: int):
return "透過"
@books.delete("/{id}")
async def delete_book(id: int):
return "透過"
內容解密：
這個檔案中定義了書籍資源的所有操作。注意，這裡使用的是@books.get()等裝飾器，而不是@app.get()，因為books是我們的APIRouter例項。
同樣地，專輯相關的路由和模型被放在albums.py檔案中，如清單7-4所示。
清單7-4. albums路由器
from fastapi import APIRouter
from pydantic import BaseModel
albums = APIRouter(prefix="/albums", tags=["albums"])
class Album(BaseModel):
id: int
@albums.get("/")
async def get_albums():
return "透過"
@albums.get("/{id}")
async def get_album(id: int):
return "透過"
@albums.post("/")
async def add_album(a1: Album):
return "透過"
@albums.put("/{id}")
async def update_album(id: int):
return "透過"
@albums.delete("/{id}")
async def delete_album(id: int):
return "透過"
主應用程式
現在，我們來看看主FastAPI應用程式碼。在這裡宣告應用程式物件。要包含子應用程式，需要匯入books和albums模組。使用這些模組中的APIRouter物件作為app物件的include_router()方法的引數。
清單7-5. 包含路由器的主應用程式
from fastapi import FastAPI
import books, albums
app = FastAPI()
app.include_router(books.books)
app.include_router(albums.albums)
@app.get("/")
async def root():
return {"message": "首頁"}
內容解密：
這裡建立了主FastAPI應用程式，並包含了books和albums路由器。這使得這兩個資源的所有路由都成為主應用程式的一部分。
執行伺服器並開啟Swagger檔案頁面。你會看到路徑操作函式按照在每個APIRouter宣告中定義的字首標籤很好地分組在一起。
路由器包
為了進一步最佳化應用程式結構，可以在應用程式資料夾內建立兩個子資料夾，分別命名為albums和books。將albums.py和books.py檔案移動到相應的子資料夾中。在這兩個子資料夾中放置一個空的__init__.py檔案，以便Python能夠識別它們為包。
將模型宣告從路由器程式碼中移除，並將其放在各自資料夾中的models.py檔案中。檔案結構應該如圖7-4所示。
重構後的專案結構
graph TD
A[專案根目錄] --> B[main.py]
A --> C[books]
A --> D[albums]
C --> E[__init__.py]
C --> F[router.py]
C --> G[models.py]
D --> H[__init__.py]
D --> I[router.py]
D --> J[models.py]
圖表翻譯：
此圖表展示了重構後的專案結構。主應用程式位於main.py中，而books和albums被組織成獨立的包，每個包包含路由器和模型定義。
這種結構使得大型應用程式的管理變得更加容易，每個資源都有自己的模組和模型定義。透過使用APIRouter，我們成功地將不同的API路徑分離到不同的檔案中，提高了程式碼的可維護性和可讀性。
