Creating efficient, scalable APIs is essential for modern applications. In this blog post, we’ll walk through building two separate CRUD APIs following the Command Query Responsibility Segregation (CQRS) pattern using Python’s FastAPI framework. We’ll use MongoDB as our document database to persist data. This tutorial includes detailed setup instructions, complete code samples, and guidance on cloning and running the app from GitHub. So let us take a deep dive and begin our journey on all things, FastAPI CQRS CRUD API with MongoDB.
Why Use CQRS with FastAPI?
CQRS separates read and write operations into different models, improving scalability, maintainability, and performance. FastAPI is a modern, fast (high-performance) web framework for building APIs with Python 3.7+ based on standard Python type hints.
FastAPI CQRS CRUD API with MongoDB – Prerequisites
- Python 3.8 or higher installed
- MongoDB instance running locally or in the cloud (e.g., MongoDB Atlas)
- Git installed
Step 1: Setup Your Environment
Create a new project directory and navigate into it:
mkdir fastapi-cqrs-api
cd fastapi-cqrs-apiCreate and activate a virtual environment
python -m venv venv
source venv/bin/activate # On Windows use `venv\Scripts\activate`Install required packages:
pip install fastapi uvicorn motor pydanticStep 2: Define MongoDB Connection
Create a file database.py:
from motor.motor_asyncio import AsyncIOMotorClient
MONGO_DETAILS = "mongodb://localhost:27017" # Update with your MongoDB URI
client = AsyncIOMotorClient(MONGO_DETAILS)
database = client.fastapi_cqrs
# Collections for commands and queries
command_collection = database.get_collection("items_command")
query_collection = database.get_collection("items_query")Step 3: Define Pydantic Models
Create a file models.py:
from pydantic import BaseModel, Field
from typing import Optional
from bson import ObjectId
class PyObjectId(ObjectId):
@classmethod
def __get_validators__(cls):
yield cls.validate
@classmethod
def validate(cls, v):
if not ObjectId.is_valid(v):
raise ValueError("Invalid objectid")
return ObjectId(v)
@classmethod
def __modify_schema__(cls, field_schema):
field_schema.update(type="string")
class ItemBase(BaseModel):
name: str
description: Optional[str] = None
class ItemCreate(ItemBase):
pass
class ItemUpdate(BaseModel):
name: Optional[str]
description: Optional[str]
class ItemInDB(ItemBase):
id: PyObjectId = Field(default_factory=PyObjectId, alias="_id")
class Config:
allow_population_by_field_name = True
arbitrary_types_allowed = True
json_encoders = {ObjectId: str}
class ItemResponse(ItemBase):
id: str
class Config:
orm_mode = TrueStep 4: Create Command API (Write Operations)
Create commands.py:
from fastapi import APIRouter, HTTPException
from models import ItemCreate, ItemUpdate, ItemInDB
from database import command_collection
from bson import ObjectId
router = APIRouter(prefix="/commands", tags=["commands"])
@router.post("/items", response_model=ItemInDB)
async def create_item(item: ItemCreate):
item_dict = item.dict()
result = await command_collection.insert_one(item_dict)
created_item = await command_collection.find_one({"_id": result.inserted_id})
return created_item
@router.put("/items/{item_id}", response_model=ItemInDB)
async def update_item(item_id: str, item: ItemUpdate):
if not ObjectId.is_valid(item_id):
raise HTTPException(status_code=400, detail="Invalid item ID")
update_data = {k: v for k, v in item.dict().items() if v is not None}
if len(update_data) >= 1:
result = await command_collection.update_one({"_id": ObjectId(item_id)}, {"$set": update_data})
if result.modified_count == 1:
updated_item = await command_collection.find_one({"_id": ObjectId(item_id)})
return updated_item
existing_item = await command_collection.find_one({"_id": ObjectId(item_id)})
if existing_item:
return existing_item
raise HTTPException(status_code=404, detail="Item not found")
@router.delete("/items/{item_id}")
async def delete_item(item_id: str):
if not ObjectId.is_valid(item_id):
raise HTTPException(status_code=400, detail="Invalid item ID")
result = await command_collection.delete_one({"_id": ObjectId(item_id)})
if result.deleted_count == 1:
return {"message": "Item deleted successfully"}
raise HTTPException(status_code=404, detail="Item not found")
Step 5: Create Query API (Read Operations)
Create queries.py:
from fastapi import APIRouter, HTTPException
from typing import List
from models import ItemResponse
from database import query_collection
from bson import ObjectId
router = APIRouter(prefix="/queries", tags=["queries"])
@router.get("/items", response_model=List[ItemResponse])
async def list_items():
items = []
cursor = query_collection.find({})
async for document in cursor:
items.append(document)
return items
@router.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: str):
if not ObjectId.is_valid(item_id):
raise HTTPException(status_code=400, detail="Invalid item ID")
item = await query_collection.find_one({"_id": ObjectId(item_id)})
if item:
return item
raise HTTPException(status_code=404, detail="Item not found")Step 6: Main Application Setup
Create main.py:
from fastapi import FastAPI
from commands import router as command_router
from queries import router as query_router
app = FastAPI(title="CQRS FastAPI MongoDB Example")
app.include_router(command_router)
app.include_router(query_router)
@app.on_event("startup")
async def startup_event():
# Sync data from command collection to query collection
from database import command_collection, query_collection
await query_collection.delete_many({})
cursor = command_collection.find({})
async for document in cursor:
await query_collection.insert_one(document)
@app.get("/")
async def root():
return {"message": "Welcome to the CQRS FastAPI MongoDB example!"}Step 7: Running the Application
Run the app with Uvicorn:
uvicorn main:app --reloadStep 8: Version control and Github
- Set up your virtual environment and install dependencies as described in Step 1.
- Ensure MongoDB is running and accessible.
- Start the Command API:
git clone https://github.com/yourusername/fastapi-cqrs-crud.git
cd fastapi-cqrs-crud
pip install -r requirements.txt # in case of you have defined them
uvicorn command_api:app --reload --port 8000
uvicorn query_api:app --reload --port 8001If you’ve pushed this project to GitHub, cloning and running is simple:
The advantages of building the APIs using the CQRS pattern are,
- Separation of concerns: Write and read operations are isolated.
- Scalability: Each API can scale independently.
- Maintainability: Cleaner codebases and easier debugging.
My Experience with Python and Django
I have previously worked extensively with Python and Django, including developing an evidence rating app for researchers at the School of Psychology, University of Sydney. This involved building robust backend logic and user-friendly interfaces to support research workflows. Additionally, I developed a scheduling app for Corrugated Cartons at Visy Pty Ltd, which required integrating complex scheduling algorithms with Django’s ORM and REST framework.
FastAPI vs Django: A Detailed Comparison
FastAPI and Django are both powerful Python frameworks but serve different purposes:
- Performance: FastAPI is asynchronous and built on ASGI, making it faster and more suitable for high-performance APIs. Django is synchronous by default but can be extended with ASGI.
- Use Case: Django is a full-stack framework with built-in ORM, templating, and admin interface, ideal for monolithic web apps. FastAPI is focused on building APIs with automatic docs and validation.
- Learning Curve: Django has a steeper learning curve due to its many built-in features. FastAPI is more lightweight and easier to pick up for API development.
- Community and Ecosystem: Django has a mature ecosystem with many plugins and a large community. FastAPI is newer but rapidly growing.
- Flexibility: FastAPI offers more flexibility for microservices and modern async programming.
Conclusion
By following this guide, you now have a fully functional CQRS-based CRUD API using FastAPI and MongoDB. This architecture helps separate concerns, improve scalability, and maintain clean codebases. FastAPI’s modern features combined with MongoDB’s flexible document storage make a powerful stack for building APIs. Happy coding!
While you are here, maybe try one of my apps for the iPhone.
Snap! I was there on the App Store
If you enjoyed this guide, don’t stop here — check out more posts on AI and APIs on my blog (From https://mydaytodo.com/blog);
Build a Local LLM API with Ollama, Llama 3 & Node.js / TypeScript
Beginners guide to building neural networks using synaptic.js
Build Neural Network in JavaScript: Step-by-Step App Tutorial – My Day To-Do
Build Neural Network in JavaScript with Brain.js: Complete Tutorial
0 Comments