MongoDB with FastAPI

10 min readApr 30, 2024

If you are having a Machine learning/ Data Science project, you may have most of your source codes written in python. Therefore it’s worth if you know FastAPI to expose some endpoints from your database or Machine Learning project for data visualization purposes in the frontend.

Today I am going to build a FastAPI from scratch which is going to interact with MongoDB.

First of all let’s see how should you arrange your development environment to getting started.

Install python and pip
Create a MongoDB Atlas cluster and copy the connection String
MongoDB Compass for checking the database entries
Postman for testing the endpoints

If you are using a mac
Check the Python installation path:
Use the command below to find out where Python is installed: which python3
This command will show the path from where the python3 command is being executed.
Verify system Python:
If the path points to a system directory like /usr/bin/python3, then Python might be the system's default version, which you generally shouldn't uninstall because some system tools might depend on it.
Alternative uninstall methods:
If Python is installed somewhere else and you still wish to uninstall it, you could potentially remove it manually if you know its installation directory or use another package manager that might have been used for the installation, like pip for Python packages.
Manage multiple Python versions:
If managing multiple versions of Python is required, consider using pyenv. It allows you to switch between multiple versions of Python easily. To install pyenv, you can use: brew install pyenv
After installing pyenv, you can manage different Python versions without affecting the system Python.
Using Pyenv:
Once pyenv is installed, you can install and manage separate Python versions with it: pyenv install 3.9.6 pyenv global 3.9.6

FastAPI Hello World

Create a folder called “mongodb-with-fastapi”
Create a file called “main.py” and add following code

main.py

# all the imports
from fastapi import FastAPI

# creating a server with python FastAPI
app = FastAPI()

# hello world endpoint
@app.get("/")
def read_root(): # function that is binded with the endpoint
    return {"Hello": "World"}

Then install fast api and unicorn. Becuase we have use thoese two packages to build above simple 5 line server.

pip install fastapi
pip install “uvicorn[standard]”

then run following command to run the server

uvicorn main:app — reload

You can see that it has created a new folder called __pycache__ which is definitely need to be added to a “.gitignore” file

if you open the URL given by the python unicorn through a web browser, you can see a output as follows.

Side Note: Python Global, Local and Virtual Environments
Global vs. Local Environment: If you’re using a globally installed Python interpreter or one managed by pyenv but not a virtual environment, the packages are installed in a directory linked to that Python interpreter (like under .pyenv/versions/3.9.6/lib/python3.9/site-packages).
Project Isolation with Virtual Environments: To manage dependencies more effectively, especially to avoid conflicts between project-specific packages, it’s a good practice to use virtual environments. Each virtual environment has its own Python interpreter and site-packages directory.

Add project to git

Before committing anything create following “.gitignore” file in your project directory

__pycache__
.vscode
.envrc

then create your initial commit and push the code to a remote repository

What is ASGI and How it works

What is ASGI

ASGI, or Asynchronous Server Gateway Interface, is a standard for Python asynchronous web servers and applications. It serves as a spiritual successor to WSGI (Web Server Gateway Interface), which is the traditional synchronous interface between web servers and web applications in Python.

ASGI is designed to support asynchronous capabilities, allowing for handling of more concurrent connections, which is useful for applications that need to manage long-lived connections, such as those using WebSockets.

How it works

Asynchronous Support: ASGI provides a standard for asynchronous web applications to communicate with asynchronous servers. This is crucial for handling modern web features like asynchronous data handling, WebSockets, and long polling efficiently.
Interface Specification: ASGI consists of a specification that describes how the web server communicates with the application. It’s divided into layers, with each layer handling different types of events:

HTTP: Handling HTTP requests and responses.
WebSocket: Handling WebSocket connections, enabling real-time bi-directional communication between the client and server.
Lifespan: Handling startup and shutdown events within the application.

3. Application Structure: An ASGI application is essentially an asynchronous callable, which accepts scope, receive, and send arguments:

Scope: Contains information about the type of connection (e.g., HTTP or WebSocket), details about the request, and other metadata.
Receive: An awaitable function that the application can call to receive events from the client.
Send: An awaitable function that the application can call to send events back to the client.

4. Middleware: ASGI supports middleware components that can wrap ASGI applications, allowing for operations such as modifying request or response data, handling exceptions, or performing other tasks around the request-response cycle.

Get Chanaka’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

5. Frameworks and Servers: Several web frameworks and servers support ASGI. For example, Starlette is a lightweight ASGI framework suitable for building high-performance asyncio services. Daphne and Uvicorn are popular ASGI servers that can serve ASGI applications.

