Basic CRUD

Use Case 1: Basic FastAPI CRUD API

Scenario: Create a simple user management API with CRUD operations.

Python

import os
from fastapi import FastAPI, Depends, HTTPException
from contextlib import asynccontextmanager
from pydantic import Field
from mongo_ops import (
    MongoConnectionManager,
    BaseDocument,
    BaseRepository,
    ModelRegistry,
)

# ---------------------------
# Model Definition
# ---------------------------
class User(BaseDocument):
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(...)
    is_active: bool = True


# ---------------------------
# Repository
# ---------------------------
class UserRepository(BaseRepository[User]):
    def __init__(self):
        super().__init__("users", User)


# Register the model only once
ModelRegistry.register("users", User, indexes=[("email", 1)])


# ---------------------------
# FastAPI Lifespan
# ---------------------------
@asynccontextmanager
async def lifespan(_app: FastAPI):
    """Manage MongoDB connection during the app lifecycle."""
    async with MongoConnectionManager.lifespan(
        uri="mongodb://localhost:27017",
        db_name="mydb"
    ):
        await ModelRegistry.initialize_all()
        yield


app = FastAPI(lifespan=lifespan)


# ---------------------------
# Dependency Injection
# ---------------------------
def get_user_repository() -> UserRepository:
    """Dependency-injected repository for users."""
    return UserRepository()


# ---------------------------
# Routes
# ---------------------------
@app.post("/users/", response_model=User)
async def create_user(user: User, repo: UserRepository = Depends(get_user_repository)):
    return await repo.create(user)


@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: str, repo: UserRepository = Depends(get_user_repository)):
    user = await repo.get_by_id(user_id)
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@app.get("/users/", response_model=list[User])
async def list_users(
    skip: int = 0,
    limit: int = 10,
    repo: UserRepository = Depends(get_user_repository),
):
    return await repo.get_many(skip=skip, limit=limit)


@app.put("/users/{user_id}", response_model=User)
async def update_user(
    user_id: str,
    email: str,
    repo: UserRepository = Depends(get_user_repository),
):
    user = await repo.update(user_id, {"email": email})
    if not user:
        raise HTTPException(status_code=404, detail="User not found")
    return user


@app.delete("/users/{user_id}")
async def delete_user(
    user_id: str,
    repo: UserRepository = Depends(get_user_repository),
):
    deleted = await repo.delete(user_id)
    if not deleted:
        raise HTTPException(status_code=404, detail="User not found")
    return {"message": "User deleted successfully"}