Horje
FastAPI - Using MongoDB

Let’s explore a new way to create APIs using FastAPI. It’s fast, easy, and powerful. In this article, we’ll also see how to use MongoDB to do things like adding, reading, updating, and deleting data in our API.

MongoDB with FastAPI

After creating an API, the next step is to have a database that can store the data received through the API, enabling CRUD (Create, Read, Update, Delete) operations. In this article, we will explore MongoDB, a widely-used NoSQL (non-relational) database management system. MongoDB stores data in a flexible, JSON-like format known as BSON (Binary JSON), making it suitable for both structured and semi-structured data.

Setting up the Project

Now, let’s explore a basic example where our primary objective is to perform crud operation in employee details through API endpoints and verify if they are successfully registered in the MongoDB database. MongoDB can be configured either on a local environment (Windows | MacOS) or through the MongoDB Atlas web version. In this article, we will demonstrate the setup with a local MongoDB installation, but the process remains the same for MongoDB Atlas, with the only difference being the connection string.

Spinning up MongoDB in MacOS

  • Running MongoDB as a Mac OS service
  • Running MongoDB manually as a background process.

In this example we would be running the MongoDB as a background process. To run MongoDB the following command can be used

mongod --config /usr/local/etc/mongod.conf --fork

For MacOS silicon processor use the command

mongod --config /opt/homebrew/etc/mongod.conf --fork

Output

image-(4)-(1)

For WindowsOS use the command

mongod.exe --config "C:\path\to\your\mongod.conf"

The output of the following command would be like in windows

about to fork child process, waiting until server is ready for connections.
forked process: 1531
child process started successfully, parent exiting

Setup DB and create user

To begin with the furthur steps we need to connect mongosh with the running instance by the command. 

mongosh

Output:

image-(5)-(1)

First switch to the database that is to be created by the following command. In the following example we will be using the employee database.

db = db.getSiblingDB('employee')

Output:

Screenshot-2023-11-01-at-101728-AM

After creating the database we need to create the user by the following command

db.createUser({
                  user: 'gfgsayantan.2023',
                  pwd: '1992',
                  roles: [
                              {
                                        role: 'root',
                                        db: 'admin',
            },
        ],
    })

Output:

Screenshot-2023-11-01-at-102018-AM

After creating the user we can verify if the user creation is successful by following command

db.getUsers()

The output should be

{
  users: [
    {
      _id: 'employee.sayantanbose',
      userId: new UUID("0e0e36aa-0212-4cec-8ebb-6d1d18c10964"),
      user: 'sayantanbose',
      db: 'employee',
      roles: [ { role: 'root', db: 'admin' } ],
      mechanisms: [ 'SCRAM-SHA-1', 'SCRAM-SHA-256' ]
    }
  ],
  ok: 1
}

Output:

Screenshot-2023-11-01-at-102753-AM-(1)

And with this our MongoDB connection is successful.

Connecting MongoDb with Fast API

Creating directory and installing dependencies.

mkdir fastAPI
cd fastAPI
python -m venv fastAPI
source fastAPI/bin/activate
#Install the dependencies
pip install fastapi uvicorn pydantic pymongo
#To start the server
python3 main.py

Output:

image-(6)-(1)-(1)

In the provided code, we create a directory and set up a virtual environment. Establishing a virtual environment is considered a best practice, as it ensures that dependencies are isolated to that specific environment rather than being installed globally. Once the virtual environment is activated, we proceed with the installation of necessary packages using pip.

Creating MangoDB Connection

Here we are creating the MongoDB connection string to connect with the running instance of MongoDB.

Python3

from pymongo import MongoClient
 
# Replace these with your MongoDB connection details
# MongoDB username
# MongoDB Password
# MongoDB hosting type
# Default port of MongoDB is 27017
# MongoDB Database name
MONGO_USERNAME = "sayantanbose"
MONGO_PASSWORD = "1992"
MONGO_HOST = "localhost"
MONGO_PORT = 27017
MONGO_DB = "employee"
 
