接口偶发变慢，MongoDB 监控却没看到明显压力。

代码翻到数据访问层，我第一眼就看见了这个东西：

@app.get("/orders/{order_no}")
async def get_order(order_no: str):
    client = MongoClient(MONGO_URI)
    order = client.shop.orders.find_one({"order_no": order_no})
    client.close()
    return order

FastAPI 用的是异步接口，里面却塞了同步 MongoDB 查询，而且每个请求重新创建一次客户端。数据少的时候看不出来，并发一上来，事件循环被堵住，连接也跟着反复折腾。

这种代码能运行，但不太像准备上线的代码。

现在的新项目，我会直接使用 PyMongo 的异步客户端 AsyncMongoClient。MongoDB 官方已经提供 PyMongo Async API，并建议 FastAPI 这类异步框架使用异步驱动；老项目如果还在用 Motor，也该开始考虑迁移了。

先把连接收住。

# app.py
import os
from contextlib import asynccontextmanager

from fastapi import FastAPI
from pymongo import AsyncMongoClient

@asynccontextmanager
async def lifespan(app: FastAPI):
    client = AsyncMongoClient(
        os.environ["MONGO_URI"],
        serverSelectionTimeoutMS=3000,
        maxPoolSize=80,
        minPoolSize=5,
    )

    await client.admin.command("ping")

    app.state.mongo_client = client
    app.state.mongo_db = client["shop_center"]

    yield

    await client.close()

app = FastAPI(lifespan=lifespan)

数据库客户端跟着应用启动，跟着应用关闭，不要在每个接口里创建。

FastAPI 现在推荐用 lifespan 管理启动和关闭阶段的资源，数据库连接、Redis 客户端、HTTP 连接池都适合放在这里。

连接放进 app.state 后，再通过依赖拿数据库：

from fastapi import Request

def get_db(request: Request):
    return request.app.state.mongo_db

接下来是写入。MongoDB 文档里默认会生成 _id，类型是 ObjectId，这东西直接扔给 FastAPI 返回，通常会撞上 JSON 序列化问题。

我一般不在项目里到处写 str(document["_id"])，先做一个统一转换。

from datetime import datetime, timezone
from typing import Annotated

from fastapi import Depends
from pydantic import BaseModel, Field

class OrderCreate(BaseModel):
    order_no: str = Field(min_length=6, max_length=40)
    user_id: int
    amount: int = Field(gt=0)

def clean_document(document: dict | None) -> dict | None:
    if document is None:
        return None

    document["id"] = str(document.pop("_id"))
    return document

@app.post("/orders")
async def create_order(
    payload: OrderCreate,
    db: Annotated[object, Depends(get_db)],
):
    document = payload.model_dump()
    document.update({
        "status": "pending",
        "created_at": datetime.now(timezone.utc),
    })

    result = await db.orders.insert_one(document)

    return {
        "id": str(result.inserted_id),
        "order_no": payload.order_no,
    }

金额这里我用了整数，存的是最小货币单位。别把浮点数直接当金额存，查数据库的问题还没开始，精度问题先来了。

查询也别直接 find({}) 然后全部转成列表。接口必须限制数量，返回字段也要收紧：

@app.get("/orders")
async def list_orders(
    user_id: int,
    db: Annotated[object, Depends(get_db)],
):
    cursor = (
        db.orders
        .find(
            {"user_id": user_id},
            {
                "order_no": 1,
                "amount": 1,
                "status": 1,
                "created_at": 1,
            },
        )
        .sort("created_at", -1)
        .limit(50)
    )

    rows = await cursor.to_list(length=50)
    return [clean_document(row) for row in rows]

这里最容易被忽略的不是 Python，而是索引。

上面的查询条件是 user_id，排序字段是 created_at。只给 user_id 建单字段索引，数据多了以后，MongoDB 仍然可能需要额外排序。我更倾向直接按真实查询建联合索引：

from pymongo import DESCENDING

async def prepare_indexes(db):
    await db.orders.create_index(
        [
            ("user_id", 1),
            ("created_at", DESCENDING),
        ],
        name="idx_user_created",
    )

    await db.orders.create_index(
        "order_no",
        unique=True,
        name="uk_order_no",
    )

MongoDB 支持异步创建单字段和联合索引，但索引不是越多越好。每多一个索引，写入和更新都要多维护一份。接口拿什么字段查、按什么字段排，再决定怎么建。

更新数据时，也别先查出来、改完再整条覆盖。订单状态变化只改状态字段：

from datetime import datetime, timezone
from fastapi import HTTPException

@app.patch("/orders/{order_no}/paid")
async def mark_paid(
    order_no: str,
    db: Annotated[object, Depends(get_db)],
):
    result = await db.orders.update_one(
        {
            "order_no": order_no,
            "status": "pending",
        },
        {
            "$set": {
                "status": "paid",
                "paid_at": datetime.now(timezone.utc),
            }
        },
    )

    if result.modified_count == 0:
        raise HTTPException(
            status_code=409,
            detail="订单不存在或状态已变化",
        )

    return {"order_no": order_no, "status": "paid"}

查询条件里带上旧状态，这一行很关键。两个请求同时支付，不至于都拿着旧数据往下跑。很多并发问题不需要上来就套事务，先把原子更新用对。

FastAPI 配 MongoDB 并不复杂，真正容易出问题的是：异步接口里偷偷跑同步查询、每个请求创建客户端、ObjectId 到处手工转换、查询不限制数量、索引跟着字段名乱建。

这些地方不收住，接口现在能返回 JSON，后面也照样得返工。

更多相关技术内容咨询欢迎前往并持续关注好学星城论坛了解详情。

想高效系统的学习Python编程语言，推荐大家关注一个微信公众号：Python编程学习圈。每天分享行业资讯、技术干货供大家阅读，关注即可免费领取整套Python入门到进阶的学习资料以及教程，感兴趣的小伙伴赶紧行动起来吧。