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://127.0.0.1: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 响应如下 -

打印页面