# Create a MongoDB client
client = MongoClient(
    f"mongodb://{MONGO_USERNAME}:{MONGO_PASSWORD}@{MONGO_HOST}:
  {MONGO_PORT}/{MONGO_DB}")
db = client[MONGO_DB]

In the provided code, we define the data schema that will be stored in the database. This can be regarded as the specific data structure that will encompass the details of employees.

Python3

from pydantic import BaseModel
 
 
class Employee(BaseModel):
    first_name: str
    last_name: str
    email: str
    phone_number: str
    salary: float

CRUD App with FastAPI and MongoDB

Create Employee: In the create_employee function, we are accepting an employee object of type Employee that we have defined in our schema. As it is a post request we will receive a body where the information of the employee object will be stored. As email is a unique field we first check if the employee with that email exists then we throw a 404 status code else we register the employee to our database.

Get All Employee: In the get_all_employees function we get all the employees in our MongoDB.

Get Particular Employee: In the get_employee_by_email function, we receive the email from the URL param. After verifying if the email exists in our database we return the employee details.

Update Particular Employee: In the update_employee_by_email function, we are receiving a body with updated details. We first check if an employee with the email exists then we check if the email is the same as with the updated details as the email is unique and cannot be changed. After verifying we just update the details of the employee to our database.

Delete Particular Employee: In the delete_employee_by_email we are receiving an email for the URL param. So we are verifying whether employees with such emails exist or not. If exist then the details of the employee are deleted from the database.

Python3

from fastapi import HTTPException
from fastapi import APIRouter, HTTPException
from MongoDB.mongo_connection import db
from models.model import Employee
 
router = APIRouter()
 
 
@router.post("/create_employee")
async def create_item(employee: Employee):
    """
    Create a new employee.
 
    Args:
        employee (Employee): The employee details to be created.
 
    Returns:
        dict: A JSON response with a message indicating the creation status.
    """
 
    # Check if the email already exists in the database
    existing_employee = db.items.find_one({"email": employee.email})
    if existing_employee:
        raise HTTPException(
            status_code=404, detail="Email Address already exists")
 
    # If the email doesn't exist, insert the new employee
    result = db.items.insert_one(employee.dict())
    inserted_employee = db.items.find_one({"email": employee.email})
    print(inserted_employee)
    inserted_email = str(inserted_employee['email'])
    return {"message":
            f"Employee with email {inserted_email} has been created."}
 
 
@router.get("/get_all_employees")
async def get_all_employees():
    """
    Get a list of all employees.
 
    Returns:
        list: A list of employee records as JSON objects.
    """
 
    # Query the database to retrieve all employees
    employees = list(db.items.find({}))
 
    # If no employees are found, return an empty list
    if not employees:
        return []
 
    for employee in employees:
        employee["_id"] = str(employee["_id"])
 
    return employees
 
 
@router.get("/get_employee_by_email")
async def get_employee_by_email(email: str):
    """
    Get an employee by email.
 
    Args:
        email (str): The email address of the employee to retrieve.
 
    Returns:
        dict: The employee record as a JSON object.
    """
 
    # Query the database to retrieve an employee by email
    employee = db.items.find_one({"email": email})
 
    # If no employee is found, return a 404 Not Found response
    if not employee:
        raise HTTPException(status_code=404, detail="Employee not found")
 
    employee["_id"] = str(employee["_id"])
 
    return employee
 
 
@router.put("/update_employee_by_email")
async def update_employee_by_email(email: str,
                                   updated_employee: Employee):
    """
    Update an employee by email.
 
    Args:
        email (str): The email address of the employee to update.
        updated_employee (Employee): The updated employee details.
 
    Returns:
        dict: A JSON response with a message indicating the update status.
    """
 
    # Check if the email exists in the database
    existing_employee = db.items.find_one({"email": email})
    if not existing_employee:
        raise HTTPException(
            status_code=404, detail="Employee not found")
 
    # Ensure that the updated_employee's email is the same
    # as the existing email
    if updated_employee.email != email:
        raise HTTPException(
            status_code=400, detail="Email cannot be changed")
 
    existing_employee_id = existing_employee.get('_id')
    if existing_employee_id:
        existing_employee_id = str(existing_employee_id)
 
    # Remove the _id field from the updated_employee dictionary
    updated_employee_dict = updated_employee.dict()
    updated_employee_dict.pop('_id', None)
 
    # Update the employee with the provided email
    db.items.update_one(
        {"email": email},
        {"$set": updated_employee_dict}
    )
 
    # Retrieve the updated employee from the database
    updated_employee_doc = db.items.find_one({"email": email})
    updated_employee_doc.pop('_id', None)
    print(updated_employee_doc)
 
    return {"message": f"Employee with email {email} has been updated.",
            "updated_employee": updated_employee_doc}
 
 
@router.delete("/delete_employee_by_email")
async def delete_employee_by_email(email: str):
    """
    Delete an employee by email.
 
    Args:
        email (str): The email address of the employee to delete.
 
    Returns:
        dict: A JSON response with a message indicating
        the deletion status.
    """
 
    # Check if the email exists in the database
    existing_employee = db.items.find_one({"email": email})
    if not existing_employee:
        raise HTTPException(
            status_code=404, detail="Employee not found")
 
    # If the email exists, delete the employee
    db.items.delete_one({"email": email})
 
    return {"message": f"Employee with email {email} has been deleted."}

Testing the API

Create Employee

To test the API, create a POST request to the ‘/create_employee/’ endpoint using Postman. We should receive a 200 status code along with the configured success message. The body of the Post request would be

{
    "first_name": "Sayantan",
    "last_name": "Bose",
    "email": "[email protected]",
    "phone_number": "6291753297",
    "salary": 800000.00
}

And the returned response should be

{
    "message": "Employee with email [email protected] has been created."
}

Get All Employees

To get the list of all employees we need to make a get request in the endpoint ‘/get_all_employees/’. We should receive a 200 status code along with all the employees that are registered in the DB

The returned response should be of structure

[
    {
        "_id": "650d4a6c3b3363329b11c8db",
        "first_name": "Sayantan",
        "last_name": "Bose",
        "email": "[email protected]",
        "phone_number": "6291753266",
        "salary": 890007.0
    },
    {
        "_id": "6511a418af34cce93c95ce6c",
        "first_name": "Sayantan",
        "last_name": "Bose",
        "email": "[email protected]",
        "phone_number": "6291753297",
        "salary": 800000.0
    }
]

Get Particular Employee

To get a particular employee we need to pass a param to the url and make a get request to the endpoint ‘/get_employee_by_email/’ so that the information of the particular employee is returned as a response. To perform our testing we can use the following url and the email id is considered a unique attribute in our example.

http://localhost:8000/get_employee_by_email/[email protected]

And the response returned is

{
    "_id": "6511a418af34cce93c95ce6c",
    "first_name": "Sayantan",
    "last_name": "Bose",
    "email": "[email protected]",
    "phone_number": "6291753297",
    "salary": 800000.0
}

Update Particular Employee in flast api using Mongodb

To update a particular employee we need to pass a param to the url so that the information of the particular employee is retrieved and updated with the updated information passed as a body of the post request made in the enpoint ‘/update_employee_by_email/’. To perform our testing we can use the following url and the email id is considered a unique attribute in our example.

http://localhost:8000/update_employee_by_email/[email protected]

And the body should be passed with the updated information. In the example we are updating the “salary” and “phone_number” attribute.

{
    "first_name": "Sayantan",
    "last_name": "Bose",
    "email": "[email protected]",
    "phone_number": "6291753266",
    "salary": 850000.00
}

And the response returned is

 {
    "message": "Employee with email [email protected] has been updated.",
    "updated_employee": {
        "first_name": "Sayantan",
        "last_name": "Bose",
        "email": "[email protected]",
        "phone_number": "6291753266",
        "salary": 850000.0
    }
}

Delete Particular Employee

To delete a particular employee we need to pass a param to the url and make a delete request to the endpoint ‘/delete_employee_by_email/’ so that the information of the particular employee is deleted from the database. To perform our testing we can use the following url and the email id is considered a unique attribute in our example.

http://localhost:8000/delete_employee_by_email/[email protected]

And the response returned is

{
    "message": "Employee with email [email protected] has been deleted."
}

So we have tested all our CRUD endpoints and all the enpoints are working correctly. This ensures that our code implementation is correct. The entire code can be found in the Github Repository.

Here is a small video demostration of the above example and setting up the DB connection.

In conclusion, we have successfully demonstrated how to set up a FastAPI application to perform CRUD operation, interact with a MongoDB database, and store data. MongoDB’s automatic generation of unique IDs for each request ensures data integrity and easy retrieval.


 


Reffered: https://www.geeksforgeeks.org

Python

Related
Competitors of Appium Competitors of Appium
Appium vs Selendroid Appium vs Selendroid
Placement Prediction App using Flask Placement Prediction App using Flask
Name and Rename Attributes in Pandas Name and Rename Attributes in Pandas
Limitation of Appium Limitation of Appium

Type:
 Geek
Category:
 Coding
Sub Category:
 Tutorial
Uploaded by:
 Admin
Views:
 12



 

Horje