FastAPI - 使用 MongoDB

FastAPI 也可以使用 NoSQL 資料庫，例如 MongoDB、Cassandra、CouchDB 等作為 REST 應用的 CRUD 操作的後端。在本主題中，我們將瞭解如何在 FastAPI 應用程式中使用 MongoDB。

MongoDB 是一個面向文件的資料庫，其中半結構化文件以 JSON 等格式儲存。文件可以包含許多不同的鍵值對、鍵陣列對，甚至巢狀文件。它是一組鍵值對，類似於 Python 字典物件。一個或多個這樣的文件儲存在一個集合中。

MongoDB 中的集合相當於關係資料庫中的表。但是，MongoDB（以及所有 NoSQL 資料庫）沒有預定義的模式。文件類似於 SQL 基於關係資料庫的表中的一行。每個文件可能包含可變數量的鍵值對。因此，MongoDB 是一個無模式資料庫。

要將 MongoDB 與 FastAPI 一起使用，必須在機器上安裝 MongoDB 伺服器。我們還需要安裝 **PyMongo**，這是 MongoDB 的官方 Python 驅動程式。

pip3 install pymongo

在透過 Python 和 FastAPI 程式碼與 MongoDB 資料庫互動之前，請確保 MongoDB 正在執行，方法是發出以下命令（假設 MongoDB 伺服器安裝在 e:\mongodb 資料夾中）。

E:\mongodb\bin>mongod
..
waiting for connections on port 27017

**MongoClient** 類在 PyMongo 模組中的物件是 Python 與 MongoDB 伺服器互動使用的控制代碼。

from pymongo import MongoClient
client=MongoClient()

我們將 Book 定義為 BaseModel 類以填充請求體（與 SQLite 示例中使用的相同）

from pydantic import BaseModel
from typing import List
class Book(BaseModel):
   bookID: int
   title: str
   author:str
   publisher: str

設定 FastAPI 應用程式物件 -

from fastapi import FastAPI, status
app = FastAPI()

POST 操作裝飾器將 **"/add_new"** 作為 URL 路由，並執行 **add_book()** 函式。它將 Book BaseModel 物件解析為字典，並在 test 資料庫的 BOOK_COLLECTION 中新增一個文件。

@app.post("/add_new", status_code=status.HTTP_201_CREATED)
def add_book(b1: Book):
   """Post a new message to the specified channel."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      result = book_collection.insert_one(b1.dict())
      ack = result.acknowledged
      return {"insertion": ack}

透過訪問 https://:8000/docs 使用 Swagger UI 的 Web 介面新增一些文件。您可以在 MongoDB 的 Compass GUI 前端驗證集合。

要檢索所有書籍的列表，讓我們包含以下 get 操作函式 **- get_books()**。當訪問 **"/books"** URL 路由時，將執行此函式。

@app.get("/books", response_model=List[str])
def get_books():
   """Get all books in list form."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      booklist = book_collection.distinct("title")
      return booklist

在這種情況下，伺服器響應將是 books 集合中所有標題的列表。

[
   "Computer Fundamentals",
   "Python Cookbook",
   "Let Us Python"
]

以下 GET 裝飾器檢索與給定 ID 作為路徑引數對應的書籍文件 -

@app.get("/books/{id}", response_model=Book)
def get_book(id: int):
   """Get all messages for the specified channel."""
   with MongoClient() as client:
      book_collection = client[DB][BOOK_COLLECTION]
      b1 = book_collection.find_one({"bookID": id})
      return b1

Swagger UI 文件頁面顯示以下介面 -

執行上述函式時，伺服器的 JSON 響應如下 -

列印頁面