0Pricing
FastAPI Backend Development Bootcamp · 课时

使用 Motor 异步访问 MongoDB

使用 Motor 异步驱动将 FastAPI 连接到 MongoDB，并在应用生命周期中管理连接生命周期。

使用 Motor 异步访问 MongoDB 是 CoddyKit 上的免费 FastAPI Backend Development Bootcamp 课时。 这是第 1 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 FastAPI Backend Development Bootcamp 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

本课时的部分内容尚未翻译，以英文显示。

Why Motor for Async MongoDB

FastAPI is an async framework. If you talk to MongoDB with a blocking driver like pymongo, every database call freezes the event loop and kills concurrency.

Motor is MongoDB's official async driver. It wraps PyMongo and exposes coroutine-based methods you can await, so the event loop stays free to handle other requests while a query is in flight.

  • motor.motor_asyncio.AsyncIOMotorClient — the async client.
  • Every I/O call (find_one, insert_one, ...) returns an awaitable.
  • Higher-level ODMs like Beanie are built on top of Motor.

Installing the Driver

Install Motor and its peer dependency. Motor pulls in a compatible PyMongo automatically.

  • motor — async driver.
  • fastapi and uvicorn — the web layer.

For Beanie integration later you would also add beanie, but Motor alone is enough to read and write documents directly.

pip install motor fastapi uvicorn
# motor brings in a compatible pymongo wheel
# verify the install
python -c "import motor; print(motor.version)"

Creating an Async Client

You create a single AsyncIOMotorClient for the whole application. The client manages an internal connection pool, so you should never create a new client per request.

Indexing into the client gives you a database, and indexing into that gives you a collection. None of this opens a socket yet — connections are established lazily on the first real operation.

from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient("mongodb://localhost:27017")
database = client["shop"]
products = database["products"]

# Equivalent attribute-style access
# database = client.shop
# products = database.products

Awaiting Your First Query

Because every Motor I/O method is a coroutine, you must await it inside an async function. Forgetting await returns an unresolved coroutine object, not your data.

insert_one returns an InsertOneResult whose inserted_id is the generated ObjectId. find_one returns the matching document as a plain dict, or None.

import asyncio
from motor.motor_asyncio import AsyncIOMotorClient

async def main():
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    products = client["shop"]["products"]

    result = await products.insert_one({"name": "Keyboard", "price": 49})
    print("inserted id:", result.inserted_id)

    doc = await products.find_one({"name": "Keyboard"})
    print(doc)

asyncio.run(main())

The Connection Lifecycle Problem

Where should the client live? Options that look tempting but are wrong:

  • A new client inside each route — exhausts connections and is slow.
  • A module-level client created at import time — connects before the app is ready and is hard to close cleanly.

The right place is the app's lifespan: open the client when the server starts, store it, and close it when the server shuts down. This guarantees one pooled client per process and a clean teardown.

The Lifespan Context Manager

Modern FastAPI uses an async context manager passed as lifespan. Code before yield runs on startup; code after yield runs on shutdown.

Store shared resources on app.state so any route can reach them. Calling client.close() on shutdown returns pooled sockets to the OS gracefully.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.mongo = AsyncIOMotorClient("mongodb://localhost:27017")
    app.state.db = app.state.mongo["shop"]
    yield
    app.state.mongo.close()

app = FastAPI(lifespan=lifespan)

Verifying the Connection on Startup

The client connects lazily, so a wrong host won't fail until the first query. To fail fast at boot, send a lightweight ping command during startup.

If the ping raises, the server crashes immediately with a clear error instead of silently serving 500s later.

from contextlib import asynccontextmanager
from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

@asynccontextmanager
async def lifespan(app: FastAPI):
    client = AsyncIOMotorClient("mongodb://localhost:27017")
    await client.admin.command("ping")  # raises if unreachable
    app.state.db = client["shop"]
    app.state.mongo = client
    yield
    client.close()

app = FastAPI(lifespan=lifespan)

Injecting the Database into Routes

Reaching into request.app.state directly works but couples routes to global state. A cleaner pattern is a small dependency that returns the database handle.

This keeps routes testable — in tests you can override the dependency to point at a throwaway database.

