Building on my previous article where I showed how somente can create a simple API using Python and FastAPI. In this article, we'll explore Python and FastAPI by incorporating persistence with MongoDB.
For the purposes of this tutorial, I'll demonstrate how to use MongoDB with Docker.
The initial step in configuring MongoDB with Docker involves verifying that Docker is operational on your system. After confirming Docker is active, you can move forward by creating a MongoDB database container. Execute the following command in your terminal to do so:
docker run --name mongo-local -d -p 27017:27017 mongo
Now, to work with MongoDB, we need to install a package. In this article, we'll be using motor.
pip3 install motor
With this steps ok, lets go to create a new Python file called db.py to create our repository:
from motor.motor_asyncio import AsyncIOMotorClient
from bson import ObjectId
class DBManager:
def __init__(self, uri: str, database_name: str):
self.client = AsyncIOMotorClient(uri)
self.db = self.client[database_name]
async def create_item(self, item):
result = await self.db.items.insert_one(item)
return await self.read_item(result.inserted_id)
async def read_item(self, item_id: ObjectId):
item = await self.db.items.find_one({"_id": item_id})
if item:
item["id"] = str(item["_id"])
del item["_id"]
return item
async def read_items(self) -> list:
items = []
cursor = self.db.items.find({})
async for item in cursor:
item["id"] = str(item["_id"])
del item["_id"]
items.append(item)
return items
async def update_item(self, item_id: ObjectId, item):
await self.db.items.update_one({"_id": item_id}, {"$set": item})
return await self.read_item(item_id)
async def delete_item(self, item_id: ObjectId) -> bool:
result = await self.db.items.delete_one({"_id": item_id})
return result.deleted_count > 0
db_manager = DBManager(uri="mongodb://localhost:27017", database_name="db_local")
In this code, we are creating a class called DBManager, within which we have developed five methods: create_item, read_item, read_items, update_item, and delete_item.
Now we can update our main.py file with this code:
from bson import ObjectId
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
from uuid import UUID, uuid4
from db import db_manager
from typing import List
class Item(BaseModel):
name: str
description: Optional[str] = None
price: float
on_offer: bool = False
app = FastAPI()
@app.post("/items/", response_model=Item)
async def create_item(item: Item):
new_item = await db_manager.create_item(item.model_dump())
return new_item
@app.get("/items/", response_model=List[Item])
async def read_items():
return await db_manager.read_items()
@app.get("/items/{item_id}", response_model=Item)
async def read_item(item_id: str):
item = await db_manager.read_item(ObjectId(item_id))
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return item
@app.put("/items/{item_id}", response_model=Item)
async def update_item(item_id: str, item: Item):
updated_item = await db_manager.update_item(ObjectId(item_id), item.model_dump())
if updated_item is None:
raise HTTPException(status_code=404, detail="Item not found")
return updated_item
@app.delete("/items/{item_id}", response_model=str)
async def delete_item(item_id: str):
deleted = await db_manager.delete_item(ObjectId(item_id))
if not deleted:
raise HTTPException(status_code=404, detail="Item not found")
return "Item deleted successfully"
In this code, we just add db_manager to store all datas in mongodb instead to store in memory.
To test your project, execute the following command in your terminal: uvicorn main:app --reload.
This will allow you to test all actions in Swagger and see the data stored at your mongoDB.
Top comments (0)