MongoDB is a flexible and scalable NoSQL database widely used in modern applications. It can be integrated with FastAPI using PyMongo, the official MongoDB driver for Python.

In this guide, you will learn how to connect FastAPI to MongoDB and implement complete CRUD operations for managing student records.

Step-by-Step Implementation

Step 1: Set Up MongoDB

Ensure MongoDB is installed and running on your computer. By default, MongoDB runs on:

mongodb://localhost:27017

If you do not want to install MongoDB locally, you can use MongoDB Atlas, MongoDB’s cloud-hosted database service.

Make sure the MongoDB user has permission to perform create, read, update, and delete operations.

Step 2: Install the Required Dependencies

Install FastAPI, Uvicorn, PyMongo, and Pydantic using the following command:

pip install fastapi uvicorn pymongo pydantic

The packages are used for the following purposes:

FastAPI: Creates the REST API.
Uvicorn: Runs the FastAPI application.
PyMongo: Connects Python to MongoDB.
Pydantic: Validates incoming and outgoing data.

Step 3: Create the Pydantic Models

Pydantic models define the expected structure of the student data and automatically validate incoming API requests.

from pydantic import BaseModel, Field


class Address(BaseModel):
    city: str
    country: str


class Student(BaseModel):
    name: str
    age: int = Field(ge=1)
    address: Address

The Address model represents a nested MongoDB document. The Student model contains the student's name, age, and address.

Step 4: Connect FastAPI to MongoDB

Use MongoClient from PyMongo to establish a connection with MongoDB:

from pymongo import MongoClient

client = MongoClient("mongodb://localhost:27017")

db = client["library_management"]
students_collection = db["students"]

In this example:

library_management is the database name.
students is the collection name.

Step 5: Create the FastAPI Application

from fastapi import FastAPI

app = FastAPI(
    title="Student Management API",
    description="A CRUD API built with FastAPI and MongoDB",
    version="1.0.0"
)

Step 6: Implement CRUD Operations

The application will contain the following API endpoints:

POST /students – Create a student.
GET /students – List students.
GET /students/{id} – Retrieve one student.
PATCH /students/{id} – Update a student.
DELETE /students/{id} – Delete a student.

Complete FastAPI and MongoDB Code

Create a file named main.py and add the following code:

from typing import Optional

from bson import ObjectId
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel, Field
from pymongo import MongoClient
from pymongo.errors import PyMongoError


app = FastAPI(
    title="Student Management API",
    description="CRUD API using FastAPI and MongoDB",
    version="1.0.0",
)


# MongoDB connection
client = MongoClient(
    "mongodb://localhost:27017",
    serverSelectionTimeoutMS=5000,
)

db = client["library_management"]
students_collection = db["students"]


class Address(BaseModel):
    city: str
    country: str


class Student(BaseModel):
    name: str
    age: int = Field(ge=1)
    address: Address


class StudentUpdate(BaseModel):
    name: Optional[str] = None
    age: Optional[int] = Field(default=None, ge=1)
    address: Optional[Address] = None


def validate_object_id(student_id: str) -> ObjectId:
    if not ObjectId.is_valid(student_id):
        raise HTTPException(
            status_code=400,
            detail="Invalid student ID",
        )

    return ObjectId(student_id)


def serialize_student(student: dict) -> dict:
    return {
        "id": str(student["_id"]),
        "name": student["name"],
        "age": student["age"],
        "address": student["address"],
    }


@app.get("/")
def home():
    return {"message": "Student Management API is running"}


@app.post("/students", status_code=201)
def create_student(student: Student):
    try:
        result = students_collection.insert_one(student.model_dump())

        return {
            "message": "Student created successfully",
            "id": str(result.inserted_id),
        }
    except PyMongoError:
        raise HTTPException(
            status_code=500,
            detail="Unable to create student",
        )


@app.get("/students")
def list_students(
    country: Optional[str] = None,
    minimum_age: Optional[int] = Query(default=None, ge=1),
):
    query = {}

    if country:
        query["address.country"] = country

    if minimum_age is not None:
        query["age"] = {"$gte": minimum_age}

    try:
        students = students_collection.find(query)
        return [serialize_student(student) for student in students]
    except PyMongoError:
        raise HTTPException(
            status_code=500,
            detail="Unable to retrieve students",
        )


