PyMongo是MongoDB官方的同步Python应用程序驱动程序。如果您想了解如何从Python应用程序连接和使用MongoDB，您就来到了正确的地点。在本PyMongo教程中，我们将使用FastAPI和MongoDB Atlas构建一个简单的CRUD（创建、读取、更新、删除）应用程序。该应用程序将能够创建、读取、更新和删除MongoDB数据库中的文档，并通过REST API公开这些功能。您可以在Github上找到完成的应用程序。
什么是PyMongo？使用Python和MongoDB入门
图书管理CRUD应用程序
我最喜欢的学习新技术的方式是通过构建一些东西。这就是为什么我们将编写最简单但又实用的后端应用程序——一个用于管理图书的CRUD应用程序。CRUD操作将通过REST API提供。该API将具有五个端点
GET /book：列出所有图书GET /book/<id>：通过其ID获取一本书POST /book：创建一本新书PUT /book/<id>：通过其ID更新一本书DELETE /book/<id>：通过其ID删除一本书
为了构建API，我们将使用FastAPI框架。这是一个轻量级、现代且易于使用的框架，用于构建API。它还生成Swagger API文档，我们将用它来测试应用程序。
我们将把图书存储在MongoDB Atlas集群中。MongoDB Atlas是MongoDB的数据库即服务平台。它是基于云的，您可以在几分钟内创建一个免费账户和集群，无需在您的机器上安装任何东西。我们将使用PyMongo连接到集群并查询数据。
项目设置和配置
在我们开始之前，我们将创建一个虚拟Python环境，以将项目从全局安装的Python包中隔离出来。我们将使用与Python安装一起提供的venv包。从终端执行以下命令
python3 -m venv env-pymongo-fastapi-crud
source env-pymongo-fastapi-crud/bin/activate注意：您可能需要使用 python3 可执行文件来运行此命令。这是因为，在某些操作系统上，Python 2 和 3 都已安装。一旦您登录到您的虚拟环境，python 可执行文件将自动使用版本 3。
现在我们已经有了虚拟环境，我们可以安装所需的包。我们将使用 pip——Python 的包安装器，它也包含在您的 Python 安装中。
python -m pip install 'fastapi[all]' 'pymongo[srv]' python-dotenv接下来，我们将为我们的项目创建一个目录，导航到它，并为项目搭建所需的文件。
mkdir pymongo-fastapi-crud
cd pymongo-fastapi-crud
touch main.py routes.py models.py .env注意：我们将使用 shell 命令来创建文件和目录，并在它们之间导航。如果您愿意，您也可以使用图形文件浏览器。
让我们先实现一个简单的根 / 端点，该端点返回一个欢迎消息。在您最喜欢的代码编辑器中打开 main.py 文件，并添加以下内容
pymongo-fastapi-crud/main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
async def root():
return {"message": "Welcome to the PyMongo tutorial!"}保存文件，并使用与 fastapi 包一起安装的 uvicorn 包来运行应用程序。
python -m uvicorn main:app --reload您应该会看到以下响应
INFO: Will watch for changes in these directories: ['/Users/you/pymongo-fastapi-crud']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [2465] using watchgod
INFO: Started server process [2467]
INFO: Waiting for application startup.
INFO: Application startup complete.在浏览器中打开 http://127.0.0.1:8000。您应该看到欢迎消息。
做得好！我们的服务器正在运行。在下一节中，我们将连接到我们的 MongoDB Atlas 集群。
连接到您的 MongoDB Atlas 集群
接下来，我们需要连接到我们之前创建的 MongoDB Atlas 集群。找到您的 连接字符串，并将其添加到 .env 文件中。将 <username> 和 <password> 替换为您的凭据。
pymongo-fastapi-crud/.env
ATLAS_URI=mongodb+srv://<username>:<password>@sandbox.lqlql.mongodb.net/?retryWrites=true&w=majority
DB_NAME=pymongo_tutorial我们将使用 python-dotenv 包从 .env 文件中加载环境变量 ATLAS_URI 和 DB_NAME。然后，我们将在应用程序启动时使用 pymongo 包连接到 Atlas 集群。我们还将添加另一个事件处理程序来关闭应用程序停止时的连接。再次打开 main.py 文件，并替换其内容如下
from fastapi import FastAPI
from dotenv import dotenv_values
from pymongo import MongoClient
config = dotenv_values(".env")
app = FastAPI()
@app.on_event("startup")
def startup_db_client():
app.mongodb_client = MongoClient(config["ATLAS_URI"])
app.database = app.mongodb_client[config["DB_NAME"]]
print("Connected to the MongoDB database!")
@app.on_event("shutdown")
def shutdown_db_client():
app.mongodb_client.close()uvicorn 进程将检测文件更改并重新启动服务器。您应该在终端中看到消息 Connected to the MongoDB database!。
为 API 请求和响应创建模型
MongoDB 具有灵活的模式模型，允许在同一集合中具有不同结构的文档。在实践中，集合中的文档通常具有相同的结构。如果需要，您甚至可以强制执行每个集合的 验证规则。在我们的 PyMongo 教程中，我们不会涵盖数据库验证。相反，我们将在将数据存储到数据库之前确保通过 REST API 传递的数据是有效的。
我们将为 API 请求和响应创建一些模型，并让 FastAPI 为我们做繁重的工作。该框架将负责验证、转换为正确的数据类型，甚至生成 API 文档。打开 models.py 文件并添加以下内容
pymongo-fastapi-crud/models.py
import uuid
from typing import Optional
from pydantic import BaseModel, Field
class Book(BaseModel):
id: str = Field(default_factory=uuid.uuid4, alias="_id")
title: str = Field(...)
author: str = Field(...)
synopsis: str = Field(...)
class Config:
allow_population_by_field_name = True
schema_extra = {
"example": {
"_id": "066de609-b04a-4b30-b46c-32537c7f1f6e",
"title": "Don Quixote",
"author": "Miguel de Cervantes",
"synopsis": "..."
}
}
class BookUpdate(BaseModel):
title: Optional[str]
author: Optional[str]
synopsis: Optional[str]
class Config:
schema_extra = {
"example": {
"title": "Don Quixote",
"author": "Miguel de Cervantes",
"synopsis": "Don Quixote is a Spanish novel by Miguel de Cervantes..."
}
}我们从《pydantic》包扩展了BaseModel，并为我们的模型添加了字段。对于Book模型，我们定义了四个必填字段：id、title、author和synopsis。id字段会自动填充一个UUID（通用唯一标识符）。我们还提供了一个Book模型的示例，它将在API文档中显示。
BookUpdate模型中的字段是可选的。这将使我们能够进行部分更新。在BookUpdate模型中没有id字段，因为我们不希望用户更新id。
现在我们已经定义了模型，让我们实现REST API端点并使用模型来验证数据。
实现REST API端点
现在是时候进入有趣的部分了！让我们为我们的书籍构建REST API端点！我们将在routes.py文件中添加端点实现，并在main.py文件中加载这些端点。我们将在routes.py中初始化一个APIRouter对象。
pymongo-fastapi-crud/routes.py
from fastapi import APIRouter, Body, Request, Response, HTTPException, status
from fastapi.encoders import jsonable_encoder
from typing import List
from models import Book, BookUpdate
router = APIRouter()如您所见，我们正在从fastapi包导入APIRouter。我们将使用此对象来定义REST API的端点。我们还导入了我们之前定义的Book和BookUpdate模型。
POST /book
我们将实现第一个端点，即用于创建新书籍的POST /books端点。在router = APIRouter()行之后添加以下内容
pymongo-fastapi-crud/routes.py
@router.post("/", response_description="Create a new book", status_code=status.HTTP_201_CREATED, response_model=Book)
def create_book(request: Request, book: Book = Body(...)):
book = jsonable_encoder(book)
new_book = request.app.database["books"].insert_one(book)
created_book = request.app.database["books"].find_one(
{"_id": new_book.inserted_id}
)
return created_book该路由是/，因为我们将在所有书籍端点前加上/books前缀。response_description将在API文档中显示。status_code是请求成功返回的HTTP状态码。我们使用Book模型来验证请求体中传递的数据和返回的响应。FastAPI为我们处理验证。在函数体中，我们使用PyMongo的insert_one()方法将新书籍添加到books集合。我们使用find_one()方法从数据库中检索新创建的书籍。您可以在PyMongo文档的集合级别操作部分中了解更多关于insert_one()和find_one()方法的信息。
最后，我们返回创建的书籍。
GET /book
接下来，我们将实现用于返回books集合中所有文档的列表的GET /book端点。将以下内容追加到routes.py文件末尾
pymongo-fastapi-crud/routes.py
@router.get("/", response_description="List all books", response_model=List[Book])
def list_books(request: Request):
books = list(request.app.database["books"].find(limit=100))
return books对于响应模型，我们使用List[Book]类型。这意味着响应将是一个Book对象列表。我们还使用find()方法从数据库中检索不超过100本书。要了解更多关于limit和find()方法的其他参数，请查看专门的PyMongo文档页面。
GET /book/{id}
让我们创建另一个用于通过其id检索单个书籍的GET端点。将以下内容添加到routes.py文件末尾
pymongo-fastapi-crud/routes.py
@router.get("/{id}", response_description="Get a single book by id", response_model=Book)
def find_book(id: str, request: Request):
if (book := request.app.database["books"].find_one({"_id": id})) is not None:
return book
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with ID {id} not found")在这里，我们使用find_one()方法从数据库中检索单个书籍。如果找到书籍，则返回它。如果没有找到书籍，则抛出带有404 Not Found状态码和适当信息的HTTPException。
PUT /book/{id}
可以说，对我们REST API来说最重要的端点是 PUT /book/{id} 端点。这个端点允许我们更新一本书。将实现代码添加到 routes.py 文件的末尾
pymongo-fastapi-crud/routes.py
@router.put("/{id}", response_description="Update a book", response_model=Book)
def update_book(id: str, request: Request, book: BookUpdate = Body(...)):
book = {k: v for k, v in book.dict().items() if v is not None}
if len(book) >= 1:
update_result = request.app.database["books"].update_one(
{"_id": id}, {"$set": book}
)
if update_result.modified_count == 0:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with ID {id} not found")
if (
existing_book := request.app.database["books"].find_one({"_id": id})
) is not None:
return existing_book
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with ID {id} not found")让我们来看一下代码。首先，我们创建了一个对象，我们将用它来更新书籍。然后，如果 book 对象中有任何字段，我们将使用 update_one() 方法在数据库中更新书籍。需要注意的是，我们使用 $set 更新运算符来确保只更新指定的字段，而不是重写整个文档。
接下来，我们检查 update_result 的 modified_count 属性以验证书籍是否已更新。如果是这样，我们将使用 find_one() 方法从数据库中检索更新的书籍并返回它。
如果 book 对象中没有字段，我们只需返回现有的书籍。然而，如果书籍未找到，我们将引发一个带有 404 Not Found 状态码的 HTTPException。
DELETE /book/{id}
我们将实现的最后一个端点是用于通过其 id 删除单一书籍的 DELETE /book/{id} 端点。将以下内容添加到 routes.py 文件的末尾
pymongo-fastapi-crud/routes.py
@router.delete("/{id}", response_description="Delete a book")
def delete_book(id: str, request: Request, response: Response):
delete_result = request.app.database["books"].delete_one({"_id": id})
if delete_result.deleted_count == 1:
response.status_code = status.HTTP_204_NO_CONTENT
return response
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"Book with ID {id} not found")这里唯一值得注意的是，如果书籍已被删除，我们将返回一个 204 No Content 状态码。这是一个成功状态码，表示请求已成功，并且在响应的有效负载正文中没有内容要发送。
注册 /book 端点
最后，我们需要注册 /book 端点。打开 main.py 文件，导入 routes 模块，并注册书籍路由器。您最终的 main.py 文件应该如下所示
pymongo-fastapi-crud/main.py
from fastapi import FastAPI
from dotenv import dotenv_values
from pymongo import MongoClient
from routes import router as book_router
config = dotenv_values(".env")
app = FastAPI()
@app.on_event("startup")
def startup_db_client():
app.mongodb_client = MongoClient(config["ATLAS_URI"])
app.database = app.mongodb_client[config["DB_NAME"]]
@app.on_event("shutdown")
def shutdown_db_client():
app.mongodb_client.close()
app.include_router(book_router, tags=["books"], prefix="/book")探索API文档页面并测试端点
在继续之前，请确保您的 uvicorn 进程仍在运行。如果不是，您可以在终端中用相同的命令启动它
python -m uvicorn main:app --reload在浏览器中导航到 https://127.0.0.1:8000/docs URL。这是FastAPI和Swagger为我们生成的API文档页面！
我们看到我们创建的所有端点，我们甚至可以直接从该页面发送请求！打开 POST 选项卡，点击 Try it out 按钮。您应该看到一个预先填充了示例书籍的请求体。点击 Execute 发送请求。您应该看到一个包含我们创建的书籍的成功响应。您可以从响应中获取书籍的id，并在其他端点中使用它——GET /book/{id}、PUT /book/{id} 或 DELETE /book/{id}。
但如果我们尝试创建相同的书籍两次呢？我们将得到一个 500 Internal Server Error 响应。如果我们检查服务器进程运行的终端，我们应该看到一个包含以下内容的错误消息
pymongo.errors.DuplicateKeyError: E11000 duplicate key error collection: pymongo_tutorial.books index: _id我们收到了一个 DuplicateKeyError，因为我们尝试插入具有相同 _id 字段的书籍两次。每个集合都有一个 _id 字段，这是MongoDB为每个集合创建的唯一索引。我们不能有两本具有相同 _id 的书籍。实际上，问题在于我们没有在代码中处理这个错误。错误“冒泡”到服务器，并以 500 Internal Server Error 响应。作为一个练习，您可以考虑向客户端发送一个合适的响应来处理这个错误。
您还可以测试我们创建的验证规则。例如，尝试从请求体中删除必需的 title 字段，然后点击 执行。您应该会看到一个错误消息，说明 title 字段是必需的。
生成的API文档页面非常适合尝试不同的场景并查看API的行为。祝您探索我们构建的API时玩得开心！
结论
在本教程中，我们看到了如何使用FastAPI和PyMongo创建一个简单的CRUD应用程序，PyMongo是MongoDB的官方同步Python应用程序驱动程序。我们还看到了如何快速设置免费的MongoDB Atlas集群并连接到它。MongoDB Atlas不仅仅是MongoDB云数据库。例如，您可以使用 Atlas Search 将您的API扩展为提供全文搜索。所有这些服务都在MongoDB Atlas中可用。如果您想尝试它们，创建您的 免费账户。
常见问题解答
您可以使用PyMongo做什么？
如何下载和安装PyMongo？
pip 安装它：python -m pip install pymongo[srv]。如何使用PyMongo？
我应该使用PyMongo还是MongoEngine？
PyMongo是MongoDB的官方Python驱动程序，而MongoEngine是一个使用PyMongo内部实现的ORM（对象关系映射器）。PyMongo由MongoDB官方支持和推荐。有关两者的区别，请参阅 专门的文章。