首页 / FastAPI 入门教程 / NoSQL 对接 MongoDB

← 上一篇 下一篇 →
FastAPI 入门教程

NoSQL 对接 MongoDB

本教程共 50 篇 · 第 37 篇 · 更新于 2026-08-12 · 约 9 分钟阅读

FastAPIFastAPI 入门教程MongoDBNoSQLmotor

本节目标：理解文档型数据库适合什么场景，学会用异步驱动 motor 连接 MongoDB，把 Pydantic 模型与 MongoDB 文档互相转换，并写出完整的增删改查接口。

前面六章都在用关系型数据库。但有些数据天生「没有固定形态」，比如用户行为日志、商品的多变属性、聊天记录。这类场景用文档型数据库 MongoDB 更顺手。本章讲它怎么和 FastAPI 配合。

37-1 什么时候该用 NoSQL

关系型数据库要求「先建表、定好列」，结构稳定、关联复杂时最香。但遇到下面情况，文档库更合适：

  • 数据结构经常变：今天加个字段，明天又加一个，不想每次都改表。
  • 数据之间是「嵌套」而非「关联」：一篇文章带一堆评论，直接嵌进一个文档最自然。
  • 需要海量水平扩展：MongoDB 天生支持分片，横向加机器很方便。
Note

不是说 NoSQL 比 SQL 好，而是「看场景」。账户、订单这类强一致、多关联的数据，仍该用关系型；灵活、松散、要扩展的数据，才考虑 MongoDB。

37-2 MongoDB 的文档模型

MongoDB 不存表，存「集合（collection）」和「文档（document）」。文档就是一段 BSON（类 JSON）数据，每个文档自带一个唯一标识 _id。

用「书籍」举例，一条文档长这样：

{
  "_id": ObjectId("..."),
  "title": "FastAPI 实战",
  "author": "码上学",
  "price": 39.9,
  "tags": ["Python", "Web"]
}

注意它和关系型的关键差别：没有固定列，不同文档的字段可以不一样；还能直接嵌套数组（tags）。这正好对应 Pydantic 的模型结构，转换非常顺。

37-3 安装与连接：motor 异步驱动

MongoDB 官方驱动 pymongo 是同步的，会阻塞事件循环。FastAPI 要用异步的 motor：

pip install motor

motor 提供 AsyncIOMotorClient，用起来和同步客户端几乎一样，但方法都能 await：

import motor.motor_asyncio

# 连到本地 MongoDB，默认端口 27017
client = motor.motor_asyncio.AsyncIOMotorClient("mongodb://localhost:27017")

# 选数据库（不存在也没关系，写入时自动建）
db = client["bookstore"]

# 选集合（类似关系库的「表」）
books_collection = db["books"]
Tip

MongoDB 是「惰性」的：数据库和集合不用提前建，你第一次插入数据时就自动创建了。这点和 SQLite 类似，开箱即用。

37-4 Pydantic 模型与文档互转

Pydantic 负责接口层的数据校验。我们定义一个 Book 模型，注意 id 是可选的字符串——MongoDB 的 _id 是 ObjectId，转成字符串再交给前端更友好。

from pydantic import BaseModel, ConfigDict

