FastAPI 使用MongoDB

FastAPI也可以使用MongoDB、Cassandra、CouchDB等NoSQL数据库作为REST应用的CRUD操作的后端。在本专题中，我们将看到如何在FastAPI应用程序中使用MongoDB。

MongoDB是一个面向文档的数据库，其中的半结构化文档以JSON等格式存储。文档可以包含许多不同的键值对，或键数对，甚至是嵌套文档。它是一个键值对的集合，类似于Python的字典对象。一个或多个这样的文档被存储在一个集合中。

MongoDB中的Collection相当于关系型数据库中的一个表。然而，MongoDB（就像所有的NoSQL数据库一样）并没有预定义的模式。一个文档类似于基于SQL的关系数据库表中的单行。每个文档可以由数量不等的键值对组成。因此，MongoDB是一个无模式的数据库。

要在FastAPI中使用MongoDB，必须在机器上安装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的网络界面添加一些文件。你可以在MongoDB的Compass GUI前端验证该集合。

FastAPI - 使用MongoDB

为了检索所有书籍的列表，让我们包括以下获取操作函数 - 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"
]

This following GET decorator retrieves a book document corresponding to given ID as path parameter −

@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