Skip to content

Use Case 9: Nested Document Population & Circular-Ref Handling

Scenario: An Author has books (a list of references), each Book references a publisher, and an Author may reference a mentor — which is another Author (a potential cycle).

📦 What's New?

Component Description
nested_rules Populates deeper levels: resolve books on Author, then publisher inside each Book.
max_depth Per-rule recursion bound — the safety net for cyclic graphs.
global_max_depth PopulationEngine(repos, global_max_depth=10) global cap.
CircularReferenceError Raised when a (Class, id) pair is revisited; carries the visited path.

🚀 Example

Python
from contextlib import asynccontextmanager

from bson import ObjectId
from fastapi import FastAPI, HTTPException
from mongo_ops import BaseDocument, ModelRegistry, MongoConnectionManager, PopulatingRepository
from mongo_ops.cache import CircularReferenceError
from mongo_ops.populate import PopulateRule, PopulationEngine


# 1. Models — each ref field holds ObjectId(s) in the DB and becomes model(s) in memory.
class Publisher(BaseDocument):
    name: str = ""
    country: str = ""


class Book(BaseDocument):
    title: str = ""
    publisher: Publisher | None = None  # ObjectId in DB, Publisher in memory


class Author(BaseDocument):
    name: str = ""
    books: list[Book] | None = None      # list[ObjectId] in DB, list[Book] in memory
    mentor: "Author" | None = None        # self-reference — potential cycle


# 2. Engine + nested rules.
engine = PopulationEngine({})

publisher_rule = PopulateRule(
    field_name="publisher",
    collection_name="publishers",
)

book_rule = PopulateRule(
    field_name="books",
    collection_name="books",
    nested_rules=[publisher_rule],  # fetch each book, then its publisher
    max_depth=3,
)

mentor_rule = PopulateRule(
    field_name="mentor",
    collection_name="authors",
    max_depth=2,  # stops mentor chains early — also avoids unbounded cycles
)


class AuthorRepository(PopulatingRepository[Author]):
    def __init__(self):
        super().__init__(
            collection_name="authors",
            model=Author,
            population_engine=engine,
            populate_rules=[book_rule, mentor_rule],
        )


@asynccontextmanager
async def lifespan(_app: FastAPI):
    async with MongoConnectionManager.lifespan(
        uri="mongodb://localhost:27017", db_name="library"
    ):
        engine.register_repo("publishers", PopulatingRepository[Publisher]("publishers", Publisher))
        engine.register_repo("books", PopulatingRepository[Book]("books", Book))
        engine.register_repo("authors", AuthorRepository())
        await ModelRegistry.initialize_all()
        yield


app = FastAPI()


@app.get("/authors/{author_id}")
async def get_author(author_id: str):
    try:
        author = await AuthorRepository().get_by_id(author_id)
    except CircularReferenceError as exc:
        raise HTTPException(status_code=409, detail=f"Circular reference: {exc}")
    if not author:
        raise HTTPException(status_code=404, detail="Author not found")
    return author

💡 Tips

  • CircularReferenceError is a ValueError subtype exposing .collection, .doc_id, and the visited .path — use it in error responses and logging.
  • Keep max_depth conservative (2–3) for most graphs; combine with the global global_max_depth=10 default.
  • List refs depopulate back to list[ObjectId] on create/update, mirroring the scalar case.