公告MongoDB 8.0 隆重推出，这是有史以来最快的MongoDB！了解详情 >

公告Voyage AI 与 MongoDB 携手合作，致力于在 Atlas 上提供更准确和更值得信赖的 AI 应用。了解更多 >

什么是 PyMongo？Python 和 MongoDB 入门

PyMongo 是用于同步 Python 应用程序的官方 MongoDB 驱动程序。如果您想学习如何通过 Python 应用程序连接和使用 MongoDB，那么您来对地方了。在本 PyMongo 教程中，我们将使用 FastAPI 和 MongoDB Atlas 构建一个简单的 CRUD（创建、读取、更新、删除）应用程序。该应用程序将能够在 MongoDB 数据库中创建、读取、更新和删除文档，并通过 REST API 公开这些功能。您可以在 Github 上找到已完成的应用程序。

什么是 PyMongo？Python 和 MongoDB 入门

图书管理 CRUD 应用程序
要求
项目设置和配置
连接到您的 MongoDB Atlas 集群
创建 API 请求和响应的模型
实现 REST API 终结点
浏览 API 文档页面并测试 API 终结点
结论

图书管理 CRUD 应用程序

我最喜欢通过构建项目来学习新技术。这就是为什么我们将编写一个最简单但最实用的后端应用程序 — 一个用于管理图书的 CRUD 应用。CRUD 操作将通过 REST API 提供。该 API 将有五个终结点：

Code SnippetGET /book: to list all books
GET /book/<id>: to get a book by its ID
POST /book: to create a new book
PUT /book/<id>: to update a book by its ID
DELETE /book/<id>: to delete a book by its ID

为了构建 API，我们将使用 FastAPI 框架。它是一个轻量级、现代且易于使用的 API 构建框架。它还会生成一个 Swagger API 文档，我们在测试应用程序时会使用该文档。

我们会将图书存储在 MongoDB Atlas 集群中。MongoDB Atlas 是 MongoDB 的数据库即服务平台。该平台基于云，您可以在几分钟内创建免费帐户和集群，整个过程无需在本地设备安装任何组件。我们将使用 PyMongo 连接到集群并查询数据。

要求

完成的项目可以在 Github 上找到。您还可以按照分步说明从头开始构建项目。为此，您需要执行以下操作：

Python 3.6+ — 您可以从 Python 网站安装它。
MongoDB Atlas 集群——您可以根据 Atlas 入门指南创建永久免费集群。完成指南中的步骤并找到您的连接字符串——您将需要该连接字符串来从应用程序连接到数据库。

项目设置和配置

在开始之前，我们将创建一个虚拟的 Python 环境，以便将项目与全局安装的 Python 包隔离开来。我们将使用 Python 安装中自带的 `venv` 包。在终端中执行以下命令：

Code Snippetpython3 -m venv env-pymongo-fastapi-crud
source env-pymongo-fastapi-crud/bin/activate

注意：您可能需要使用 python3 可执行文件运行此命令。这是因为在某些操作系统上同时安装了 Python 2 和 Python 3。登录虚拟环境后， python 可执行文件将自动使用版本 3。

现在我们已经有了虚拟环境，可以安装所需的包。我们将使用 pip——Python 的包安装程序，它也包含在 Python 的安装中：

Code Snippet

接下来，我们将为项目创建一个目录，导航到该目录，并构建项目所需的文件。

Code Snippetmkdir pymongo-fastapi-crud
cd pymongo-fastapi-crud
touch main.py routes.py models.py .env

注意：我们将使用 shell 命令来创建文件和目录，并进行导航。如果您愿意，也可以使用图形文件资源管理器。

让我们先实现一个简单的根 / 终结点，该终结点返回一条欢迎消息。在您常用的代码编辑器中打开 main.py 文件，并添加以下内容：

pymongo-fastapi-crud/main.py

Code Snippetfrom fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Welcome to the PyMongo tutorial!"}

保存文件，并使用与 fastapi 包一起安装的 uvicorn 包运行应用程序。

Code Snippet

您应看到以下响应：

Code SnippetINFO:     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 文件。将用户名和密码替换为您的凭证。

pymongo-fastapi-增删改查/.env

Code SnippetATLAS_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 文件，将其内容替换为以下内容：

Code Snippetfrom 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 进程将检测文件更改并重启服务器。您应该会在终端看到”已连接到 MongoDB 数据库！“这条消息。

创建 API 请求和响应的模型

MongoDB 具有灵活的架构模型，允许在同一个集合中包含不同结构的文档。实际上，集合中的文档通常具有相同的结构。如果需要，您甚至可以为每个集合实施验证规则。我们的 PyMongo 教程中不会涵盖数据库验证。相反，我们将确保通过 REST API 传输的数据在存储到数据库之前是有效的。

