Async MongoDB in Python with Motor and FastAPI

FastAPI runs your endpoints on an event loop. That's what lets a single worker juggle hundreds of concurrent requests: while one request waits on the network, the loop moves on to the next. But the whole model depends on every slow operation being awaitable. Drop a blocking database call into an async def endpoint and you freeze the loop for every request on that worker until the query comes back.

For years, the answer for MongoDB was Motor, an async wrapper around PyMongo. It worked well, and a huge amount of FastAPI code in the wild still uses it. The landscape has changed, though. PyMongo now ships its own native async API, AsyncMongoClient, and Motor has been deprecated in its favor. If you're starting a project today, you should use PyMongo's async client. If you have a Motor codebase, you should plan a migration.

This guide covers how to build a FastAPI service on async MongoDB: managing the client with lifespan events, modeling documents with Pydantic v2, writing CRUD endpoints, handling ObjectId, indexes, transactions, and testing. Along the way it shows the Motor equivalents so you can read older code and move it forward.

Motor, PyMongo Async, and Where Things Stand

Motor was built by wrapping synchronous PyMongo and running its blocking calls on a thread pool, then exposing coroutines on top. That design carried overhead: every operation paid for a thread hop, and Motor always lagged a little behind PyMongo features.

Starting with PyMongo 4.9, the driver includes an async implementation written directly against asyncio. It went GA in PyMongo 4.13, and MongoDB announced that Motor is deprecated, with only critical bug fixes until it reaches end of life (scheduled for 2026). MongoDB's own benchmarks show the native client is faster, and it's where new features land first.

Here's the practical summary:

	Motor	PyMongo async
Package	motor	pymongo (4.9+, GA in 4.13)
Client class	AsyncIOMotorClient	AsyncMongoClient
Implementation	Thread pool around sync PyMongo	Native asyncio
Status	Deprecated, EOL 2026	Recommended
client.close()	Synchronous	Coroutine, must be awaited

The APIs are close enough that most code ports with a find-and-replace, plus a few details covered in the migration section below.

Project Setup

Install FastAPI and PyMongo. You don't need Motor at all for new code:

python -m venv .venv
source .venv/bin/activate
pip install "fastapi[standard]" "pymongo>=4.13" pydantic-settings

A small settings module keeps the connection string out of your code:

# app/config.py
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    mongodb_uri: str = "mongodb://localhost:27017"
    mongodb_db: str = "bookstore"


settings = Settings()

pydantic-settings reads MONGODB_URI and MONGODB_DB from the environment, so the same code runs locally and in production.

Managing the Client with Lifespan Events

The single most important rule for any MongoDB driver: create one client per process and reuse it. The client owns a connection pool, and creating one per request opens new TCP connections, TLS handshakes, and authentication every time. (If you want the full picture, see MongoDB Connection Pooling: Avoiding Too Many Connections.)

FastAPI's lifespan handler is the right place to create and close the client:

# app/main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI
from pymongo import AsyncMongoClient

from app.config import settings


@asynccontextmanager
async def lifespan(app: FastAPI):
    client = AsyncMongoClient(settings.mongodb_uri, appname="bookstore-api")
    # Fail fast if the database is unreachable
    await client.admin.command("ping")

    db = client[settings.mongodb_db]
    await db.books.create_index("isbn", unique=True)
    await db.books.create_index([("author", 1), ("published", -1)])

    app.state.mongo = client
    app.state.db = db
    yield
    await client.close()


app = FastAPI(lifespan=lifespan)

A few things worth noting:

• AsyncMongoClient(...) doesn't connect immediately. The ping forces a round trip so a bad URI or a blocked IP fails at startup rather than on the first user request.
• create_index is idempotent. If the index already exists with the same definition, the call is a no-op, so running it on every startup is safe for small collections. For large production collections, build indexes through a migration step instead so a deploy never kicks off a long index build.
• await client.close() is a coroutine in PyMongo's async API. In Motor, close() was synchronous. Forgetting the await here gives you a "coroutine was never awaited" warning and leaves connections open.

