Introduction

In this tutorial, we will dive into the process of building a powerful API using FastAPI and Motor, an asynchronous MongoDB driver.

FastAPI provides a straightforward way to create web APIs with incredible performance, while Motor enables seamless integration with MongoDB.

By the end of this tutorial, you will have a solid understanding of how to set up the environment, integrate MongoDB with FastAPI, create complex models, and perform CRUD operations efficiently.

I will walk you through creating a API using FastAPI and Motor(Async PyMongo)

Table of Contents

📍Setting up the Environment

📍Integrate MongoDB with FastAPI

📍Creating Complex Student Model

📍Performing CRUD Operations

Installation

Before we begin, make sure to install FastAPI and Motor by running the following commands in your terminal✅:

pip install fastapi 
pip install motor 

Setting up the Environment

To get started, make sure to spin up a MongoDB M0 cluster, which is completely free and doesn’t require a credit card.

To integrate FastAPI with MongoDB, we need to establish a connection between the two. Use the following code:

import motor.motor_asyncio
from fastapi import FastAPI

app = FastAPI()

uri = "mongodb+srv://<username>:<password>@task3.lnfrkc5.mongodb.net/?
retryWrites=true&w=majority" 

# Create a new client and connect to the server 
client = motor.motor_asyncio.AsyncIOMotorClient(uri) 

# Send a ping to confirm a successful connection 
try:     
    client.admin.command('ping')     
    print("Pinged your deployment. You have successfully connected to 
MongoDB!") 
except Exception as e:     
    print(e) 

⚠️ Remember to replace <username> and <password> with your actual MongoDB credentials. Ensure that the ping to the database is successful, as this confirms the integration between the backend and the database.

Creating the Pydantic Standard Schema

Next, we will define the Pydantic standard schema, which represents the structure of our data. Here’s an example of how the JSON representation of the schema should look:

{ 
  "_id": "", 
  "name": "Suraj", 
  "age": 21, 
  "address": { 
        "city": "Maihar", 
        "country": "India" 
  }, 
  "phoneNum": ["9898989989", "8989898989"], 
  "socials": [ 
        { 
           "social_type": "Github", 
           "link": "https://github.com/suraj7879" 
        }, 
        { 
           "social_type": "LinkedIn", 
           "link": "https://linkedin.com/in/suraj7879" 
        } 
    ] 
} 

* Please note that the _id field will be automatically generated by MongoDB.

Many people think of MongoDB as being schema-less, which is wrong❌MongoDB has a flexible schema✅

To ensure the validity of the data, we will make use of Pydantic. Define the following models:

from pydantic import BaseModel, Field 
from typing import List 

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

class Socials(BaseModel):     
    type: str     
    link: str class 

Student(BaseModel):     
    name: str = Field(..., min_length=3, max_length=40)     
    age: int = Field(..., gt=0)     
    address: Address     
    phoneNum: List[str]     
    socials: List[Socials] 

‼️ Will it be able to process if there are n number of social profiles ?

CRUD Operations:

Create a Student

@app.post('/student/') 
async def create_student(student: Student): 
    try: 
        student = student.dict() 
        result = await collection.insert_one(student) 
        print(result) 
    except Exception as e: 
        print(e) return {"message": "Success"} 

⚠️ Remember to convert the Student object into a dictionary before passing it to the MongoDB functions(i.e. insert_one)

Read a Student

from bson import ObjectId 
... 
... 
@app.get('/student/{id}') 
async def read_student(id: str): 
    student = await collection.find_one({"_id": ObjectId(id)}) 
    if student: student['_id'] = str(student['_id']) 
        return student 
    else: 
        return {"message": "Student not found"} 

⚠️ Make sure to convert id parameter into ObjectId before before querying the MongoDB collection

Update a Student

@app.put('/student/{id}') 
async def update_student(id: str, updated_fields: dict): 
    await collection.update_one({"_id": ObjectId(id)}, {"$set": updated_fields}) 
    return {"message": "Student updated"} 

⚠️ Instead of object, our function is accepting a dictionary called updated_fields. The updated_fields dictionary should contain the specific fields you want to update and their new values.

Delete a Student

@app.delete('/student/{id}') 
async def delete_student(id: str): 
    student = await collection.find_one({"_id": ObjectId(id)}) 
    if student: 
        await collection.delete_one({"_id": ObjectId(id)}) 
        return {"message": "Student deleted"} 
    else: 
      return {"message": "Student not found"}

‼️Think, what will happen if we wont write await when calling async functions?

Running the Backend:

To run the backend, use the following command in your terminal:

uvicorn main:app --reload 

FastAPI provides built-in documentation, which includes interactive API documentation and the ability to test your API’s operations using a user-friendly GUI. Access it through the following URL:

http://127.0.0.1:8000/docs 

Wrapping Up:

Congratulations! You have learned the basics of FastAPI, integration with MongoDB, and some of the great features FastAPI provides. You can now build robust APIs with ease, ensuring high performance and efficiency.

If you like this short introduction, do ❤️ and Repost 🔃 with your network! 🍀

Disclaimer: This post was originally written over linkedin by Me, sharing it in NamasteDev Community, if this get good amount of support, will come up with more exclusive content for this community