MongoDB integration with Motor in FastAPI

Introduction

We use Motor to connect FastAPI with MongoDB because Motor lets us talk to the database without stopping our app. It works smoothly with FastAPI's async style.

When you want to save user data from a web form into MongoDB.

When you need to fetch and show data from MongoDB in a FastAPI app.

When you want your FastAPI app to handle many database requests at the same time without slowing down.

When you want to update or delete records in MongoDB from your FastAPI backend.

Syntax

FastAPI

from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient('mongodb://localhost:27017')
db = client['database_name']
collection = db['collection_name']

# Example async function to insert a document
async def insert_document(doc):
    result = await collection.insert_one(doc)
    return result.inserted_id

Use AsyncIOMotorClient to create a connection that works with async code.

Always use await when calling Motor methods to avoid blocking.

Examples

Connect to MongoDB on your local machine and select the 'testdb' database and 'users' collection.

FastAPI

client = AsyncIOMotorClient('mongodb://localhost:27017')
db = client['testdb']
collection = db['users']

Insert a new user document asynchronously and get its unique ID.

FastAPI

async def add_user(user_data):
    result = await collection.insert_one(user_data)
    return result.inserted_id

Find one user document by name asynchronously.

FastAPI

async def find_user(name):
    user = await collection.find_one({'name': name})
    return user

Sample Program

This FastAPI app connects to MongoDB using Motor. It has two routes: one to add an item and one to get an item by name.

FastAPI

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()

client = AsyncIOMotorClient('mongodb://localhost:27017')
db = client['sampledb']
collection = db['items']

@app.post('/items/')
async def create_item(item: dict):
    inserted_id = await collection.insert_one(item)
    return {'inserted_id': str(inserted_id.inserted_id)}

@app.get('/items/{name}')
async def read_item(name: str):
    item = await collection.find_one({'name': name})
    if item:
        item['_id'] = str(item['_id'])
        return item
    return {'error': 'Item not found'}

Important Notes

Motor is an async driver, so always use async functions and await Motor calls.

Convert MongoDB ObjectId to string before returning JSON to avoid errors.

Make sure MongoDB server is running before connecting.

Summary

Motor lets FastAPI talk to MongoDB without waiting or blocking.

Use async functions and await Motor calls to keep your app fast.

Remember to convert ObjectId to string when sending data back to users.