我们将为 API 请求和响应创建几个模型，并让 FastAPI 为我们承担繁重的工作。该框架将负责验证、转换为正确的数据类型，甚至生成 API 文档。打开 models.py 文件并添加以下内容：

pymongo-fastapi-crud/models.py

Code Snippetimport 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

Code Snippetfrom 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 /books`终结点。在 router = APIRouter() 行后添加以下内容：

pymongo-fastapi-crud/routes.py

Code Snippet@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

接下来，我们将实现`GET /book`终结点，以返回`books`集合中所有文档的列表。将以下内容附加到`routes.py`文件：

pymongo-fastapi-crud/routes.py

Code Snippet@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}

让我们创建另一个`GET`终结点，通过其`id`检索单本图书。将以下内容附加到`routes.py`文件：

pymongo-fastapi-crud/routes.py

Code Snippet@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

Code Snippet@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}

我们将实现的最后一个终结点是`DELETE /book/{id}`终结点，用于通过其`id`删除单本图书。将以下内容附加到`routes.py`文件：

pymongo-fastapi-crud/routes.py

Code Snippet@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`终结点。打开 main.py 文件，导入`routes`模块，并注册图书路由器。您的 main.py 文件最终版本应该是这样的：

pymongo-fastapi-crud/main.py

Code Snippetfrom 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`进程仍在运行。如果没有运行，您可以在终端运行相同的命令：

Code Snippet

在浏览器中导航到 http://localhost: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`响应。如果检查运行服务器进程的终端，应该会看到一条包含以下内容的错误消息：

Code Snippetpymongo.errors.DuplicateKeyError: E11000 duplicate key error collection: pymongo_tutorial.books index: _id

我们收到了 DuplicateKeyError，因为试图插入具有相同 _id 字段的图书两次。`_id` 字段是 MongoDB 为每个集合创建的唯一索引。我们不能有两本图书具有相同的`_id`。这里的实际问题是我们没有在代码中处理这个错误。错误会“冒泡”，服务器返回“500 内部服务器错误”。作为一项练习，您可以设想将一个适当的响应发送回客户端并处理此错误。

您还可以测试我们创建的验证规则。例如，尝试从请求正文中删除必需的“title”字段，然后单击“Execute”。您应该会看到一条错误消息，提示“title”字段是必需的。

生成的 API 文档页面对于尝试不同场景和观察 API 的行为非常有用。请享受探索我们构建的 API 的乐趣！

结论

在本教程中，我们学习了如何使用 FastAPI 和 PyMongo（用于同步 Python 应用程序的 MongoDB 官方驱动程序）创建一个简单的 CRUD 应用程序。我们还了解了如何快速设置并连接到免费的 MongoDB Atlas 集群。MongoDB Atlas 不仅仅是一个 MongoDB 云数据库。例如，您可以轻松扩展 API，以使用 Atlas Search 提供 Atlas 全文搜索功能。所有这些服务都可以在 MongoDB Atlas 中使用。如果您想尝试一下，请创建免费账户。

常见问题解答

PyMongo 是用于同步 Python 应用程序的官方 MongoDB 驱动程序。您可以连接到 MongoDB 实例并查询数据。它是用于构建与 MongoDB 交互的应用程序的绝佳工具。

PyMongo 是一个 Python 包。您可以使用 `pip` 安装它：`python -m pip install pymongo[srv]`。

一旦安装了 PyMongo，您就可以将其导入到您的应用程序中，连接到 MongoDB 实例，并开始查询数据。

PyMongo 是官方的 MongoDB Python 驱动程序，而 MongoEngine 是一个使用 PyMongo 的 ORM（对象关系映射器）。PyMongo 是由 MongoDB 官方支持和推荐的工具。请阅读专门文章，了解关于两者之间差异的更多信息。

立即开始使用 Atlas

只需几秒就能上手。我们的免费集群附带 512 MB 的存储空间，因此您可以使用示例数据进行试用并了解我们的平台。

免费试用联系销售团队

开始使用：

全球超过 125 个地区
示例数据集
永远在线身份验证
端到端加密

命令行工具

什么是 PyMongo？Python 和 MongoDB 入门

图书管理 CRUD 应用程序

要求

项目设置和配置

连接到您的 MongoDB Atlas 集群

创建 API 请求和响应的模型

实现 REST API 终结点

帖子/图书

GET /book

GET /book/{id}

PUT /book/{id}

DELETE /book/{id}

注册该项

浏览 API 文档页面并测试终结点

结论

常见问题解答

您可以用 PyMongo 做什么？

我如何下载和安装 PyMongo？

您如何使用 PyMongo？

我应该使用 PyMongo 还是 MongoEngine？

立即开始使用 Atlas