Profile

Unframed AI Backend

The Beauty of being an AI Backend Engineer

The Adaptable Backend: Python REST API with MongoDB persistence

By Hendrix Roa
December 9, 2025
5 min read
Posted in The Adaptable Backend: Python
Tags: backend software architecture agility tech explained persistence database python mongodb
The Adaptable Backend: Python REST API with MongoDB persistence

In the previous article we created the foundations to implement a database engine. The IRepository set the direction of how we can interact with the technologies and database engine.

We implement the PostgreSQL but we could have used MySQL. In this opportunity we will integrate the NoSQL approach to demonstrate that not only can we switch to a db engine to another, also that good practices, separations of concerns and software engineering fundamentals play an important role.

Common Criticisms of NoSQL

NoSQL is the next replacement of SQL

Not really, NoSQL comes to play to start thinking the projects in terms of scalability and as an extension of Relational Model capabilities.

NoSQL is slow

Poor queries and bad data modeling in NoSQL is an indicator of bad performance.

Is MongoDB use an ORM?

In Python, we have ODMs (Object Document Mappers) like Beanie or MongoEngine. However, to keep our architecture lightweight and adaptable, we will use the official async driver: PyMongo.

Installing MongoDB using Docker compose

In this article I will use the MongoDB official database to start creating our CRUD of Note entity, the docker image package is on the mongodb_registry. The docker-containers will be managed with Docker Compose, which requires creating a compose.yml file. MongoDB docker image requires a SSL to start the database engine to keep everything Encrypted at Rest so we will extend their image using a docker file. To keep everything separated in the root folder lets create a folder containers/mongodb and the Dockerfile for mongodb extension.

The Dockerfile contains the next content:

containers/mongodb/Dockerfile
FROM mongo:latest
RUN openssl rand -base64 756 > /etc/mongo-keyfile
RUN chmod 400 /etc/mongo-keyfile
RUN chown mongodb:mongodb /etc/mongo-keyfile

Which run openssl command to generate the key that will use mongodb engine to start. Then we are able to use that docker image in the compose.yml

compose.yml
services:
  mongodb:
    build:
      context: ./containers/mongodb
      dockerfile: Dockerfile
    restart: always
    environment:
       MONGO_INITDB_ROOT_USERNAME: mongo
       MONGO_INITDB_ROOT_PASSWORD: mongo
       MONGO_INITDB_DATABASE: mongo
    ports:
      - 27017:27017
    command: --replSet rs0 --keyFile /etc/mongo-keyfile --bind_ip_all --port 27017
    healthcheck:
      test: echo "try { rs.status() } catch (err) { rs.initiate({_id:'rs0',members:[{_id:0,host:'127.0.0.1:27017'}]}) }" | mongosh --port 27017 -u mongo -p mongo --authenticationDatabase admin
      interval: 5s
      timeout: 15s
      start_period: 15s
      retries: 10
    volumes:
      - "mongo_data:/data/db"
      - "mongo_config:/data/configdb"
    networks:
      - adaptable_backend_python


volumes:
  mongo_data:
  mongo_config:

Running the mongodb service

Running the command:

Terminal window
make docker-up

Integrate MongoDB to our Python project

Install the PyMongo driver:

Terminal window
uv add pymongo

The MongoDB Repository

Implementing the IRepository using the mongodb client.

src/core/database/nosql/nosql_repository.py
from typing import TypeVar, List, Optional, Any
from pymongo import AsyncMongoClient
from bson import ObjectId
from datetime import datetime
from src.core.configuration.configuration import config
from src.core.database.i_repository import IRepository


T = TypeVar("T")


