Using MongoDB

MongoDB is a document database: collections of JSON-like documents with flexible schemas. In FastAPI you connect with the official motor driver (the async MongoDB driver) or the sync pymongo client, and map Pydantic models to documents for validation in both directions.

Install

pip install fastapi motor

Connect Async with Motor

from motor.motor_asyncio import AsyncIOMotorClient

client = AsyncIOMotorClient("mongodb://127.0.0.1:27017")
db = client["appdb"]
items_col = db["items"]

Pydantic Models as Schemas

from pydantic import BaseModel, Field, ConfigDict

class Item(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    name: str
    price: float = Field(gt=0)
    id: str | None = None  # mongo _id as str

Async CRUD Endpoints

from fastapi import FastAPI, HTTPException, status

app = FastAPI()

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    doc = item.model_dump(exclude_none=True)
    result = await items_col.insert_one(doc)
    return {**doc, "id": str(result.inserted_id)}

@app.get("/items/")
async def list_items(limit: int = 100):
    docs = await items_col.find().to_list(length=limit)
    return [{"id": str(d["_id"]), **{k: v for k, v in d.items() if k != "_id"}} for d in docs]

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    doc = await items_col.find_one({"_id": ObjectId(item_id)})
    if doc is None:
        raise HTTPException(status_code=404, detail="Not found")
    return {"id": str(doc["_id"]), **{k: v for k, v in doc.items() if k != "_id"}}

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: str):
    res = await items_col.delete_one({"_id": ObjectId(item_id)})
    if res.deleted_count == 0:
        raise HTTPException(status_code=404, detail="Not found")
    return None

(import bson and use from bson import ObjectId at the top of the module.)

Sync Alternative with pymongo

from pymongo import MongoClient

client = MongoClient("mongodb://127.0.0.1:27017")
items = client["appdb"]["items"]

@app.post("/items/")
def create_item_sync(item: Item):
    result = items.insert_one(item.model_dump())
    return {"id": str(result.inserted_id)}

Sync pymongo works in def (not async def) endpoints; FastAPI runs them in a threadpool so the event loop stays free.

Summary

Use motor for async MongoDB access from async FastAPI endpoints (insert_one, find, find_one, delete_one) or pymongo in sync endpoints. Store _id as a string in your API surface, validate with Pydantic models, and raise 404 when a document is missing.