class Book(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: str | None = None
    title: str
    author: str
    price: float
    tags: list[str] = []

插入时，把 Pydantic 转成普通字典，但去掉 id（让 MongoDB 自动生成 _id）：

data = book.model_dump(exclude={"id"})   # {'title':..., 'author':..., ...}

读出时，MongoDB 返回带 _id 的字典，需要把 _id 变成字符串、改名 id，再交给 Pydantic：

doc = await books_collection.find_one({"_id": obj_id})
doc["id"] = str(doc.pop("_id"))   # ObjectId -> 字符串
return Book.model_validate(doc)
Note

model_validate(dict) 是 Pydantic v2 的写法，替代 v1 的 parse_obj。from_attributes=True 让模型也能从对象属性读取（虽本文档用字典，保留它方便以后对接 ORM 对象）。

37-5 增 Create：插入文档

用 insert_one 插入一条，inserted_id 是 MongoDB 生成的 _id：

from bson import ObjectId
from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.post("/books/", response_model=Book)
async def create_book(book: Book):
    data = book.model_dump(exclude={"id"})
    result = await books_collection.insert_one(data)
    # 把生成的 id 补回，返回给客户端
    new_book = await books_collection.find_one({"_id": result.inserted_id})
    new_book["id"] = str(new_book.pop("_id"))
    return Book.model_validate(new_book)
Tip

insert_one 接收一个纯字典，不能直接塞 Pydantic 对象。先用 model_dump() 转字典，这是文档库和 ORM 最大的使用差异。

37-6 查 Read：查询与过滤

查全部用 find()，再 .to_list(length=...) 取成列表：

@app.get("/books/", response_model=list[Book])
async def list_books(skip: int = 0, limit: int = 20):
    cursor = books_collection.find().skip(skip).limit(limit)
    docs = await cursor.to_list(length=limit)
    return [_to_book(d) for d in docs]

def _to_book(doc):
    doc["id"] = str(doc.pop("_id"))
    return Book.model_validate(doc)

按条件查，比如按作者过滤：

@app.get("/books/{book_id}", response_model=Book)
async def get_book(book_id: str):
    if not ObjectId.is_valid(book_id):
        raise HTTPException(status_code=400, detail="id 格式错误")
    doc = await books_collection.find_one({"_id": ObjectId(book_id)})
    if doc is None:
        raise HTTPException(status_code=404, detail="书籍不存在")
    doc["id"] = str(doc.pop("_id"))
    return Book.model_validate(doc)
Note

MongoDB 的 _id 是 ObjectId 类型，前端传来的字符串要先 ObjectId(...) 转回，且要先校验合法性，否则会抛异常。

37-7 改 Update：局部更新

用 update_one 更新。$set 表示「只改这些字段」，其余保持不变：

@app.patch("/books/{book_id}", response_model=Book)
async def update_book(book_id: str, book: Book):
    if not ObjectId.is_valid(book_id):
        raise HTTPException(status_code=400, detail="id 格式错误")
    # 只取客户端传来的字段
    data = book.model_dump(exclude={"id"}, exclude_unset=True)
    if not data:
        raise HTTPException(status_code=400, detail="没有可更新字段")
    await books_collection.update_one(
        {"_id": ObjectId(book_id)},
        {"$set": data},
    )
    return await get_book(book_id)

exclude_unset=True 只更新客户端真正发来的字段，没传的保持原值。

37-8 删 Delete：删除文档

用 delete_one 按 _id 删除，返回结果里的 deleted_count 告诉你删了几条：

@app.delete("/books/{book_id}")
async def delete_book(book_id: str):
    if not ObjectId.is_valid(book_id):
        raise HTTPException(status_code=400, detail="id 格式错误")
    result = await books_collection.delete_one({"_id": ObjectId(book_id)})
    if result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="书籍不存在")
    return {"ok": True}

37-9 与关系型写法的对照

把两种思路放一起看，更容易记住：

  • 连接：SQLAlchemy 用 create_engine；MongoDB 用 AsyncIOMotorClient。
  • 表/集合：SQLAlchemy 要 Base.metadata.create_all 建表；MongoDB 写入即建集合。
  • 增：SQLAlchemy session.add + commit；MongoDB insert_one。
  • 查：SQLAlchemy select + execute；MongoDB find + to_list。
  • 模型：SQLAlchemy 用 Mapped 定义列；MongoDB 直接用 Pydantic 描述文档形状。
Tip

同一个 FastAPI 项目里，关系型和文档型可以共存：账户用 PostgreSQL，日志用 MongoDB，各取所长。只要分别管好连接即可。

37-10 索引与嵌套查询

频繁按某字段查，就该给集合建索引，否则每次都是全表扫描，数据一大就慢：

# 给 author 字段建升序索引，重复建不会报错
await books_collection.create_index([("author", 1)])

查嵌套数组用起来很直观。比如找带 “Python” 标签的书，MongoDB 会匹配数组里任一元素：

docs = await books_collection.find({"tags": "Python"}).to_list(length=50)

要统计总数用 count_documents：

total = await books_collection.count_documents({"author": "码上学"})

嵌套文档也能按内层字段查，比如文章里存了 {"rating": {"score": 9}}，可以用 {"rating.score": 9} 当条件。文档库的灵活就体现在这里：字段想嵌套多深都行，不用像关系库那样先拆表、再 join。

这份灵活也有代价，值得提前想清楚。MongoDB 默认不校验文档结构，同一个集合里完全可以并存字段不一致的新旧文档。数据库不会拦你，出问题的地方会推迟到读取时——旧文档缺字段，Pydantic 校验就会失败。所以用文档库时，「保证数据形状一致」这件事的责任落到了应用层，也就是你定义的 Pydantic 模型上。给可能缺失的字段配上默认值，是一个成本很低但很有效的防护习惯。

Tip

索引能加速查询，但会拖慢写入、占用空间。只在「经常用来过滤或排序」的字段上建，别给每个字段都加。

到这一章，你已经掌握了 FastAPI 对接两类数据库的能力：关系型（SQLAlchemy，第 32–36 章）和文档型（MongoDB + motor，本章）。选哪个，取决于你的数据结构与扩展需求。数据库这部分就告一段落，接下来还有异步、中间件、测试、部署等主题。

上一篇
异步数据库
下一篇
异步支持 async/await