class NoSQLRepository(IRepository[T]):
    def __init__(self, collection_name: str):
        self.collection_name = collection_name
        self.client = AsyncMongoClient(config.get("DATABASE_URL"))
        self.db = self.client.get_database()
        self.collection = self.db[self.collection_name]


    async def create(self, data: dict) -> dict:
        data["created_at"] = datetime.now()
        data["updated_at"] = datetime.now()
        result = await self.collection.insert_one(data)
        data["id"] = str(result.inserted_id)
        return data


    async def find_all(self) -> List[dict]:
        results = []
        async for document in self.collection.find({}):
            document["id"] = str(document.pop("_id"))
            results.append(document)
        return results


    async def find_by_id(self, id: str) -> Optional[dict]:
        try:
            document = await self.collection.find_one({"_id": ObjectId(id)})
        except Exception:
            return None


        if document:
            document["id"] = str(document.pop("_id"))
            return document
        return None

Adding the database mongodb environment variables

Update .env:

.env
# Database SQL
#DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/postgres
#DATABASE_ENGINE=postgres


# Database NoSQL
DATABASE_URL=mongodb://mongo:mongo@localhost:27017
DATABASE_ENGINE=mongodb

Creating the seeders

We can create a seeder script using PyMongo.

src/core/database/nosql/seeder.py
import asyncio
from pymongo import AsyncMongoClient
from src.core.configuration.configuration import config


class NoSQLSeeder:


    @staticmethod
    async def run():
        print("Running NoSQL seeders...")
        client = AsyncMongoClient(config.get("DATABASE_URL"))
        db = client.get_database()
        collection = db["notes"]


        # Clear existing
        await collection.delete_many({})


        notes = [
            {"content": "The only limit to our realization of tomorrow is our doubts of today.", "times_sent": 0},
            {"content": "Do what you can, with what you have, where you are.", "times_sent": 0},
            {"content": "The best way to predict the future is to invent it.", "times_sent": 0}
        ]


        await collection.insert_many(notes)
        print("Seeded MongoDB")
        await client.close()


if __name__ == "__main__":
    asyncio.run(NoSQLSeeder.run())

Updating the NoteRepository

Now we update our Strategy Pattern to include the NoSQL engine.

src/core/features/note/note_repository.py
from src.core.database.nosql.nosql_repository import NoSQLRepository


class NoteRepository:
    def __init__(self):
        db_engine = config.get("DATABASE_ENGINE")


        if db_engine == DatabaseEngine.SQL:
            self.repository = SQLRepository("notes")
        elif db_engine == DatabaseEngine.NOSQL:
            self.repository = NoSQLRepository("notes")

Updating the Note Entity with DTOs

To manage the data transfer objects clearly, we separate the creation and update payloads from the main entity (which includes the ID).

src/core/features/note/note.py
from typing import Union
from pydantic import BaseModel, Field


class CreateNoteDto(BaseModel):
    content: str = Field(..., example="Hello World")


class UpdateNoteDto(CreateNoteDto):
    pass


class Note(CreateNoteDto):
    id: Union[int, str] = Field(..., example=1)

Updating the NoteController

Now we need to update the controller to use the CreateNoteDto and UpdateNoteDto to strictly define the input for creation and updates.

src/core/features/note/note_controller.py
from .note import Note, CreateNoteDto, UpdateNoteDto
from .note_repository import NoteRepository


class NoteController:
    def __init__(self, note_repository: NoteRepository):
        self.note_repository = note_repository


    async def get_notes(self) -> list[Note]:
        return await self.note_repository.find_all()


    async def create_note(self, note: CreateNoteDto) -> Note:
        return await self.note_repository.create(note.dict())


    async def get_note(self, id: str) -> Note:
        return await self.note_repository.find_by_id(id)


    async def update_note(self, id: str, note: UpdateNoteDto) -> Note:
        return await self.note_repository.update(id, note.dict())


    async def delete_note(self, id: str) -> bool:
        return await self.note_repository.delete(id)

Conclusion

In this example switching between SQL to NoSQL approach just by using a few design patterns and good practices.

In the inference and AI world a type of database can coexist in the same backend application, we need to deep dive into the needs of the project and take strategic decisions that make the backend easy to change in the long-term.

Have you ever faced similar issues when required to implement a database engine at your Back-end applications? - Please let me know in the comments.

Join to Newsletter

Subscribe to get our latest content by email and read at your own pace

    We respect your privacy. Unsubscribe at any time.

    Comments

    Loading comments...

    You Might Also Like