from fastapi import Depends, Request
from motor.motor_asyncio import AsyncIOMotorDatabase

def get_db(request: Request) -> AsyncIOMotorDatabase:
    return request.app.state.db

@app.get("/products/{name}")
async def get_product(name: str, db: AsyncIOMotorDatabase = Depends(get_db)):
    doc = await db["products"].find_one({"name": name})
    return doc or {"error": "not found"}

Serializing the ObjectId

MongoDB documents carry an _id field of type ObjectId, which is not JSON serializable. Returning a raw document from a route triggers a serialization error.

Convert _id to a string before returning, or map it into a Pydantic model. A simple helper keeps your routes clean.

def serialize(doc: dict) -> dict:
    if doc and "_id" in doc:
        doc["id"] = str(doc["_id"])
        del doc["_id"]
    return doc

# Usage inside a route:
# raw = await db["products"].find_one({"name": name})
# return serialize(raw)

print(serialize({"_id": "507f1f77bcf86cd799439011", "name": "Mouse"}))

Iterating Cursors Asynchronously

find() returns an async cursor, not a list. You consume it with async for, or materialize it with to_list().

  • await cursor.to_list(length=100) — load up to 100 docs at once.
  • async for doc in cursor: — stream documents one at a time, ideal for large result sets.
from fastapi import Depends
from motor.motor_asyncio import AsyncIOMotorDatabase

@app.get("/products")
async def list_products(db: AsyncIOMotorDatabase = Depends(get_db)):
    cursor = db["products"].find({"price": {"$lt": 100}})
    return await cursor.to_list(length=50)

# Streaming alternative:
# async for doc in cursor:
#     process(doc)

Configuring the Pool and Timeouts

The client constructor accepts tuning options. Read them from environment variables so the same code works across dev and production.

  • maxPoolSize — cap on concurrent connections.
  • serverSelectionTimeoutMS — how long to wait before declaring the server unreachable.

Loading the URI from the environment also keeps credentials out of source control.

import os
from motor.motor_asyncio import AsyncIOMotorClient

def make_client() -> AsyncIOMotorClient:
    uri = os.environ.get("MONGODB_URI", "mongodb://localhost:27017")
    return AsyncIOMotorClient(
        uri,
        maxPoolSize=20,
        serverSelectionTimeoutMS=5000,
    )

Quick Check

Where should the AsyncIOMotorClient be created and destroyed in a FastAPI app?

Recap

You connected FastAPI to MongoDB with the async Motor driver:

  • Motor exposes awaitable methods so MongoDB I/O never blocks the event loop.
  • Create one AsyncIOMotorClient per process — it owns a connection pool.
  • Open and close the client in the lifespan context manager, and ping on startup to fail fast.
  • Expose the database through a Depends dependency for clean, testable routes.
  • Convert _id (an ObjectId) to a string before returning JSON.
  • Consume find() cursors with to_list() or async for, and tune maxPoolSize and timeouts from environment variables.

This Motor foundation is exactly what Beanie builds on next.

常见问题解答

「使用 Motor 异步访问 MongoDB」课时是免费的吗？

是的 — 「使用 Motor 异步访问 MongoDB」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 FastAPI Backend Development Bootcamp 课程的其余内容，请升级到 CoddyKit PRO。 FastAPI Backend Development Bootcamp 课程共包含 4 节课。

「使用 Motor 异步访问 MongoDB」这节课中我会学到什么？

使用 Motor 异步驱动将 FastAPI 连接到 MongoDB，并在应用生命周期中管理连接生命周期。 你通过在浏览器中直接运行的动手代码来练习 FastAPI Backend Development Bootcamp，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 FastAPI Backend Development Bootcamp 需要有经验吗？

无需任何先前经验。CoddyKit 上的 FastAPI Backend Development Bootcamp 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 1 节课，共 4 节。

「使用 Motor 异步访问 MongoDB」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 FastAPI Backend Development Bootcamp 课中编写并运行代码吗？

能。每节 FastAPI Backend Development Bootcamp 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

  1. 使用 Motor 异步访问 MongoDB
  2. 使用 Beanie ODM 建模文档
  3. 聚合流水线与复杂查询
  4. 模式演进与文档迁移
← 返回 FastAPI Backend Development Bootcamp