0Pricing
FastAPI Backend Development Bootcamp · レッスン

Motorによる非同期MongoDBアクセス

Motor非同期ドライバーでFastAPIとMongoDBを接続し、アプリのライフスパンで接続のライフサイクルを管理します。

「Motorによる非同期MongoDBアクセス」はCoddyKit上の無料FastAPI Backend Development Bootcampレッスンです。 これはレッスン1/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応の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アクセス」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、FastAPI Backend Development Bootcampコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 FastAPI Backend Development Bootcampコースには全4レッスンが含まれています。

「Motorによる非同期MongoDBアクセス」で何を学びますか？

Motor非同期ドライバーでFastAPIとMongoDBを接続し、アプリのライフスパンで接続のライフサイクルを管理します。 ブラウザで直接実行するハンズオンコードでFastAPI Backend Development Bootcampを演習し、24時間対応の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に戻る