Fastapi 简明教程
FastAPI - Using 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
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}
通过访问 [role="bare"]http://localhost:8000/docs ,使用 Swagger UI 的 Web 界面添加几个文档。您可以在 MongoDB 的 Compass GUI 前端中验证集合。
要检索所有书籍的列表,让我们包含以下 get 操作函数 * − get_books() . It will be executed when *"/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 响应如下 −