Getting the Database into Endpoints

Rather than reaching into app.state everywhere, expose the database through a dependency:

# app/deps.py
from typing import Annotated

from fastapi import Depends, Request
from pymongo.asynchronous.database import AsyncDatabase


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


Db = Annotated[AsyncDatabase, Depends(get_db)]

This also makes testing easier: you can override get_db to point at a test database.

Modeling Documents with Pydantic v2

MongoDB documents use _id with an ObjectId, which JSON doesn't understand. Pydantic needs to know how to turn that into a string on the way out. The cleanest pattern with Pydantic v2 is an annotated type that coerces ObjectId to str:

# app/models.py
from datetime import datetime, timezone
from typing import Annotated

from pydantic import BaseModel, BeforeValidator, ConfigDict, Field

PyObjectId = Annotated[str, BeforeValidator(str)]


class BookIn(BaseModel):
    title: str = Field(min_length=1, max_length=300)
    author: str
    isbn: str = Field(pattern=r"^\d{13}$")
    price: float = Field(gt=0)
    tags: list[str] = []


class BookOut(BookIn):
    model_config = ConfigDict(populate_by_name=True)

    id: PyObjectId = Field(alias="_id")
    published: datetime


class BookUpdate(BaseModel):
    title: str | None = None
    price: float | None = Field(default=None, gt=0)
    tags: list[str] | None = None


def utcnow() -> datetime:
    return datetime.now(timezone.utc)

BookIn validates incoming requests. BookOut maps _id to id and converts the ObjectId to a string. BookUpdate has every field optional so it can back a PATCH endpoint.

Keep these models as the boundary between HTTP and the database. Inside your code, you can still work with plain dictionaries, which is what the driver returns.

Writing CRUD Endpoints

With the client, dependency, and models in place, the endpoints are short:

# app/routes/books.py
from bson import ObjectId
from bson.errors import InvalidId
from fastapi import APIRouter, HTTPException, Query, status
from pymongo import ReturnDocument
from pymongo.errors import DuplicateKeyError

from app.deps import Db
from app.models import BookIn, BookOut, BookUpdate, utcnow

router = APIRouter(prefix="/books", tags=["books"])


def parse_id(book_id: str) -> ObjectId:
    try:
        return ObjectId(book_id)
    except InvalidId:
        raise HTTPException(status_code=404, detail="Book not found")


@router.post("", response_model=BookOut, status_code=status.HTTP_201_CREATED)
async def create_book(book: BookIn, db: Db):
    doc = book.model_dump() | {"published": utcnow()}
    try:
        result = await db.books.insert_one(doc)
    except DuplicateKeyError:
        raise HTTPException(status_code=409, detail="ISBN already exists")
    doc["_id"] = result.inserted_id
    return doc


@router.get("/{book_id}", response_model=BookOut)
async def get_book(book_id: str, db: Db):
    doc = await db.books.find_one({"_id": parse_id(book_id)})
    if doc is None:
        raise HTTPException(status_code=404, detail="Book not found")
    return doc


@router.get("", response_model=list[BookOut])
async def list_books(
    db: Db,
    author: str | None = None,
    limit: int = Query(default=20, le=100),
):
    query = {"author": author} if author else {}
    cursor = db.books.find(query).sort("published", -1).limit(limit)
    return await cursor.to_list()


@router.patch("/{book_id}", response_model=BookOut)
async def update_book(book_id: str, changes: BookUpdate, db: Db):
    update = changes.model_dump(exclude_unset=True)
    if not update:
        raise HTTPException(status_code=400, detail="No fields to update")
    doc = await db.books.find_one_and_update(
        {"_id": parse_id(book_id)},
        {"$set": update},
        return_document=ReturnDocument.AFTER,
    )
    if doc is None:
        raise HTTPException(status_code=404, detail="Book not found")
    return doc


