Skip to content

Error Handling

What can raise, what it means, and how to map it in a FastAPI app.

📋 Library-Raised Exceptions

Exception Source Meaning / fix
RuntimeError("Database not connected. Call connect() first.") get_database() / get_client() and any repository constructed first MongoConnectionManager.connect() hasn't run — wire the lifespan.
RuntimeError("No cache backend registered. Call set_cache_backend() first.") ModelRegistry.initialize_cache() Call set_cache_backend(backend) before initialize_cache().
KeyError("Model for collection '...' not registered") ModelRegistry.get_model() Collection was never registered (or typo).
ValueError("Cannot patch FK fields via patch(): ...") PopulatingRepository.patch() patch must not touch populated ref fields — use update() with a model.
ValueError("...contains embedded dict(s) — run repair script") _populate on read A FK field holds an embedded document instead of an ObjectId — migrate the data.
ValueError("...contains ObjectId — was populate skipped?") _depopulate on write A populate-ruled field is still an ObjectId at depopulate time — the read must have populated it first.
CircularReferenceError(collection, doc_id, path) PopulationEngine.populate A (Class, id) pair was revisited — raise max_depth or fix the graph.
ImportError("redis package required ... mongo-ops[redis]") CacheConfig / RedisCacheBackend Missing redis extra.
ValueError("redis_client required when backend='redis'") CacheConfig backend="redis" without a client.
Pymongo DuplicateKeyError any insert/update Unique index violation (e.g., duplicate email).
bson.errors.InvalidId ObjectId(...) on a bad string Wrapped by PyObjectId model validation on API inputs.

KeyError for ModelRegistry.get_model and the RuntimeError/ValueError guards are by design — they fail loudly at startup or first call instead of misbehaving silently.

🚀 FastAPI Mapping Example

Python
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from bson.errors import InvalidId
from pymongo.errors import DuplicateKeyError
from mongo_ops.cache import CircularReferenceError


app = FastAPI()


@app.exception_handler(DuplicateKeyError)
async def duplicate_key_handler(_: Request, __: DuplicateKeyError) -> JSONResponse:
    return JSONResponse(status_code=409, content={"detail": "Resource already exists"})


@app.exception_handler(InvalidId)
async def invalid_id_handler(_: Request, __: InvalidId) -> JSONResponse:
    return JSONResponse(status_code=400, content={"detail": "Invalid ID format"})


@app.exception_handler(CircularReferenceError)
async def circular_ref_handler(_: Request, exc: CircularReferenceError) -> JSONResponse:
    return JSONResponse(
        status_code=409,
        content={"detail": f"Circular reference detected: {exc.path}"},
    )


# Or handle inline for route-specific responses:
@app.post("/users/")
async def create_user(user: User, repo=Depends(get_user_repository)):
    try:
        return await repo.create(user)
    except DuplicateKeyError:
        raise HTTPException(status_code=409, detail="User already exists")
    except Exception as exc:
        raise HTTPException(status_code=500, detail=str(exc))


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

Prefer exception handlers for the library-level exceptions (409/400 above) and per-endpoint try/except for domain decisions (404).

💡 Tips

  • The ValueError populate guards are your friends: they surface data-shape drift (embedded docs, skipped population) at the exact call site.
  • In dev, log the CircularReferenceError.path — it prints the visited Class:id chain.
  • Never swallow RuntimeErrors at startup; let the app fail to load so the misconfiguration is obvious.