FastAPI Backend Development Bootcamp · Урок

Асинхронный доступ к MongoDB с Motor

Подключайте FastAPI к MongoDB с помощью асинхронного драйвера Motor и управляйте жизненным циклом подключений в периоде работы приложения.

«Асинхронный доступ к MongoDB с Motor» — бесплатный урок FastAPI Backend Development Bootcamp на CoddyKit. Это урок 1 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения 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.

Часто задаваемые вопросы

Урок «Асинхронный доступ к MongoDB с Motor» бесплатный?

Да — полный текст урока «Асинхронный доступ к MongoDB с Motor» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс FastAPI Backend Development Bootcamp, подпишись на CoddyKit PRO. Курс FastAPI Backend Development Bootcamp содержит 4 уроков всего.

Чему я научусь в уроке «Асинхронный доступ к MongoDB с Motor»?

Подключайте FastAPI к MongoDB с помощью асинхронного драйвера Motor и управляйте жизненным циклом подключений в периоде работы приложения. Ты практикуешь FastAPI Backend Development Bootcamp с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать FastAPI Backend Development Bootcamp?

Предыдущий опыт не требуется. FastAPI Backend Development Bootcamp на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 1 из 4.

Сколько времени занимает урок «Асинхронный доступ к MongoDB с Motor»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке FastAPI Backend Development Bootcamp?

Да. Каждый урок FastAPI Backend Development Bootcamp включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

← Назад к FastAPI Backend Development Bootcamp