|
FastAPI, a modern web framework for building APIs with Python, provides support for integrating with databases, allowing developers to create powerful and efficient applications. In this article, we’ll explore how to integrate databases with FastAPI, focusing specifically on MongoDB integration using PyMongo.
Integrating MongoDB with FastAPIMongoDB, a NoSQL database, is a popular choice for many developers due to its flexibility and scalability. Integrating MongoDB with FastAPI can be done using PyMongo, the official Python driver for MongoDB. Below is the step-by-step procedure by which we can integrate MongoDB with FastAPI using Python:
Step 1: Setting Up MongoDBBefore integrating MongoDB with FastAPI, you need to have MongoDB installed and running on your system or use a cloud-based solution like MongoDB Atlas. Ensure that you have the necessary permissions to create, read, update, and delete documents in your MongoDB database. For more information, refer to this: Install MongoDB
Step 2: Installing DependenciesYou’ll need to install the necessary Python packages for FastAPI, Pydantic, PyMongo, and any other dependencies required for your project. You can install these packages using pip:
pip install fastapi uvicorn pymongo Step 3: Creating ModelsDefine Pydantic models to represent the data structures that will be stored in MongoDB. These models will also handle data validation and serialization/deserialization.
Python
from pydantic import BaseModel
class Address(BaseModel):
city: str
country: str
class Student(BaseModel):
name: str
age: int
address: Address
Step 4: Connecting to MongoDBUse PyMongo to connect to your MongoDB database. Specify the connection URL and database name to establish a connection.
Python
from pymongo import MongoClient
# Connect to MongoDB
client = MongoClient("mongodb://localhost:27017")
db = client["library_management"]
students_collection = db["students"]
Step 5: Defining API EndpointsNow, define the API endpoints using FastAPI decorators. These endpoints will handle CRUD (Create, Read, Update, Delete) operations on the MongoDB database.
Python
from fastapi import FastAPI, HTTPException
app = FastAPI()
Step 6: Implementing CRUD OperationsImplement the API endpoints to perform CRUD operations on the MongoDB database. Use PyMongo’s methods (insert_one, find, update_one, delete_one, etc.) to interact with the database.
Python
@app.post("/students", status_code=201)
async def create_student(student: Student):
result = students_collection.insert_one(student.dict())
return {"id": str(result.inserted_id)}
@app.get("/students", response_model=list[Student])
async def list_students(country: str = None, age: int = None):
query = {}
if country:
query["address.country"] = country
if age:
query["age"] = {"$gte": age}
students = list(students_collection.find(query, {"_id": 0}))
return students
@app.get("/students/{id}", response_model=Student)
async def get_student(id: str):
student = students_collection.find_one({"_id": ObjectId(id)}, {"_id": 0})
if student:
return student
else:
raise HTTPException(status_code=404, detail="Student not found")
@app.patch("/students/{id}", status_code=204)
async def update_student(id: str, student: Student):
updated_student = student.dict(exclude_unset=True)
result = students_collection.update_one(
{"_id": ObjectId(id)}, {"$set": updated_student})
if result.modified_count == 0:
raise HTTPException(status_code=404, detail="Student not found")
else:
return
@app.delete("/students/{id}", status_code=200)
async def delete_student(id: str):
result = students_collection.delete_one({"_id": ObjectId(id)})
if result.deleted_count == 0:
raise HTTPException(status_code=404, detail="Student not found")
else:
return {"message": "Student deleted successfully"}
Complete Code Implementation
Python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from pymongo import MongoClient
from bson import ObjectId
app = FastAPI()
# Connect to MongoDB Atlas
client = MongoClient("mongodb://localhost:27017")
db = client["library_management"]
students_collection = db["students"]
class Address(BaseModel):
city: str
country: str
class Student(BaseModel):
name: str
age: int
address: Address
@app.post("/students", status_code=201)
async def create_student(student: Student):
result = students_collection.insert_one(student.dict())
return {"id": str(result.inserted_id)}
@app.get("/students", response_model=list[Student])
async def list_students(country: str = None, age: int = None):
query = {}
if country:
query["address.country"] = country
if age:
query["age"] = {"$gte": age}
students = list(students_collection.find(query, {"_id": 0}))
return students
@app.get("/students/{id}", response_model=Student)
async def get_student(id: str):
student = students_collection.find_one({"_id": ObjectId(id)}, {"_id": 0})
if student:
return student
else:
raise HTTPException(status_code=404, detail="Student not found")
@app.patch("/students/{id}", status_code=204)
async def update_student(id: str, student: Student):
updated_student = student.dict(exclude_unset=True)
result = students_collection.update_one(
{"_id": ObjectId(id)}, {"$set": updated_student})
if result.modified_count == 0:
raise HTTPException(status_code=404, detail="Student not found")
else:
return
@app.delete("/students/{id}", status_code=200)
async def delete_student(id: str):
result = students_collection.delete_one({"_id": ObjectId(id)})
if result.deleted_count == 0:
raise HTTPException(status_code=404, detail="Student not found")
else:
return {"message": "Student deleted successfully"}
Running the ApplicationFinally, run the FastAPI application using an ASGI server like Uvicorn.
uvicorn main:app --reload Output
|
