MongoDB integration with Motor in FastAPI - Mini Project: Build & Apply

MongoDB Integration with Motor in FastAPI

📖 Scenario: You are building a simple FastAPI application that stores and retrieves user information from a MongoDB database. You will use Motor, an asynchronous MongoDB driver, to connect and interact with the database.

🎯 Goal: Create a FastAPI app that connects to MongoDB using Motor, inserts a user document, and retrieves it asynchronously.

📋 What You'll Learn

Create a Motor client connected to MongoDB

Define a database and collection for users

Insert a user document asynchronously

Retrieve the inserted user document asynchronously

💡 Why This Matters

🌍 Real World

Many modern web applications use FastAPI with MongoDB for scalable, asynchronous data storage and retrieval.

💼 Career

Understanding how to integrate FastAPI with MongoDB using Motor is valuable for backend developers working on asynchronous Python web services.

1

Setup Motor Client and FastAPI App

Import FastAPI and AsyncIOMotorClient from motor.motor_asyncio. Create a FastAPI app called app. Create a Motor client called client connected to mongodb://localhost:27017. Define a database called testdb from the client.

FastAPI

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

# Create FastAPI app
# Your code here

Hint

Use AsyncIOMotorClient to connect to MongoDB asynchronously. Create the FastAPI app instance first.

2

Define User Collection

Create a variable called user_collection that refers to the users collection inside the testdb database.

FastAPI

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()
client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client.testdb

# Define the users collection
# Your code here

Hint

Access the collection by using db.users.

3

Create Async Endpoint to Insert User

Define an async function called create_user with a POST route at /users/. Inside the function, insert a document with {"name": "Alice", "age": 30} into user_collection using insert_one asynchronously.

FastAPI

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()
client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client.testdb
user_collection = db.users

# Create async POST endpoint to insert user
# Your code here

Hint

Use @app.post("/users/") decorator and define an async function. Use await with insert_one.

4

Create Async Endpoint to Retrieve User

Define an async function called get_user with a GET route at /users/{name}. Inside the function, find one document in user_collection where name matches the path parameter. Return the found document as a dictionary, converting _id to string.

FastAPI

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient

app = FastAPI()
client = AsyncIOMotorClient("mongodb://localhost:27017")
db = client.testdb
user_collection = db.users

@app.post("/users/")
async def create_user():
    user = {"name": "Alice", "age": 30}
    result = await user_collection.insert_one(user)
    return {"inserted_id": str(result.inserted_id)}

# Create async GET endpoint to retrieve user by name
# Your code here

Hint

Use @app.get("/users/{name}") and an async function with a name parameter. Use await with find_one. Convert _id to string before returning.

Practice

1. What is the main benefit of using Motor with FastAPI for MongoDB operations?

easy

A. It automatically converts ObjectId to string.

B. It provides a graphical interface for MongoDB.

C. It allows asynchronous, non-blocking database calls.

D. It replaces the need for FastAPI routing.

Solution

1. Step 1: Understand Motor's role in FastAPI

   Motor is an async driver for MongoDB, enabling non-blocking calls in FastAPI apps.

2. Step 2: Identify the main benefit

   Non-blocking calls keep the app responsive and fast by not waiting for DB operations.

3. Final Answer:

   It allows asynchronous, non-blocking database calls. -> Option C

4. Quick Check:

   Motor = async, non-blocking calls [OK]

Hint: Motor means async MongoDB calls in FastAPI [OK]

Common Mistakes:

• Thinking Motor provides UI tools
• Assuming Motor auto-converts ObjectId
• Confusing Motor with FastAPI routing

2. Which of the following is the correct way to define a Motor client in FastAPI?

easy

A. client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017')

B. client = MotorClient('mongodb://localhost:27017')

C. client = AsyncMongoClient('localhost:27017')

D. client = motor.MongoClient('mongodb://localhost:27017')

Solution

1. Step 1: Recall Motor client syntax

   The Motor async client is created using motor.motor_asyncio.AsyncIOMotorClient with the MongoDB URI.

2. Step 2: Check each option

   Only client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017') uses the correct class and URI format for Motor.

3. Final Answer:

   client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017') -> Option A

4. Quick Check:

   Motor client = AsyncIOMotorClient [OK]

Hint: Use AsyncIOMotorClient from motor.motor_asyncio [OK]

Common Mistakes:

• Using MotorClient instead of AsyncIOMotorClient
• Omitting 'motor.motor_asyncio' prefix
• Using synchronous MongoClient

3. Given this FastAPI async function using Motor, what will be the output?

async def get_user(db, user_id):
    user = await db.users.find_one({"_id": user_id})
    return str(user["_id"])

medium

A. Returns None if user_id is not found.

B. Returns the string representation of the user's ObjectId.

C. Raises a TypeError because ObjectId can't be converted to string.

D. Returns the user document as a dictionary.

Solution

1. Step 1: Analyze the async function

   The function awaits a find_one query by _id and then converts the _id field to string.

2. Step 2: Understand the output

   It returns the string form of the ObjectId, not the whole document or None.

3. Final Answer:

   Returns the string representation of the user's ObjectId. -> Option B

4. Quick Check:

   str(ObjectId) = string id [OK]

Hint: str() converts ObjectId to string safely [OK]

Common Mistakes:

• Expecting full user dict as output
• Thinking ObjectId can't be stringified
• Assuming None is returned if not found

4. Identify the error in this Motor usage inside FastAPI:

async def fetch_items(db):
    items = db.items.find()
    return await items.to_list(length=100)

medium

A. Missing await before db.items.find() call.

B. to_list() should not be awaited.

C. find() returns a cursor and must be awaited before to_list().

D. No error; code is correct.

Solution

1. Step 1: Understand Motor's find() behavior

   Motor's find() returns an async cursor immediately; no await needed here.

2. Step 2: Check to_list() usage

   to_list() is a coroutine and must be awaited to get the list of documents.

3. Final Answer:

   No error; code is correct. -> Option D

4. Quick Check:

   find() no await, to_list() await [OK]

Hint: Await to_list(), not find() [OK]

Common Mistakes:

• Awaiting find() call incorrectly
• Not awaiting to_list() causing runtime error
• Confusing synchronous and async cursor methods

5. You want to return a list of users from MongoDB in FastAPI, but each user's _id must be a string, not ObjectId. Which code snippet correctly does this?

hard

A. users = await db.users.find().to_list(100)
return [{**u, '_id': str(u['_id'])} for u in users]

B. users = await db.users.find().to_list(100)
return [u for u in users if str(u['_id'])]

C. users = await db.users.find().to_list(100)
return [str(u) for u in users]

D. users = await db.users.find().to_list(100)
return [u['_id'] for u in users]

Solution

1. Step 1: Fetch users as list of dicts

   await db.users.find().to_list(100) returns list of user dicts with ObjectId _id fields.

2. Step 2: Convert each user's _id to string

   Use list comprehension to copy each dict and replace _id with str(u['_id']).

3. Final Answer:

   users = await db.users.find().to_list(100) return [{**u, '_id': str(u['_id'])} for u in users] -> Option A

4. Quick Check:

   Convert ObjectId to string per user [OK]

Hint: Use dict unpacking and str() on _id in list comprehension [OK]

Common Mistakes:

• Returning ObjectId directly without conversion
• Trying to convert whole user dict to string
• Filtering users instead of converting _id