Skip to content

Use Case 1: Basic FastAPI CRUD API

Scenario: A simple user management API with CRUD endpoints, index registration, and a correct connection lifecycle.

📦 What's New?

Component Description
MongoConnectionManager.lifespan Connect on startup, disconnect on shutdown.
BaseRepository[User] Generic CRUD — no endpoint-level Mongo code.
ModelRegistry.register Declares the email index; initialize_all creates it at startup.

🚀 Example

Python
import os
from contextlib import asynccontextmanager

from fastapi import Depends, FastAPI, HTTPException
from pydantic import Field

from mongo_ops import BaseDocument, BaseRepository, ModelRegistry, MongoConnectionManager

# ---------------------------
# Model
# ---------------------------
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)


# ---------------------------
# Registration + index
# ---------------------------
ModelRegistry.register("users", User, indexes=[("email", 1)])


# ---------------------------
# Lifecycle
# ---------------------------
@asynccontextmanager
async def lifespan(_app: FastAPI):
    async with MongoConnectionManager.lifespan(
        uri=os.getenv("MONGO_URI", "mongodb://localhost:27017"),
        db_name="mydb",
    ):
        await ModelRegistry.initialize_all()
        yield


app = FastAPI(lifespan=lifespan)


# ---------------------------
# Dependency — created per request, after connect()
# ---------------------------
def get_user_repository() -> UserRepository:
    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"}

💡 Tips

  • UserRepository() is created inside the dependency, i.e., only after the lifespan has connected. Instantiating a repository at module import time raises RuntimeError because the database isn't connected yet.
  • BaseRepository.update(id, {...}) takes a dict; only PopulatingRepository.update takes a model.
  • Use patch() for REST PATCH semantics — it accepts a partial dict like update, but PopulatingRepository rejects FK fields.