@app.get("/students/{student_id}")
def get_student(student_id: str):
    object_id = validate_object_id(student_id)

    try:
        student = students_collection.find_one({"_id": object_id})
    except PyMongoError:
        raise HTTPException(
            status_code=500,
            detail="Unable to retrieve student",
        )

    if student is None:
        raise HTTPException(
            status_code=404,
            detail="Student not found",
        )

    return serialize_student(student)


@app.patch("/students/{student_id}")
def update_student(student_id: str, student: StudentUpdate):
    object_id = validate_object_id(student_id)

    update_data = student.model_dump(exclude_unset=True)

    if not update_data:
        raise HTTPException(
            status_code=400,
            detail="No fields were provided for updating",
        )

    try:
        result = students_collection.update_one(
            {"_id": object_id},
            {"$set": update_data},
        )
    except PyMongoError:
        raise HTTPException(
            status_code=500,
            detail="Unable to update student",
        )

    if result.matched_count == 0:
        raise HTTPException(
            status_code=404,
            detail="Student not found",
        )

    updated_student = students_collection.find_one({"_id": object_id})

    return {
        "message": "Student updated successfully",
        "student": serialize_student(updated_student),
    }


@app.delete("/students/{student_id}")
def delete_student(student_id: str):
    object_id = validate_object_id(student_id)

    try:
        result = students_collection.delete_one({"_id": object_id})
    except PyMongoError:
        raise HTTPException(
            status_code=500,
            detail="Unable to delete student",
        )

    if result.deleted_count == 0:
        raise HTTPException(
            status_code=404,
            detail="Student not found",
        )

    return {"message": "Student deleted successfully"}

Running the FastAPI Application

Start the development server using Uvicorn:

uvicorn main:app --reload

The API will be available at:

http://127.0.0.1:8000

FastAPI automatically generates interactive API documentation:

Swagger UI: http://127.0.0.1:8000/docs
ReDoc: http://127.0.0.1:8000/redoc

Testing the API

Create a Student

Send a POST request to /students with the following JSON:

{
    "name": "Rahul Kumar",
    "age": 21,
    "address": {
        "city": "Hyderabad",
        "country": "India"
    }
}

List All Students

GET /students

Filter Students by Country

GET /students?country=India

Filter Students by Minimum Age

GET /students?minimum_age=18

Get One Student

GET /students/64f21e127bf781256cd12345

Update a Student

Send a PATCH request with only the fields that need to be changed:

{
    "age": 22
}

Delete a Student

DELETE /students/64f21e127bf781256cd12345

How the Application Works

The Address model validates the nested city and country fields.
The Student model validates the complete student record.
The StudentUpdate model allows partial updates through PATCH requests.
insert_one() creates a new MongoDB document.
find() retrieves multiple documents.
find_one() retrieves a single document.
update_one() updates an existing document.
delete_one() removes a document.
MongoDB’s ObjectId is converted into a string before it is returned as JSON.

Important: PyMongo performs synchronous database operations. Therefore, the endpoints in this example use regular def functions instead of async def. FastAPI runs these endpoints in a thread pool, preventing MongoDB operations from blocking the main event loop.

Using Environment Variables

Avoid placing database credentials directly inside the source code in production. Store the MongoDB connection URL in an environment variable instead.

Create a .env file:

MONGODB_URL=mongodb://localhost:27017
MONGODB_DATABASE=library_management

Install python-dotenv:

pip install python-dotenv

Load the environment variables in Python:

import os

from dotenv import load_dotenv
from pymongo import MongoClient

load_dotenv()

mongodb_url = os.getenv("MONGODB_URL")
database_name = os.getenv("MONGODB_DATABASE")

client = MongoClient(mongodb_url)
db = client[database_name]

Conclusion

FastAPI and MongoDB provide a simple and powerful combination for building scalable APIs. FastAPI handles request validation, routing, and automatic documentation, while MongoDB provides flexible document-based data storage.

By using PyMongo with properly structured Pydantic models, validated MongoDB IDs, partial update models, and clear error handling, you can build a reliable CRUD API suitable for real-world FastAPI projects.

Mongodb Integration in Fastapi with PyMongo