Using ASGI enables Python applications to be built with native support for asynchronous processing, which is beneficial for handling a large number of simultaneous connections or for applications that require real-time updates.

How to use async await with FastAPI

Here’s how you can modify the given “Hello World” endpoint in FastAPI to use async and await:

from fastapi import FastAPI

app = FastAPI()
@app.get("/")
async def read_root():  # Define the function as async
    return {"Hello": "World"

In this example, the read_root function is defined with the async keyword, which makes it an asynchronous function. Even though in this simple case, using async doesn't provide a benefit (because there's no awaitable operation inside the function), it sets up the structure you would use for more complex operations. For instance, if you needed to fetch data from a database asynchronously or call an external API asynchronously, you would use the await keyword within the function like so:

from fastapi import FastAPI
import httpx  # Asynchronous HTTP client

app = FastAPI()

@app.get("/")
async def read_root():
    async with httpx.AsyncClient() as client:
        response = await client.get('https://api.example.com/data')
        return {"data": response.json()}

In this modified example, the endpoint uses an asynchronous HTTP client (httpx) to make an external API call. The await keyword is used to wait for the client.get() method to complete without blocking the rest of your application. This is beneficial in web server environments where you want to handle many requests simultaneously without waiting for each request to complete sequentially.

Implementing a model class for users

In the following code, I have a created a class called “User” with pydantic data validations. And also for the gender and the role of the user I have created another two classes of the Enum type. When it comes to the id, I have generated it using the python UUID package.

It’s so simple. Nothing fansy is there. But you can see I have imported another class called Optional from the package typing and used it for making the parameter middle_name a optional one.

from typing import Optional, List # Supports for type hints
from pydantic import BaseModel # Most widely used data validation library for python
from enum import Enum # Supports for enumerations

# enum for gender
class Gender(str, Enum):
    male = "male"
    female = "female"

# enum for role
class Role(str, Enum):
    admin = "admin"
    user = "user"
    student = "student"
    teacher = "teacher"

class User(BaseModel):
    first_name: str
    last_name: str
    middle_name: Optional[str] = None  # Make middle name optional
    gender: Gender
    email_address: str
    phone_number: str
    roles: List[Role] # user can have several roles

Connecting FastAPI with MongoDB

create a mongo db cluster, create a deployment and copy the connection string along with username and the password.

select compass

Select Compass and then from the next menu, copy the connection String.

in this example my connection string is: mongodb+srv://UserName:Password@cluster0.abc.mongodb.net/

Using previously installed MongoDB compass software connection to your newly created database to see the changes.

Implementing the database connection

# imports for the MongoDB database connection
from motor.motor_asyncio import AsyncIOMotorClient

# import for fast api lifespan
from contextlib import asynccontextmanager

# define a lifespan method for fastapi
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start the database connection
    await startup_db_client(app)
    yield
    # Close the database connection
    await shutdown_db_client(app)

# method for start the MongoDb Connection
async def startup_db_client(app):
    app.mongodb_client = AsyncIOMotorClient(
        "mongodb+srv://UserName:Password@cluster0.abc.mongodb.net/")
    app.mongodb = app.mongodb_client.get_database("college")
    print("MongoDB connected.")

# method to close the database connection
async def shutdown_db_client(app):
    app.mongodb_client.close()
    print("Database disconnected.")

# creating a server with python FastAPI
app = FastAPI(lifespan=lifespan)

Defining CRUD Operations

Create user endpoint

# C <=== Create
@app.post("/api/v1/create-user", response_model=User)
async def insert_user(user: User):
    result = await app.mongodb["users"].insert_one(user.dict())
    inserted_user = await app.mongodb["users"].find_one({"_id": result.inserted_id})
    return inserted_user

Read user endpoint

# R <=== Read
# Read all users
@app.get("/api/v1/read-all-users", response_model=List[User])
async def read_users():
    users = await app.mongodb["users"].find().to_list(None)
    return users

# Read one user by email_address
@app.get("/api/v1/read-user/{email_address}", response_model=User)
async def read_user_by_email(email_address: str):
    user = await app.mongodb["users"].find_one({"email_address": email_address})
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

Update user endpoint

# U <=== Update
# Update user

class UpdateUserDTO(BaseModel):
    other_names: List[str] = None
    age: int = None
    # Include other fields as needed, with defaults to None or use the exclude_unset=True option