@router.delete("/{book_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_book(book_id: str, db: Db):
    result = await db.books.delete_one({"_id": parse_id(book_id)})
    if result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Book not found")

Register the router in main.py with app.include_router(router) and run it with fastapi dev app/main.py.

Some details that trip people up:

• find() is not awaited. It returns an AsyncCursor immediately. You await the methods that actually fetch data: to_list(), or iterate with async for.
• to_list() without a length loads everything. That's fine here because of the limit. On an unbounded query, it will pull the entire result set into memory. Always pair it with limit() or pass a length.
• exclude_unset=True is what makes PATCH work correctly. Without it, fields the client didn't send would be None and you'd overwrite real data with nulls.
• Invalid ids become 404s. A malformed id string is just a book that doesn't exist from the client's point of view, so there's no need to leak a 500.

Streaming Large Results

For exports or large reports, iterate the cursor instead of building a list:

from fastapi.responses import StreamingResponse
import json


@router.get("/export/ndjson")
async def export_books(db: Db):
    async def generate():
        async for doc in db.books.find({}, {"_id": 0}).batch_size(500):
            yield json.dumps(doc, default=str) + "\n"

    return StreamingResponse(generate(), media_type="application/x-ndjson")

Memory stays flat no matter how many documents you export, because the driver fetches one batch at a time.

Aggregations

Aggregation is one of the places where PyMongo async differs from Motor. In PyMongo's async API, aggregate() is a coroutine that returns a cursor, so you await it first:

@router.get("/stats/by-author")
async def author_stats(db: Db):
    pipeline = [
        {"$group": {
            "_id": "$author",
            "books": {"$sum": 1},
            "avgPrice": {"$avg": "$price"},
        }},
        {"$sort": {"books": -1}},
        {"$limit": 10},
    ]
    cursor = await db.books.aggregate(pipeline)
    return [
        {"author": d["_id"], "books": d["books"], "avgPrice": round(d["avgPrice"], 2)}
        async for d in cursor
    ]

Example response:

[
  { "author": "Ursula K. Le Guin", "books": 14, "avgPrice": 16.49 },
  { "author": "Octavia E. Butler", "books": 9, "avgPrice": 15.2 }
]

In Motor, aggregate() returned the cursor directly without an await. This is exactly the kind of difference that makes a mechanical migration fail at runtime, so search for aggregate( and watch( when you port code.

Transactions

Multi-document transactions work the same way as in the sync driver, but with async context managers. The with_transaction helper retries on transient errors for you:

from pymongo import AsyncMongoClient


async def transfer_stock(client: AsyncMongoClient, sku: str, src: str, dst: str, qty: int):
    db = client["bookstore"]

    async def txn(session):
        res = await db.inventory.update_one(
            {"sku": sku, "warehouse": src, "qty": {"$gte": qty}},
            {"$inc": {"qty": -qty}},
            session=session,
        )
        if res.modified_count == 0:
            raise ValueError("Insufficient stock")
        await db.inventory.update_one(
            {"sku": sku, "warehouse": dst},
            {"$inc": {"qty": qty}},
            upsert=True,
            session=session,
        )

    async with client.start_session() as session:
        await session.with_transaction(txn)

Two reminders: pass session=session to every operation that should be part of the transaction, and remember that transactions require a replica set or sharded cluster. A standalone local mongod will reject them. Atlas clusters, including the free M0 tier, are replica sets, so they work out of the box.

Migrating from Motor

If you have an existing Motor codebase, the migration is usually a few hours of work. The main changes:

# Before (Motor)
from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient(uri)
cursor = db.books.aggregate(pipeline)
docs = await cursor.to_list(length=100)
client.close()

# After (PyMongo async)
from pymongo import AsyncMongoClient

client = AsyncMongoClient(uri)
cursor = await db.books.aggregate(pipeline)
docs = await cursor.to_list(length=100)
await client.close()

A checklist for the port:

1. Replace AsyncIOMotorClient with AsyncMongoClient and update type hints (AsyncIOMotorDatabase becomes AsyncDatabase from pymongo.asynchronous.database, and similarly for collections).
2. Add await in front of aggregate(), watch(), and client.close().
3. Check GridFS usage: PyMongo has its own async GridFS classes (AsyncGridFSBucket in gridfs), which replace Motor's AsyncIOMotorGridFSBucket.
4. Remove motor from your requirements so nothing imports it by accident.
5. Run your test suite and watch for "coroutine was never awaited" warnings, which point at missed awaits.

Libraries that depend on Motor, such as older versions of Beanie ODM, have been moving to PyMongo's async API too. Check the version you're on before upgrading your own code, since mixing the two clients in one process means two separate connection pools.

Testing the API

FastAPI's TestClient runs the lifespan handler when used as a context manager, which means your tests connect to a real database. Point it at a disposable one:

# tests/test_books.py
import os

os.environ["MONGODB_DB"] = "bookstore_test"

import pytest
from fastapi.testclient import TestClient
from pymongo import MongoClient

from app.config import settings
from app.main import app


@pytest.fixture(autouse=True)
def clean_db():
    # A plain sync client is fine for setup and teardown in sync tests
    sync_client = MongoClient(settings.mongodb_uri)
    yield
    sync_client.drop_database("bookstore_test")
    sync_client.close()


def test_create_and_fetch_book():
    with TestClient(app) as client:
        payload = {
            "title": "The Dispossessed",
            "author": "Ursula K. Le Guin",
            "isbn": "9780061054884",
            "price": 17.99,
        }
        created = client.post("/books", json=payload)
        assert created.status_code == 201
        book_id = created.json()["id"]

        fetched = client.get(f"/books/{book_id}")
        assert fetched.json()["title"] == "The Dispossessed"

        dup = client.post("/books", json=payload)
        assert dup.status_code == 409

For async tests, use httpx.AsyncClient with ASGITransport and pytest-asyncio, and run a MongoDB container in CI (the official mongo Docker image works well). Dropping the test database in a fixture, as above, means every run starts clean.

Notice the fixture uses the synchronous MongoClient. Test code that isn't running on an event loop can't await the async client's coroutines, and calling them without await silently does nothing. It's a common slip when mixing the two worlds.

Common Mistakes

Calling the synchronous MongoClient from async def endpoints. It works, which is exactly why it's dangerous. Each query blocks the event loop, and under load your latency climbs for every request on that worker. If you must use sync PyMongo, declare the endpoint with plain def so FastAPI runs it in a thread pool.

Creating a client per request. Putting AsyncMongoClient(uri) inside an endpoint or a dependency that runs per request creates a new pool every time. Connections pile up until you hit your server's limit. Create it once in lifespan.

Forgetting that each worker is a separate process. Running uvicorn --workers 4 or Gunicorn with four workers gives you four clients and four pools. Multiply maxPoolSize (default 100) by your worker count when estimating connections against your cluster's limit.

Returning raw documents with ObjectIds. Without a response model or conversion, FastAPI's JSON encoder fails on ObjectId and Decimal128. Always return through a Pydantic model that converts them.

Unbounded to_list(). An endpoint that calls to_list() on an unfiltered, unlimited cursor is a memory incident waiting for your collection to grow. Cap every list endpoint.

Conclusion

Async MongoDB with FastAPI comes down to a few habits: one AsyncMongoClient per process created in lifespan, Pydantic models at the HTTP boundary that handle ObjectId, bounded cursors, and await on everything that touches the network. Motor got the Python community here, but PyMongo's native async API is faster, better maintained, and the only path forward as Motor reaches end of life.

If you have a Motor project, pick one service, swap the import to AsyncMongoClient, add the missing awaits on aggregate() and close(), and run your tests. For a refresher on the synchronous driver concepts this builds on, see Connecting Python Apps to MongoDB with PyMongo.