FastAPI - 使用 MongoDB

FastAPI 还可以使用 MongoDB、Cassandra、CouchDB 等 NoSQL 数据库作为 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

PyMongo 模块中MongoClient 类的对象是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 对象解析为一个字典，并在测试数据库的 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}

通过访问 http://localhost: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

在这种情况下，服务器响应将是图书集合中所有标题的列表。

[
   "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响应如下 −