@app.put("/api/v1/update-user/{email_address}", response_model=User)
async def update_user(email_address: str, user_update: UpdateUserDTO):
    updated_result = await app.mongodb["users"].update_one(
        {"email_address": email_address}, {"$set": user_update.dict(exclude_unset=True)})
    if updated_result.modified_count == 0:
        raise HTTPException(status_code=404, detail="User not found or no update needed")
    updated_user = await app.mongodb["users"].find_one({"email_address": email_address})
    return updated_user

Delete user endpoint

# D <=== Delete
# Delete user by email_address
@app.delete("/api/v1/delete-user/{email_address}", response_model=dict)
async def delete_user_by_email(email_address: str):
    delete_result = await app.mongodb["users"].delete_one({"email_address": email_address})
    if delete_result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="User not found")
    return {"message": "User deleted successfully"}

Requirements.txt file

Generating a requirements.txt file

pip freeze > requirements.txt

Install all the required dependencies at once

pip install -r requirements.txt

Full Code

main.py

# import fast api
from fastapi import FastAPI, HTTPException

from models import User # import the user model defined by us

# imports for the MongoDB database connection
from motor.motor_asyncio import AsyncIOMotorClient

# import for fast api lifespan
from contextlib import asynccontextmanager

from typing import List # Supports for type hints

from pydantic import BaseModel # Most widely used data validation library for python

# define a lifespan method for fastapi
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Start the database connection
    await startup_db_client(app)
    yield
    # Close the database connection
    await shutdown_db_client(app)

# method for start the MongoDb Connection
async def startup_db_client(app):
    app.mongodb_client = AsyncIOMotorClient(
        "mongodb+srv://UserName:Password@cluster0.abc.mongodb.net/")
    app.mongodb = app.mongodb_client.get_database("college")
    print("MongoDB connected.")

# method to close the database connection
async def shutdown_db_client(app):
    app.mongodb_client.close()
    print("Database disconnected.")

# creating a server with python FastAPI
app = FastAPI(lifespan=lifespan)

# hello world endpoint
@app.get("/")
def read_root():  # function that is binded with the endpoint
    return {"Hello": "World"}

# C <=== Create
@app.post("/api/v1/create-user", response_model=User)
async def insert_user(user: User):
    result = await app.mongodb["users"].insert_one(user.dict())
    inserted_user = await app.mongodb["users"].find_one({"_id": result.inserted_id})
    return inserted_user

# R <=== Read
# Read all users
@app.get("/api/v1/read-all-users", response_model=List[User])
async def read_users():
    users = await app.mongodb["users"].find().to_list(None)
    return users

# Read one user by email_address
@app.get("/api/v1/read-user/{email_address}", response_model=User)
async def read_user_by_email(email_address: str):
    user = await app.mongodb["users"].find_one({"email_address": email_address})
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user


# U <=== Update
# Update user

class UpdateUserDTO(BaseModel):
    other_names: List[str] = None
    age: int = None
    # Include other fields as needed, with defaults to None or use the exclude_unset=True option

@app.put("/api/v1/update-user/{email_address}", response_model=User)
async def update_user(email_address: str, user_update: UpdateUserDTO):
    updated_result = await app.mongodb["users"].update_one(
        {"email_address": email_address}, {"$set": user_update.dict(exclude_unset=True)})
    if updated_result.modified_count == 0:
        raise HTTPException(status_code=404, detail="User not found or no update needed")
    updated_user = await app.mongodb["users"].find_one({"email_address": email_address})
    return updated_user

# D <=== Delete
# Delete user by email_address
@app.delete("/api/v1/delete-user/{email_address}", response_model=dict)
async def delete_user_by_email(email_address: str):
    delete_result = await app.mongodb["users"].delete_one({"email_address": email_address})
    if delete_result.deleted_count == 0:
        raise HTTPException(status_code=404, detail="User not found")
    return {"message": "User deleted successfully"}

models.py

from typing import Optional, List # Supports for type hints
from pydantic import BaseModel # Most widely used data validation library for python
from enum import Enum # Supports for enumerations

# enum for gender
class Gender(str, Enum):
    male = "male"
    female = "female"

# enum for role
class Role(str, Enum):
    admin = "admin"
    user = "user"
    student = "student"
    teacher = "teacher"

class User(BaseModel):
    first_name: str
    last_name: str
    middle_name: Optional[str] = None  # Make middle name optional
    gender: Gender
    email_address: str
    phone_number: str
    roles: List[Role] # user can have several roles