DEV Community

Thiago da Silva Adriano
Thiago da Silva Adriano

Posted on

Python 3 + FastAPI + MongoDB

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.

Link for my previews article

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
Enter fullscreen mode Exit fullscreen mode

Now, to work with MongoDB, we need to install a package. In this article, we'll be using motor.

pip3 install motor
Enter fullscreen mode Exit fullscreen mode

With this steps ok, lets go to create a new Python file called 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"]
        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")

Enter fullscreen mode Exit fullscreen mode

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 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()"/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"       

Enter fullscreen mode Exit fullscreen mode

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)