Scenario: You want to see what populate/depopulate actually do to your documents — field by field, at every layer (Python model → depopulated → stored in Mongo → raw read back → populated).
+
These examples all ran against the real mongo_ops engine (mongo_ops/populate/engine.py). Timestamps and ObjectIds are abbreviated with … for readability.
+
+
Models & rules used below
+
Python
frombsonimportObjectId
+frommongo_opsimportBaseDocument,PopulatingRepository
+frommongo_ops.populateimportPopulateRule,PopulationEngine
+
+classProfile(BaseDocument):
+avatar_url:str=""
+bio:str=""
+
+classPublisher(BaseDocument):
+name:str=""
+
+classBook(BaseDocument):
+title:str=""
+publisher:Publisher|None=None# nested ref, reached via nested_rules
+
+classUser(BaseDocument):
+name:str=""
+profile:Profile|None=None# scalar FK — DB stores an ObjectId
+books:list[Book]|None=None# list FK — DB stores list[ObjectId]
+
+engine=PopulationEngine({})# repos keyed by collection name, see UC 15
+
+# A rule WITHOUT nested_rules — the write path is safe.
+profile_rule=PopulateRule(field_name="profile",collection_name="profiles")
+
+# A rule WITH nested_rules — designed for READ-side deep population.
+books_rule=PopulateRule(
+field_name="books",
+collection_name="books",
+nested_rules=[PopulateRule(field_name="publisher",collection_name="publishers")],
+)
+
+
+
1. Write path — depopulate() turns models into references
+
engine.depopulate(document, rules) walks each rule field and mutates the model in place before you insert_one / find_one_and_update it. What you get on the other side:
+
1a. Scalar FK, no nested_rules → stored as ObjectId
+
Python
asyncdefdemo():
+profile=Profile(id=ObjectId("656…04"),avatar_url="a.png",bio="hi")
+user=User(id=ObjectId("656…05"),name="Ada",profile=profile)
+
+awaitengine.depopulate(user,[profile_rule])
+
+returnuser.profile# ObjectId('656…04') — was a Profile model
+
+
What Mongo receives (via model_dump, minus _id handling):
+
JSON
{"name":"Ada","profile":{"$oid":"656…04"}}
+
+
1b. Scalar FK withnested_rules → reference lost (do not do this)
+
Python
profile_rule_deep=PopulateRule(# same field, but with nested_rules
+field_name="profile",
+collection_name="profiles",
+nested_rules=[PopulateRule(field_name="bio",collection_name="profiles")],# any child field
+)
+
+asyncdefdemo():
+user=User(id=ObjectId("656…05"),name="Ada",
+profile=Profile(id=ObjectId("656…04"),avatar_url="a.png"))
+awaitengine.depopulate(user,[profile_rule_deep])
+returnuser.profile# None — reference silently dropped!
+
+
Because depopulate recurses into the child but then assigns the parent field None (see engine.py — depopulated_value is never set on that branch). On save you lose the reference entirely. There is no warning — the field is just gone.
+
1c. List FK, no nested_rules → stored as list[ObjectId] (the safe pattern)
+
Python
asyncdefdemo():
+b1=Book(id=ObjectId("656…06"),title="MongoDB in Action")
+b2=Book(id=ObjectId("656…07"),title="MongoDB: The Definitive Guide")
+user=User(id=ObjectId("656…08"),name="Ada",books=[b1,b2])
+
+awaitengine.depopulate(user,[PopulateRule(field_name="books",collection_name="books")])
+
+returnuser.books# [ObjectId('656…06'), ObjectId('656…07')] — clean refs
+
+
1d. List FK withnested_rules → stored as embedded documents (do not do this)
books stays a list of Book models — the nested publisher collapsed to ObjectId, but the Book itself was left embedded. Serialized to Mongo you get a nested sub-document:
+
JSON
{"name":"Ada","books":[{"title":"MongoDB in Action","publisher":{"$oid":"656…09"}}]}
+
+
…and reading that back through population raises ValueError(…contains embedded dict(s) — run repair script…), because the engine expects books to hold references, not embedded docs.
+
+
Why UC 09's Author → books → publisher graph can't be written as-is: a rule with nested_rules breaks the write path (1b logs silently, 1d stores embedded). The safe round-trip: save referenced documents first, store list[ObjectId]/ObjectId with a non-nested rule for reads, and keep nested_rules only on rules you never hand back to depopulate.
+
+
+
2. Read path — populate() turns references back into models
+
PopulatingRepository.data_to_model(raw_dict) resolves references before building the model. It first runs _populate(dict) — replacing the raw ObjectId references with full models via the registered repos — and only then constructs self.model(**data), so every FK field is a real model by construction. The two data_to_model examples below (2a/2b) show the underlying resolution _populate performs per rule. (repo below is a PopulatingRepository wired per UC 15 — books and publishers repos registered in its PopulationEngine.)
+
2a. Scalar FK — ObjectId → Profile model
+
Python
# what came back from Mongo (raw dict, profile is an ObjectId reference):
+raw={"_id":ObjectId("656…05"),"name":"Ada","profile":ObjectId("656…04")}
+
+asyncdefdemo():
+user=awaitrepo.data_to_model(raw)# _populate() → model(**data)
+returnuser.profile# Profile(id=ObjectId('656…04'), avatar_url='a.png')
+
+
2b. List FK + nested_rules — two levels deep
+
Python
raw={"_id":ObjectId("656…08"),"name":"Ada",
+"books":[ObjectId("656…06"),ObjectId("656…07")]}
+
+asyncdefdemo():
+user=awaitrepo.data_to_model(raw)# _populate() resolves both levels
+return(type(user.books[0]).__name__,# 'Book'
+type(user.books[0].publisher).__name__,# 'Publisher' — nested_rules reached it
+user.books[0].publisher.name)# "O'Reilly"
+
+
Walking through one book's resolution:
+
Text Only
books_rule → field "books", collection "books"
+ nested_rules[0].publisher rule → field "publisher", collection "publishers"
+1. for ObjectId('656…06') → repo("books").get_by_id(...) → Book("MongoDB in Action")
+2. book has nested rule → repo("publishers").get_by_id(...) → Publisher("O'Reilly")
+3. author.books[0] → Book(publisher=Publisher("O'Reilly"))
+
+
Per-level cost: 1 query per reference (no $lookup). A hub with 20 hooks and a publisher nested rule is 40 queries — that is why the next section exists.
+
+
3. Cache round trip — what actually goes into the cache
+
CachedBaseRepository stores model_dump(by_alias=True) JSON-encoded with default=str (the encode_value helper in mongo_ops/cache/in_memory.py). Crucially it snapshots the raw, just-out-of-Mongo shape — FK fields still hold ObjectIds — so the cache holds references, exactly like the DB row:
+
+
+
+
Layer
+
profile field looks like
+
+
+
+
+
Mongo doc (raw)
+
“profile”: ObjectId("656…04")
+
+
+
model_dump(by_alias=True)
+
"profile": ObjectId("656…04")
+
+
+
encode_value(...)
+
"...\"profile\": \"656…04\"..." — hex string
+
+
+
decode_value(...)
+
"profile": "656…04" — still a hex string
+
+
+
+
Verbatim (timestamps abbreviated):
+
Python
frommongo_ops.cache.in_memoryimportdecode_value,encode_value
+
+snapshot={"id":"656…05","profile":ObjectId("656…04")}# what a cached raw doc looks like
+
+cached=encode_value(snapshot)# json.dumps(snapshot, default=str).encode()
+decoded=decode_value(cached)
+
+decoded["profile"]# '656…04' — hex STRING, not ObjectId, not Profile
+
+
+
Typed-FK caveat: the round trip above only works when the FK field's model type accepts the raw value (str/ObjectId typed). A field typed Profile | None cannot even be materialized from the raw Mongo doc — both hit and miss raise ValidationError (UC 16's intricacy). That is exactly why UC 10's composed repo overrides get_by_id/create/update to carry raw data through the cache and resolve refs with _populate.
+
+
Two consequences (the "intricacies", detailed in UC 16):
+
+
On a cache hit, decode_value gives you hex-string references. A field typed Profile | None receiving a hex string raises Pydantic ValidationError — so a plain CachedBaseRepositorycannot serve model-typed FK fields, hit or miss. The composed pattern in UC 10 re-resolves them in its get_by_id: data = decode_value(await cache.get(key)), then populated = await self._populate(data), then self.model(**populated).
+
On a miss, BaseRepository.get_by_id reads the raw Mongo doc whose profile is an ObjectId and immediately raises the same ValidationError — which is why UC 10 overrides get_by_id to fetch raw, cache raw, then populate. (Real ObjectIds are covered by UC 16's table.)
+
+
+
4. When to use which
+
+
+
+
You need…
+
Use
+
+
+
+
+
Scalar ObjectId store, model-struct reads
+
PopulatingRepository (UC 08)
+
+
+
Deep nested reads (2+ levels)
+
PopulatingRepository + nested_rules (UC 09) — read side only
+
+
+
Same shape, but cache-first reads
+
UC 10 composed repo (engine + cache + _populate)
+
+
+
Scalar docs, no refs, hot get_by_id
+
CachedBaseRepository (UC 07, UC 16)
+
+
+
Writing a reference graph back
+
depopulate withoutnested_rules — save children first, then point at their ids
mongo-ops is a modular, high-performance MongoDB operations library for FastAPI microservices. It standardizes connection lifecycle, Pydantic v2 document models, and repository-style async CRUD \u2014 with added layers for caching, reference population, and multi-document transactions. It is built on top of Motor.
Doc model: this wiki is written for humans \u2014 how\u2011to guides, examples, and testing recipes. The authoritative API contracts live in the code (docstrings) and the machine\u2011readable bundle under docs/mcp/.
mongo-ops gives FastAPI/microservice teams a small, opinionated toolkit for talking to MongoDB asynchronously. This page builds the mental model: what the layers are, in what order they must be wired up, and which repository to reach for.
"},{"location":"01_overview/#architecture","title":"\ud83c\udfd7\ufe0f Architecture","text":"Text Only
mongo-ops has a strict startup order. Violating it raises fast, loud exceptions:
Step Call Why 1 await MongoConnectionManager.connect(uri, db_name) Without a connection, get_database()/get_client() raise RuntimeError(\"Database not connected...\"). 2 await ModelRegistry.initialize_all() Create registered indexes (idempotent via create_index). 3 await ModelRegistry.set_cache_backend(...) then await ModelRegistry.initialize_cache() Must happen after connection, before any cache-backed operation. Raises RuntimeError(\"No cache backend registered...\") if skipped. 4 Use repositories Constructed repos resolve the collection from the live database. 5 await ModelRegistry.shutdown_cache() + await MongoConnectionManager.disconnect() On shutdown (in-memory TTL task cancelled; Redis pub/sub closed).
The canonical wiring is the FastAPI lifespan context manager (see use case 01).
"},{"location":"01_overview/#which-repository-should-you-use","title":"\ud83e\udded Which Repository Should You Use?","text":"Repository Use when Adds BaseRepository[T] Plain CRUD \u2014 the default nothing extra CachedBaseRepository[T] Read\u2011heavy, low\u2011write fields (lookups by _id) cache\u2011first get_by_id, cache on create, invalidate on update/delete, warm_cachePopulatingRepository[T] You return related docs (FK references) denormalized _populate on read, _depopulate on write, FK\u2011guarded patchCRUDMixin Reusing CRUD inside an existing class raw CRUD against a collection you already have
There is no built\u2011in CachedPopulatingRepository. Use case 10 shows how to compose caching + population in a small subclass.
Generic CRUD over a Motor collection. The building block of all repositories.
Method Signature Notes createasync (data: T) -> T Dumps model (excludes id, None), stamps created_at/updated_at, inserts, returns model with assigned _id. get_by_idasync (id: str \\| ObjectId) -> Optional[T]str is accepted and converted to ObjectId. get_manyasync (filter: dict \\| None = None, skip: int = 0, limit: int = 100, sort: list[tuple] \\| None = None) -> list[T] Cursor .skip().limit().sort(...) then to_list(limit). limit=0 disables the limit clause. updateasync (id, data: dict[str, Any]) -> Optional[T]$set + refreshed updated_at via find_one_and_update. patchasync (id, data: dict[str, Any]) -> Optional[T] Same as update but intended for REST PATCH semantics. deleteasync (id) -> bool True if a document was deleted. countasync (filter: dict \\| None = None) -> intcount_documents."},{"location":"02_components/#4-baserepositoryt","title":"4. BaseRepository[T]","text":"
BaseRepository(collection_name: str, model: type[T]) \u2014 resolves the collection from MongoConnectionManager.get_database()[collection_name]. Requires an active connection at construction time. Provides everything in CRUDMixin plus collection_name.
_depopulate(document) \u2014 collapses populated FK fields back to ObjectId before create/update.
create / update accept a model T (not a dict) so depopulation can run.
patch blocks FK fields \u2014 raises ValueError(\"Cannot patch FK fields via patch(): ... Use update() to change FK fields.\").
Populate semantics (important): a PopulateRule names a field that holds either an ObjectId or a list[ObjectId] and is the same field that gets replaced with the resolved document(s). There is no separate \"ref field\" vs \"target field\". See use case 08.
"},{"location":"02_components/#6-transactionmanager","title":"6. TransactionManager","text":"Method Signature Behavior start_sessionasync ctx manager (**kwargs) -> AsyncIOMotorClientSession Yields a session with an active transaction. Pass session= to every collection call inside. execute_transactionasync (operations: list[Callable[[session], Awaitable[Any]]], **kwargs) -> list[Any] Runs each op inside one transaction and returns results in order; any exception aborts the transaction and propagates."},{"location":"02_components/#7-modelregistry","title":"7. ModelRegistry","text":"
Centralized models, indexes, and cache lifecycle for multi-collection services.
Method Signature Behavior register(collection_name: str, model: type[BaseDocument], indexes: list[Any] \\| None = None) Records model + index specs. Index specs are passed as-is to pymongo create_index \u2014 single tuples, compound lists, or dicts with keys/options. initialize_allasync (db: AsyncIOMotorDatabase \\| None = None) -> Nonecreate_index per registered spec (idempotent). Uses the manager database if db omitted. get_model(collection_name) -> type[BaseDocument] Raises KeyError if unregistered. list_collections() -> list[str] Registered collection names. set_cache_backend(backend: CacheBackend) -> None Register the single shared backend. initialize_cacheasync () -> None Starts the backend (background TTL cleanup) \u2014 raises RuntimeError if no backend registered. shutdown_cacheasync () -> None Stops the backend cleanly and clears it. get_cache_backend() -> Optional[CacheBackend] Current backend, if any."},{"location":"02_components/#8-cache-layer","title":"8. Cache Layer","text":""},{"location":"02_components/#81-cachebackend-abstract","title":"8.1 CacheBackend (abstract)","text":"
Dataclass: enabled: bool = True, backend: Literal[\"memory\", \"redis\"] = \"memory\", redis_client, default_ttl: int = 300, max_entries: int = 10000, key_prefix: str = \"\", cleanup_interval: int = 60. Raises ValueError if backend=\"redis\" without a client, and ImportError if redis is not installed.
RedisCacheBackend(redis_client, key_prefix=\"\", default_ttl=300) \u2014 setex storage, SCAN-based clear_pattern, and publish_invalidate(key) for cross-service invalidation on delete via the mongo_ops:cache:invalidate channel.
Cache keys are \"{key_prefix}{id}\" (prefix defaults to \"{collection_name}:\").
get_by_id \u2014 cache-first; cache miss reads DB and stores model_dump(by_alias=True) (JSON-encoded) for default_ttl. Honors config.enabled=False (bypass).
create \u2014 inserts then caches the result.
update/delete \u2014 refresh or remove the cache entry.
warm_cache(ids) -> int \u2014 prefetch a list of IDs, returns count warmed.
invalidate_cache(id) \u2014 manual eviction.
"},{"location":"02_components/#9-population-layer","title":"9. Population Layer","text":""},{"location":"02_components/#91-populaterule","title":"9.1 PopulateRule","text":"
Dataclass:
Python
@dataclass\nclass PopulateRule:\n field_name: str # field holding the ObjectId / list[ObjectId]; replaced in-place with the resolved doc(s)\n collection_name: str # collection the references point at\n nested_rules: list[PopulateRule] | None = None\n max_depth: int = 1\n filter: dict | None = None # DECLARED but NOT yet applied by the engine\n projection: dict | None = None # DECLARED but NOT yet applied by the engine\n
\u26a0\ufe0f filter and projection are accepted but currently ignored by PopulationEngine \u2014 do not rely on them yet.
One repository per collection. Encapsulate every query the domain needs behind repository methods; keep Mongo details ($regex, $inc, projections) inside the repository.
Keep models thin. BaseDocument for the shape; use Pydantic Field constraints for validation; never put business rules in the model.
Use services for cross-repository logic. A Service composes multiple repositories (and TransactionManager) \u2014 routes stay thin.
Expose get_many(filter=..., skip=..., limit=..., sort=...) instead of raw find for list endpoints \u2014 you get controlled pagination for free.
Connect once, in the lifespan. MongoConnectionManager.lifespan(...) (or explicit connect/disconnect) \u2014 never lazily per request.
Construct repositories after connect(). Module-level Repo() before connection raises RuntimeError(\"Database not connected...\"). Use dependencies or construct inside the lifespan/request.
Order the cache lifecycle strictly: set_cache_backend(backend) \u2192 initialize_cache() (after connect, before use) \u2192 shutdown_cache() on exit.
Register all models up front via ModelRegistry.register(...) and let initialize_all() create indexes once at startup (idempotent).
"},{"location":"04_best_practices/#data-performance","title":"\ud83d\uddc4\ufe0f Data & Performance","text":"
Declare indexes for every hot query. Single-field, composite, and optioned (unique/TTL) specs all work via ModelRegistry.register \u2014 see use case 13.
Cache only hot, low-write _id reads. Use CachedBaseRepository for lookups-by-id; invalidate (update/delete handle it) and pick a sensible default_ttl.
Populate at the repository boundary. PopulatingRepository resolves refs on read and depopulates on write; do not hand-roll joins in endpoints.
Respect the populate invariants: rules name the field that holds the reference and that is replaced; patch() cannot touch FK fields \u2014 use update().
Use transactions for multi-document writes. TransactionManager.start_session (inline) or execute_transaction (list of ops) \u2014 and pass session= to every collection call inside.
Handle the library's real exceptions at the edges: DuplicateKeyError \u2192 409, InvalidId \u2192 400, CircularReferenceError \u2192 409, ValueError guides \u2192 400/422 (see Error Handling).
Default to mock-based unit tests. Patch MongoConnectionManager.get_database, use AsyncMock collections and cursor chains \u2014 the whole suite runs without MongoDB (use case 14).
Mirror the library tests. tests/test_{repository,populating_repository,cache,registry,transactions}.py are canonical examples of every pattern above.
Use type hints end-to-end \u2014 mypy-gated CI (see pyproject) catches drift early.
What can raise, what it means, and how to map it in a FastAPI app.
"},{"location":"06_error_handling/#library-raised-exceptions","title":"\ud83d\udccb Library-Raised Exceptions","text":"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 \u2014 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 \u2014 use update() with a model. ValueError(\"...contains embedded dict(s) \u2014 run repair script\")_populate on read A FK field holds an embedded document instead of an ObjectId \u2014 migrate the data. ValueError(\"...contains ObjectId \u2014 was populate skipped?\")_depopulate on write A populate-ruled field is still an ObjectId at depopulate time \u2014 the read must have populated it first. CircularReferenceError(collection, doc_id, path)PopulationEngine.populate A (Class, id) pair was revisited \u2014 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'\")CacheConfigbackend=\"redis\" without a client. Pymongo DuplicateKeyError any insert/update Unique index violation (e.g., duplicate email). bson.errors.InvalidIdObjectId(...) 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 \u2014 they fail loudly at startup or first call instead of misbehaving silently.
from mongo_ops.repository import MongoConnectionManager \u2014 patch where it is used (mongo_ops.repository.MongoConnectionManager), matching the library's own tests.
pytest-asyncio runs as auto mode per pyproject.toml, so @pytest.mark.asyncio works without extra config.
For population, cache, registry, and transaction mockups \u2014 see the full testing guide.
"},{"location":"03_use_cases/01_basic_crud/","title":"Use Case 1: Basic FastAPI CRUD API","text":"
Scenario: A simple user management API with CRUD endpoints, index registration, and a correct connection lifecycle.
"},{"location":"03_use_cases/01_basic_crud/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description MongoConnectionManager.lifespan Connect on startup, disconnect on shutdown. BaseRepository[User] Generic CRUD \u2014 no endpoint-level Mongo code. ModelRegistry.register Declares the email index; initialize_all creates it at startup."},{"location":"03_use_cases/01_basic_crud/#example","title":"\ud83d\ude80 Example","text":"Python
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 \u2014 it accepts a partial dict like update, but PopulatingRepository rejects FK fields.
Note: this snippet omits the FastAPI lifespan connection wiring for brevity \u2014 copy it from use case 01 so ProductRepository() is created only after MongoConnectionManager.connect().
Methods that hit self.collection directly (regex search, $inc) bypass the caching and population layers. If a feature composes them \u2014 extend CachedBaseRepository or PopulatingRepository instead and add the domain methods there.
Prefer get_many(filter=...) over raw find when you want pagination/sort defaults for free.
Reuse self.model(**doc) to convert raw dicts to model instances consistently.
"},{"location":"03_use_cases/03_transactions/","title":"Use Case 3: Transaction Support for Multi-Document Operations","text":"
Scenario: Order processing must update inventory and create an order atomically. Any failure rolls both back.
"},{"location":"03_use_cases/03_transactions/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description TransactionManager.start_session Async context manager yielding a session with an active transaction. session= kwarg Pass to every insert_one / update_one / find_one inside the block."},{"location":"03_use_cases/03_transactions/#example","title":"\ud83d\ude80 Example","text":"Python
"},{"location":"03_use_cases/05_soft_deletes/","title":"Use Case 5: Soft Deletes Pattern","text":"
Scenario: Deleting a task should be recoverable. Instead of removing the document, set a tombstone flag and filter it from normal queries.
"},{"location":"03_use_cases/05_soft_deletes/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description SoftDeleteDocument Base model carrying is_deleted, deleted_at, deleted_by. SoftDeleteRepository[T] Repository-level soft delete / restore / filtering using update + get_many. permanent_delete Escapes to the real delete."},{"location":"03_use_cases/05_soft_deletes/#example","title":"\ud83d\ude80 Example","text":"Python
BaseRepository[T] is generic \u2014 subclasses must parameterize it (see SoftDeleteRepository[T] above). from mongo_ops import BaseRepository; BaseRepository[T] works out of the box.
Keep a compound index on {\"is_deleted\": 1, \"status\": 1} for active-list queries (see use case 13).
Soft-deleted documents should be excluded at the repository boundary, never re-filtered ad hoc in endpoints.
"},{"location":"03_use_cases/06_multi_model/","title":"Use Case 6: Multi-Model Service with Registration","text":"
Scenario: A social app manages users, posts, and comments. Each has its own model, repository, and indexes \u2014 registered centrally and initialized at startup.
"},{"location":"03_use_cases/06_multi_model/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description ModelRegistry.register One call per collection \u2014 model + indexes together. ModelRegistry.initialize_all Creates every registered index at startup (idempotent). Repositories One repository class per collection, all sharing the same connection."},{"location":"03_use_cases/06_multi_model/#example","title":"\ud83d\ude80 Example","text":"Python
Place user_repo = ... inside the lifespan/after connect. Module-level instantiation before connect() raises RuntimeError(\"Database not connected...\").
Model relationships here are plain ObjectId strings stored on the child docs. To resolve them on read, see use case 08 \u2013 Population.
Registering indexes on created_at/author_id/post_id keeps the common queries indexed (see use case 13).
"},{"location":"03_use_cases/07_caching/","title":"Use Case 7: Caching for High-Performance Reads","text":"
Scenario: A read-heavy API (product catalog) reduces DB load by caching documents by _id \u2014 in-memory locally, or shared via Redis.
"},{"location":"03_use_cases/07_caching/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description InMemoryCacheBackend TTL + LRU cache with a background cleanup task. RedisCacheBackend Distributed cache on redis.asyncio with pub/sub invalidation. CachedBaseRepository[T] Extends BaseRepository \u2014 cache-first get_by_id, cache on create, invalidate on update/delete, warm_cache(ids), invalidate_cache(id). Backend lifecycle The same backend instance must be both passed to the repository AND registered via ModelRegistry.set_cache_backend so initialize_cache() starts its task."},{"location":"03_use_cases/07_caching/#example","title":"\ud83d\ude80 Example","text":"Python
from contextlib import asynccontextmanager\n\nfrom fastapi import FastAPI, HTTPException\nfrom mongo_ops import BaseDocument, CachedBaseRepository, ModelRegistry, MongoConnectionManager\nfrom mongo_ops.cache import CacheConfig, InMemoryCacheBackend\n\n\nclass Product(BaseDocument):\n name: str = \"\"\n price: float = 0.0\n\n\n# One shared backend \u2014 used by both the repository and the registry lifecycle.\ncache = InMemoryCacheBackend(max_entries=10_000, default_ttl=300)\nModelRegistry.set_cache_backend(cache)\n\n\nclass ProductRepo(CachedBaseRepository[Product]):\n def __init__(self):\n super().__init__(\n collection_name=\"products\",\n model=Product,\n cache_backend=cache,\n config=CacheConfig(enabled=True, backend=\"memory\"),\n )\n\n\n@asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\", db_name=\"shop\"\n ):\n await ModelRegistry.initialize_all()\n await ModelRegistry.initialize_cache() # starts the TTL cleanup task\n yield\n await ModelRegistry.shutdown_cache() # cancels it on exit\n\n\napp = FastAPI(lifespan=lifespan)\n\n\n@app.post(\"/products/\", response_model=Product)\nasync def create_product(product: Product):\n return await ProductRepo().create(product) # created after connect()\n\n\n@app.get(\"/products/{product_id}\", response_model=Product)\nasync def get_product(product_id: str):\n product = await ProductRepo().get_by_id(product_id)\n if not product:\n raise HTTPException(status_code=404, detail=\"Product not found\")\n return product\n\n\n@app.put(\"/products/{product_id}\", response_model=Product)\nasync def update_product(product_id: str, name: str | None = None, price: float | None = None):\n data = {}\n if name is not None:\n data[\"name\"] = name\n if price is not None:\n data[\"price\"] = price\n return await ProductRepo().update(product_id, data)\n\n\n# Outside request handlers:\n# await ProductRepo().warm_cache([object_id_1, object_id_2]) # pre-load n ids -> int\n# await ProductRepo().invalidate_cache(object_id_3) # manual eviction\n
Cache keys are \"{key_prefix}{id}\" (default prefix \"products:\"). clear_pattern(\"products:*\") wipes a whole collection's entries.
Values are JSON-encoded (json.dumps(obj, default=str)) \u2014 nested models inside a cached document are stored as dicts, not objects.
Enable/disable per repository with CacheConfig(enabled=False); a disabled repo bypasses the cache entirely.
Distinguish the two initialize* calls: initialize_cache() starts the backend task; initialize_all() creates indexes. Both belong in the lifespan, after connect().
"},{"location":"03_use_cases/08_population/","title":"Use Case 8: Document Population","text":"
Scenario: An API returns a User with its related Profile embedded in one JSON payload \u2014 no second round-trip from the client, no joins.
"},{"location":"03_use_cases/08_population/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description PopulateRule Declares which field to resolve. field_name holds the ObjectId (or list[ObjectId]) and is the same field the resolved document replaces in place. collection_name says where the referenced documents live. PopulationEngine Holds repositories by collection name and resolves rules recursively, detecting cycles (CircularReferenceError). PopulatingRepository[T]get_by_id/get_many populate on read; create/update depopulate on write; patch rejects FK fields.
\u26a0\ufe0f PopulateRule does not have a separate \"ref field\" vs \"target field\" \u2014 the ref field is the populated field. filter/projection on PopulateRule are declared but not yet applied by the engine.
Patching an FK field raises ValueError \u2014 switch FK changes to update(user_id, model) instead. See use case 05 for the reasoning with soft deletes.
Missing references resolve to None, not an error.
A field holding an embedded dict (instead of an ObjectId) raises a ValueError (\"run repair script\") \u2014 migrate embedded docs to a separate collection first.
If create/update receives a field that is already an ObjectId under a populate rule, _depopulate raises ValueError(\"...was populate skipped?\") \u2014 populate-then-depopulate pairs must be balanced.
09 \u2013 Advanced population \u00b7 10 \u2013 Cache + population \u00b7 14 \u2013 Testing guide
"},{"location":"03_use_cases/09_advanced_population/","title":"Use Case 9: Nested Document Population & Circular-Ref Handling","text":"
Scenario: An Author has books (a list of references), each Book references a publisher, and an Author may reference a mentor \u2014 which is another Author (a potential cycle).
"},{"location":"03_use_cases/09_advanced_population/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description nested_rules Populates deeper levels: resolve books on Author, then publisher inside each Book. max_depth Per-rule recursion bound \u2014 the safety net for cyclic graphs. global_max_depthPopulationEngine(repos, global_max_depth=10) global cap. CircularReferenceError Raised when a (Class, id) pair is revisited; carries the visited path."},{"location":"03_use_cases/09_advanced_population/#example","title":"\ud83d\ude80 Example","text":"Python
from contextlib import asynccontextmanager\n\nfrom bson import ObjectId\nfrom fastapi import FastAPI, HTTPException\nfrom mongo_ops import BaseDocument, ModelRegistry, MongoConnectionManager, PopulatingRepository\nfrom mongo_ops.cache import CircularReferenceError\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\n\n\n# 1. Models \u2014 each ref field holds ObjectId(s) in the DB and becomes model(s) in memory.\nclass Publisher(BaseDocument):\n name: str = \"\"\n country: str = \"\"\n\n\nclass Book(BaseDocument):\n title: str = \"\"\n publisher: Publisher | None = None # ObjectId in DB, Publisher in memory\n\n\nclass Author(BaseDocument):\n name: str = \"\"\n books: list[Book] | None = None # list[ObjectId] in DB, list[Book] in memory\n mentor: \"Author\" | None = None # self-reference \u2014 potential cycle\n\n\n# 2. Engine + nested rules.\nengine = PopulationEngine({})\n\npublisher_rule = PopulateRule(\n field_name=\"publisher\",\n collection_name=\"publishers\",\n)\n\nbook_rule = PopulateRule(\n field_name=\"books\",\n collection_name=\"books\",\n nested_rules=[publisher_rule], # fetch each book, then its publisher\n max_depth=3,\n)\n\nmentor_rule = PopulateRule(\n field_name=\"mentor\",\n collection_name=\"authors\",\n max_depth=2, # stops mentor chains early \u2014 also avoids unbounded cycles\n)\n\n\nclass AuthorRepository(PopulatingRepository[Author]):\n def __init__(self):\n super().__init__(\n collection_name=\"authors\",\n model=Author,\n population_engine=engine,\n populate_rules=[book_rule, mentor_rule],\n )\n\n\n@asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\", db_name=\"library\"\n ):\n engine.register_repo(\"publishers\", PopulatingRepository[Publisher](\"publishers\", Publisher))\n engine.register_repo(\"books\", PopulatingRepository[Book](\"books\", Book))\n engine.register_repo(\"authors\", AuthorRepository())\n await ModelRegistry.initialize_all()\n yield\n\n\napp = FastAPI()\n\n\n@app.get(\"/authors/{author_id}\")\nasync def get_author(author_id: str):\n try:\n author = await AuthorRepository().get_by_id(author_id)\n except CircularReferenceError as exc:\n raise HTTPException(status_code=409, detail=f\"Circular reference: {exc}\")\n if not author:\n raise HTTPException(status_code=404, detail=\"Author not found\")\n return author\n
08 \u2013 Population \u00b7 10 \u2013 Cache + population \u00b7 Error Handling
"},{"location":"03_use_cases/10_cache_and_population/","title":"Use Case 10: Caching + Population (Read-Through, Populated on Read)","text":"
Scenario: A service needs fast reads of a User and its related Profile. We want the cache-hit path to skip the DB entirely while still returning a populated document.
"},{"location":"03_use_cases/10_cache_and_population/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description CachedBaseRepository Provides the cache layer (cache-first get_by_id, invalidation on update/delete). PopulationEngine Provides the reference resolution. Composed subclass A small CachedBaseRepository subclass adds two _populate helpers so cache hits return populated models.
\u26a0\ufe0f Known limitation: there is no built-in CachedPopulatingRepository in the library. This use case documents the composition. We cache the raw (depopulated) document \u2014 references stay ObjectIds until read time, so Profile changes are reflected on the next fetch (within TTL) and update/delete invalidation remains correct.
from contextlib import asynccontextmanager\nfrom datetime import datetime\nfrom typing import Any, Union\n\nfrom bson import ObjectId\nfrom fastapi import FastAPI\nfrom mongo_ops import BaseDocument, CachedBaseRepository, ModelRegistry, MongoConnectionManager\nfrom mongo_ops.cache import CacheConfig, InMemoryCacheBackend\nfrom mongo_ops.cache.in_memory import decode_value, encode_value\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\n\n\n# 1. Models \u2014 `profile` holds an ObjectId in DB, a Profile in memory.\nclass Profile(BaseDocument):\n avatar_url: str = \"\"\n bio: str = \"\"\n\n\nclass User(BaseDocument):\n username: str = \"\"\n email: str = \"\"\n profile: Profile | None = None\n\n\n# 2. Engine + rule.\nengine = PopulationEngine({})\nprofile_rule = PopulateRule(field_name=\"profile\", collection_name=\"profiles\")\n\n\n# 3. Composed repository.\nclass CachedUserRepository(CachedBaseRepository[User]):\n def __init__(\n self,\n cache_backend: InMemoryCacheBackend,\n population_engine: PopulationEngine,\n populate_rules: list[PopulateRule],\n config: CacheConfig | None = None,\n ):\n super().__init__(\"users\", User, cache_backend, config)\n self.population_engine = population_engine\n self._populate_rules = populate_rules\n\n async def _populate(self, data: dict[str, Any]) -> dict[str, Any]:\n \"\"\"Dict-level resolution (mirrors PopulatingRepository._populate).\"\"\"\n for rule in self._populate_rules:\n ref = data.get(rule.field_name)\n if ref is None:\n continue\n if isinstance(ref, str): # JSON round-trip turns ObjectId into hex str\n ref = ObjectId(ref)\n if isinstance(ref, list):\n resolved = []\n for item in ref:\n item = ObjectId(item) if isinstance(item, str) else item\n repo = self.population_engine._repos.get(rule.collection_name)\n doc = await repo.get_by_id(item) if repo else None\n resolved.append(doc)\n data[rule.field_name] = resolved\n elif isinstance(ref, ObjectId):\n repo = self.population_engine._repos.get(rule.collection_name)\n data[rule.field_name] = await repo.get_by_id(ref) if repo else None\n return data\n\n async def _depopulate(self, data: User) -> dict[str, Any]:\n \"\"\"Model -> raw dict, FK fields collapsed to ObjectId (mirrors _depopulate).\"\"\"\n doc = data.model_dump(exclude={\"id\"}, exclude_none=True)\n for rule in self._populate_rules:\n value = getattr(data, rule.field_name, None)\n if isinstance(value, list):\n doc[rule.field_name] = [item.id for item in value]\n elif isinstance(value, BaseDocument):\n doc[rule.field_name] = value.id\n return doc\n\n async def create(self, data: User) -> User:\n doc = await self._depopulate(data) # store FK refs as ObjectIds\n doc[\"created_at\"] = datetime.utcnow()\n doc[\"updated_at\"] = datetime.utcnow()\n result = await self.collection.insert_one(doc)\n doc[\"_id\"] = result.inserted_id\n if self._cache_config.enabled:\n await self._cache.set(self._cache_key(doc[\"_id\"]), encode_value(doc), self._cache_config.default_ttl)\n return self.model(**await self._populate(doc))\n\n async def update(self, id: Union[str, ObjectId], data: User) -> User | None:\n if isinstance(id, str):\n id = ObjectId(id)\n doc = await self._depopulate(data) # update() takes a full model here\n doc[\"updated_at\"] = datetime.utcnow()\n result = await self.collection.find_one_and_update(\n {\"_id\": id}, {\"$set\": doc}, return_document=True\n )\n key = self._cache_key(id)\n if result is None:\n if self._cache_config.enabled:\n await self._cache.delete(key)\n return None\n raw = dict(result)\n if self._cache_config.enabled:\n await self._cache.set(key, encode_value(raw), self._cache_config.default_ttl)\n return self.model(**await self._populate(raw))\n\n async def get_by_id(self, id: Union[str, ObjectId]) -> User | None:\n if not self._cache_config.enabled:\n # No cache: still a raw fetch + populate \u2014 the base get_by_id()\n # would build a typed model from the raw doc and fail on FK fields.\n raw = await self.collection.find_one({\"_id\": ObjectId(id) if isinstance(id, str) else id})\n return self.model(**await self._populate(raw)) if raw else None\n\n cached = await self._cache.get(self._cache_key(id))\n if cached is not None:\n # Cache hit: decode the RAW doc, then populate before returning.\n data = await self._populate(decode_value(cached))\n return self.model(**data)\n\n # Cache miss: one RAW DB read. The base get_by_id() would rebuild the\n # model from the raw doc, which fails for model-typed FK fields \u2014\n # so fetch the raw dict, cache it, and populate before materializing.\n key = self._cache_key(id)\n raw = await self.collection.find_one({\"_id\": ObjectId(id) if isinstance(id, str) else id})\n if raw is None:\n return None\n await self._cache.set(key, encode_value(raw), self._cache_config.default_ttl)\n return self.model(**await self._populate(raw))\n\n\n# 4. Wire-up \u2014 one backend for both the repo and the registry lifecycle.\ncache = InMemoryCacheBackend(max_entries=20_000, default_ttl=600)\nModelRegistry.set_cache_backend(cache)\n\n\n@asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\", db_name=\"app_db\"\n ):\n # Register the repositories the engine resolves refs against:\n # engine.register_repo(\"profiles\", ProfileRepo())\n engine.register_repo(\"users\", CachedUserRepository(cache, engine, [profile_rule]))\n await ModelRegistry.initialize_all()\n await ModelRegistry.initialize_cache()\n yield\n await ModelRegistry.shutdown_cache()\n\n\napp = FastAPI(lifespan=lifespan)\n
Note: replace the placeholder engine.register_repo(\"profiles\", ...) line with a repository for Profile. A matching CachedBaseRepository[Profile](\"profiles\", Profile, cache) keeps profile reads cached too \u2014 e.g. engine.register_repo(\"profiles\", CachedBaseRepository[Profile](\"profiles\", Profile, cache)).
Cache the raw doc, populate at read time. Ref fields stay ObjectIds in the cache; JSON round-trips them to hex strings, so _populate must normalize str \u2192 ObjectId (shown above).
update/create above are model-based (update(id, User)), mirroring PopulatingRepository \u2014 they depopulate before writing and cache a raw snapshot. The plain CachedBaseRepository.update(id, dict) and delete(id) keep working and invalidate the same key.
Set a sensible default_ttl \u2014 cached User entries resolve Profile on each read, so profile edits show up within the TTL (or call invalidate_cache(user_id) explicitly).
To reuse this compose logic across many collections, extract the _populate / _depopulate helpers plus the create / update / get_by_id overrides into a mixin and parameterize the rules per subclass.
"},{"location":"03_use_cases/11_cache_lifecycle/","title":"Use Case 11: Proper Cache Lifecycle in FastAPI","text":"
Scenario: A microservice uses an in-memory (or Redis) cache backend and must start the cleanup task on app start, then shut it down cleanly on termination.
"},{"location":"03_use_cases/11_cache_lifecycle/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description ModelRegistry.set_cache_backend(backend) Registers the single shared backend. ModelRegistry.initialize_cache() Starts the backend (spawns the in-memory TTL cleanup task). Raises RuntimeError if no backend was registered. ModelRegistry.shutdown_cache() Cancels the cleanup task, closes pub/sub, clears the backend. If a repo uses a backend directly Same instance must be registered so initialize_cache starts its task."},{"location":"03_use_cases/11_cache_lifecycle/#example","title":"\ud83d\ude80 Example","text":"Python
from contextlib import asynccontextmanager\n\nfrom fastapi import FastAPI\nfrom mongo_ops import ModelRegistry, MongoConnectionManager\nfrom mongo_ops.cache import InMemoryCacheBackend\n\ncache = InMemoryCacheBackend(max_entries=10_000, default_ttl=300)\nModelRegistry.set_cache_backend(cache) # before any cache-backed repo is used\n\n\n@asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\", db_name=\"mydb\"\n ):\n await ModelRegistry.initialize_all() # create indexes (idempotent)\n await ModelRegistry.initialize_cache() # start the TTL cleanup task\n yield\n await ModelRegistry.shutdown_cache() # cancel task + close cleanly\n\n\napp = FastAPI(lifespan=lifespan)\n
"},{"location":"03_use_cases/11_cache_lifecycle/#what-actually-happens","title":"\ud83d\udd01 What Actually Happens","text":"
InMemoryCacheBackend.initialize() spawns an asyncio.Task that evicts expired entries every cleanup_interval seconds. Without a shutdown, the event loop flags the dangling task on exit \u2014 shutdown_cache() cancels it and awaits it.
shutdown_cache() also NULs the registry cache backend and closes the Redis pub/sub handle (if Redis).
If initialize_cache() is called before set_cache_backend(), it raises:
Text Only
RuntimeError: No cache backend registered. Call set_cache_backend() first.\n
Per-repository toggling is independent of the lifecycle: CacheConfig(enabled=False) bypasses the cache for that repo even after initialize_cache().
The cleanup task uses cleanup_interval seconds for scans; entries also expire on access via the TTL heap (default_ttl=0 expires immediately).
Register the backend before constructing any CachedBaseRepository that references it \u2014 otherwise a repo may hold an uninitialized backend (no cleanup task, no Redis pub/sub).
"},{"location":"03_use_cases/12_transaction_helper/","title":"Use Case 12: Using TransactionManager.execute_transaction","text":"
Scenario: Perform several writes across different collections atomically \u2014 e.g., create an Order and decrement Inventory. The low-level start_session context works, but execute_transaction collects results from a list of async operations.
"},{"location":"03_use_cases/12_transaction_helper/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description TransactionManager.execute_transaction Runs a list of async callables (each receives a session) inside one transaction; returns the result of each callable, in order. Automatic rollback Any raised exception aborts the transaction and propagates to the caller. start_session The underlying async context manager async with TransactionManager.start_session() as session:."},{"location":"03_use_cases/12_transaction_helper/#example","title":"\ud83d\ude80 Example","text":"Python
Return values: each callable can return whatever you need; results are collected in the same order.
Errors: raise inside any callable \u2192 the whole transaction aborts (session rolls back) and the exception propagates.
Reads inside a transaction: pass session=session to find_one/find too.
Testing (no Mongo): monkeypatch a fake client on mongo_ops.transactions.MongoConnectionManager.get_client and stub start_session \u2014 see tests/test_transactions.py.
Repo models still carry created_at/updated_at; for raw collection inserts inside the transaction you set them manually (as shown).
"},{"location":"03_use_cases/13_index_creation/","title":"Use Case 13: Declaring Indexes (Single-Field, Composite, Unique)","text":"
Scenario: Ensure each collection has the right indexes for fast queries and data integrity \u2014 declared in one place and created at startup.
"},{"location":"03_use_cases/13_index_creation/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description ModelRegistry.register(indexes=...) Each spec is passed as-is to pymongo create_index. Supported forms: tuple (\"field\", direction), compound list [(\"a\", 1), (\"b\", -1)], or a dict with keys + options. ModelRegistry.initialize_all Creates every registered index during startup (idempotent \u2014 create_index skips existing indexes)."},{"location":"03_use_cases/13_index_creation/#example","title":"\ud83d\ude80 Example","text":"Python
from contextlib import asynccontextmanager\n\nfrom pymongo import ASCENDING, DESCENDING\nfrom mongo_ops import BaseDocument, ModelRegistry, MongoConnectionManager\n\n\nclass User(BaseDocument):\n username: str = \"\"\n email: str = \"\"\n\n\nclass BlogPost(BaseDocument):\n author_id: str = \"\"\n created_at: str = \"\"\n title: str = \"\"\n\n\nclass Passenger(BaseDocument):\n email: str = \"\"\n seat: str = \"\"\n\n\n# 1\ufe0f\u20e3 Single-field index \u2014 email lookups.\nModelRegistry.register(\n collection_name=\"users\",\n model=User,\n indexes=[(\"email\", ASCENDING)],\n)\n\n# 2\ufe0f\u20e3 Composite index \u2014 queries filtering by author + creation date.\nModelRegistry.register(\n collection_name=\"posts\",\n model=BlogPost,\n indexes=[[(\"author_id\", ASCENDING), (\"created_at\", DESCENDING)]],\n)\n\n# 3\ufe0f\u20e3 Unique index with a custom name \u2014 enforce unique emails.\nModelRegistry.register(\n collection_name=\"passengers\",\n model=Passenger,\n indexes=[\n {\n \"keys\": [(\"email\", ASCENDING)],\n \"options\": {\"unique\": True, \"name\": \"uq_passenger_email\"},\n }\n ],\n)\n\n\n# 4\ufe0f\u20e3 Everything is created on startup.\n@asynccontextmanager\nasync def lifespan(_app):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\",\n db_name=\"mydb\",\n ):\n await ModelRegistry.initialize_all()\n yield\n
Every spec passes through to collection.create_index(spec) \u2014 so MongoDB options like unique, sparse, and expireAfterSeconds (TTL) belong in the options dict.
Idempotent by construction: create_index is a no-op when a same-shape index already exists.
Verify with the Mongo shell:
JavaScript
db.<collection>.getIndexes()\n
A unique index on an already-duplicated field will fail with DuplicateKeyError on startup \u2014 clean the data first.
Scenario: Write unit tests that never touch a real MongoDB \u2014 mock the collection, the cache backend, and the client, exactly like the library's own test suite (tests/).
"},{"location":"03_use_cases/14_testing_guide/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description AsyncMock collections Stub find_one, insert_one, find_one_and_update, cursor chains. Patching MongoConnectionManager.get_database Gives repositories a mock collection without a live connection. monkeypatch on get_client Fakes sessions for TransactionManager tests. Reference tests tests/test_populating_repository.py, tests/test_cache.py, tests/test_registry.py, tests/test_transactions.py."},{"location":"03_use_cases/14_testing_guide/#example-boilerplate","title":"\ud83d\ude80 Example Boilerplate","text":"Python
import pytest\nfrom unittest.mock import AsyncMock, MagicMock, patch\n\nfrom bson import ObjectId\nfrom mongo_ops.cache import CacheConfig, InMemoryCacheBackend\nfrom mongo_ops.cache.repository import CachedBaseRepository\nfrom mongo_ops.models import BaseDocument\nfrom mongo_ops.registry import ModelRegistry\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\nfrom mongo_ops.repository import PopulatingRepository\n\n\n# ----------------------------------------------------------------------\n# 1. Models (same shape as the library tests)\n# ----------------------------------------------------------------------\nclass Profile(BaseDocument):\n avatar_url: str = \"\"\n\n\nclass User(BaseDocument):\n name: str = \"\"\n profile: Profile | None = None # ObjectId in DB, Profile in memory\n\n\n# ----------------------------------------------------------------------\n# 2. Fixtures\n# ----------------------------------------------------------------------\n@pytest.fixture\ndef mock_collection():\n return AsyncMock()\n\n\n@pytest.fixture\ndef engine():\n profile_repo = AsyncMock()\n return PopulationEngine({\"profiles\": profile_repo})\n\n\n@pytest.fixture\ndef repo(mock_collection, engine):\n with patch(\"mongo_ops.repository.MongoConnectionManager.get_database\") as mock_db:\n mock_db.return_value.__getitem__.return_value = mock_collection\n r = PopulatingRepository(\n \"users\",\n User,\n population_engine=engine,\n populate_rules=[PopulateRule(field_name=\"profile\", collection_name=\"profiles\")],\n )\n r.collection = mock_collection\n return r\n\n\n# ----------------------------------------------------------------------\n# 3. Population \u2014 get_by_id resolves the reference\n# ----------------------------------------------------------------------\n@pytest.mark.asyncio\nasync def test_get_by_id_populates(repo, mock_collection, engine):\n uid, pid = ObjectId(), ObjectId()\n mock_collection.find_one.return_value = {\n \"_id\": uid,\n \"name\": \"Alice\",\n \"profile\": pid, # ObjectId stored in DB\n \"created_at\": \"2024-01-01T00:00:00\",\n \"updated_at\": \"2024-01-01T00:00:00\",\n }\n engine._repos[\"profiles\"].get_by_id.return_value = Profile(id=pid, avatar_url=\"pic.png\")\n\n result = await repo.get_by_id(uid)\n\n assert result is not None\n assert result.name == \"Alice\"\n assert isinstance(result.profile, Profile)\n assert result.profile.avatar_url == \"pic.png\"\n\n\n# ----------------------------------------------------------------------\n# 4. Patch FK guard\n# ----------------------------------------------------------------------\n@pytest.mark.asyncio\nasync def test_patch_rejects_fk_field(repo):\n with pytest.raises(ValueError, match=\"Cannot patch FK fields\"):\n await repo.patch(ObjectId(), {\"profile\": ObjectId()})\n\n\n# ----------------------------------------------------------------------\n# 5. Registry \u2014 index specs pass through to create_index\n# ----------------------------------------------------------------------\n@pytest.mark.asyncio\nasync def test_initialize_all_creates_indexes():\n ModelRegistry.register(\"users\", User, indexes=[(\"email\", 1)])\n\n fake_collection = AsyncMock()\n await ModelRegistry.initialize_all(db={\"users\": fake_collection})\n\n fake_collection.create_index.assert_awaited_once_with((\"email\", 1))\n\n\n# ----------------------------------------------------------------------\n# 6. Cached repository \u2014 cache-first reads + invalidation\n# ----------------------------------------------------------------------\n@pytest.mark.asyncio\nasync def test_cached_get_by_id_populates_cache():\n backend = InMemoryCacheBackend(\n max_entries=100, default_ttl=300, cleanup_interval=9999\n )\n await backend.initialize()\n try:\n with patch(\"mongo_ops.repository.MongoConnectionManager.get_database\") as mock_db:\n mock_collection = AsyncMock()\n mock_db.return_value.__getitem__.return_value = mock_collection\n repo = CachedBaseRepository(\n \"users\", User, backend, CacheConfig(enabled=True)\n )\n repo.collection = mock_collection\n\n oid = ObjectId()\n mock_collection.find_one.return_value = {\n \"_id\": oid,\n \"name\": \"cached\",\n \"created_at\": \"2024-01-01T00:00:00\",\n \"updated_at\": \"2024-01-01T00:00:00\",\n }\n\n first = await repo.get_by_id(oid)\n assert first is not None\n\n mock_collection.find_one.return_value = None # DB now \"empty\"\n second = await repo.get_by_id(oid) # served from cache\n\n assert second is not None\n assert second.name == \"cached\"\n mock_collection.find_one.assert_awaited_once() # only one DB read\n finally:\n await backend.shutdown()\n\n\n# ----------------------------------------------------------------------\n# 7. Transactions \u2014 fake the client's start_session\n# ----------------------------------------------------------------------\nfrom mongo_ops.transactions import TransactionManager\n\n\n@pytest.mark.asyncio\nasync def test_execute_transaction(monkeypatch):\n session_ctx = AsyncMock()\n session_ctx.__aenter__.return_value = AsyncMock()\n session_ctx.__aexit__.return_value = None\n\n client = MagicMock()\n client.start_session = AsyncMock(return_value=session_ctx)\n\n # NOTE: start_transaction must return a context manager, not a coroutine.\n async_session = session_ctx.__aenter__.return_value\n async_session.start_transaction = lambda **_: session_ctx\n\n monkeypatch.setattr(\n \"mongo_ops.transactions.MongoConnectionManager.get_client\",\n lambda: client,\n )\n\n async def fake_op(session):\n return \"ok\"\n\n results = await TransactionManager.execute_transaction([fake_op])\n assert results == [\"ok\"]\n
pytest-asyncio is already configured in pyproject.toml (asyncio_mode = \"auto\"), so @pytest.mark.asyncio tests work out of the box. Run with pytest (coverage reports are enabled there too).
Never hit the network. Keep the patches in fixtures (or a conftest.py) and reuse them.
Mock cursor chains with MagicMock() + .to_list = AsyncMock(...), exactly like tests/test_repository.py.
For an optional integration check (real Mongo), use MongoConnectionManager.lifespan against a local replica set and drop the test database in teardown \u2014 keep it separate from the unit suite.
The pattern works symmetrically for Redis: mock the RedisCacheBackend methods (get, set, delete) \u2014 no Redis process required.
"},{"location":"03_use_cases/15_populating_repository_wiring/","title":"Use Case 15: Inside PopulatingRepository \u2014 the Object \u21c4 ObjectId Lifecycle","text":"
Scenario: You want to see what the repository actually does, in which order, before you trust it with your data \u2014 how a Profile becomes an ObjectId for storage and comes back as a Profile on read, and where every piece is wired.
"},{"location":"03_use_cases/15_populating_repository_wiring/#two-representations-one-field","title":"\ud83d\udce6 Two Representations, One Field","text":"
A populate-ruled field is a shape shifter \u2014 the same name holds different things depending on where you look:
Place Value held in the field In the MongoDB doc ObjectId (or list[ObjectId]) In the cache/JSON hex string (see use case 16) In the app model the referenced model (or None)
PopulatingRepository.create / update depopulate (model \u2192 ObjectId) before writing; get_by_id / get_many populate (ObjectId \u2192 model) after reading:
"},{"location":"03_use_cases/15_populating_repository_wiring/#write-path-create-step-by-step","title":"\ud83d\ude80 Write Path \u2014 create() Step by Step","text":"
Given user = User(username=\"alice\", profile=saved_profile) where saved_profile.id exists:
Python
class UserRepository(PopulatingRepository[User]):\n def __init__(self):\n super().__init__(\n collection_name=\"users\",\n model=User,\n population_engine=engine,\n populate_rules=[profile_rule],\n )\n
Guard check (_depopulate, repository.py): for each rule field, if the value is already an ObjectId (or list[ObjectId]) an error is raised \u2014 see _depopulate guards below. Saved models pass.
engine.depopulate(user, rules) collapses the graph in place:
saved_profile (a BaseDocument with .id) \u2192 becomes saved_profile.id \u2192 ObjectId
a Profile without .id (unsaved) \u2192 is left as a model object, which model_dump then embeds as a dict \u2014 see intricacy #3
when nested_rules are present, the nested model is depopulated recursively first, then collapsed
model_dump(exclude={\"id\"}, exclude_none=True) produces the raw insert dict; timestamps are added.
insert_one(doc) writes {..., \"profile\": ObjectId(\"...\"), ...} to MongoDB.
On the way out, data_to_model runs _populate (the read path below) so create returns a fully populated model, not the raw one.
"},{"location":"03_use_cases/15_populating_repository_wiring/#the-_depopulate-guards-loud-failures-silent-corruption","title":"The _depopulate guards (loud failures > silent corruption)","text":"Stored/held value under a rule field Behaviour ObjectId / list[ObjectId]ValueError(\"...contains ObjectId \u2014 was populate skipped?\") \u2014 the doc was loaded raw (e.g. via BaseRepository or an unpopulated read) and handed back to create/update a saved BaseDocument (has .id) collapse to .id (ObjectId) an unsaved BaseDocument (no .id) kept as a model \u2192 embedded dict below a dict not BaseDocument \u2192 left as-is \u2192 embedded non-BaseDocument, non-dict value AttributeError raised by engine.depopulate no engine / no rules passthrough \u2014 _depopulate just does model_dump, whatever shape you gave is stored"},{"location":"03_use_cases/15_populating_repository_wiring/#read-path-get_by_id-step-by-step","title":"\ud83d\ude80 Read Path \u2014 get_by_id() Step by Step","text":"
get_by_id inherits CRUD and only changes data_to_model (repository.py:247):
_populate(data) walks each rule field in the raw dict:
ref = data.get(\"profile\") \u2192 ObjectId. If ref is None \u2192 field untouched (stays absent).
Look up the repo: repo = engine._repos.get(\"profiles\"). If it's not registered, doc = None.
doc = await repo.get_by_id(ref) \u2014 a single read on the referenced collection.
If doc is None \u2192 data[\"profile\"] = None (missing refs resolve to None, never raise).
If the rule has nested_rules \u2192 engine.populate(doc, nested_rules, depth=1) deepens the result.
data[\"profile\"] = doc \u2192 self.model(**data) builds the User with a real Profile.
Lists behave the same per item; a ref that is a dict (an embedded document) raises ValueError(\"...contains embedded dict(s) \u2014 run repair script\").
Engine is optional. Set population_engine=None and populate_rules=[], and PopulatingRepository is just a BaseRepository \u2014 FK fields come back as raw ObjectId, and your model field type must agree (that is the whole point of the guard in the WRITE path: a Profile | None-typed field populated with a raw ObjectId is a broken round-trip waiting to happen).
"},{"location":"03_use_cases/15_populating_repository_wiring/#how-its-wired","title":"\ud83d\udd0c How It's Wired","text":"Python
from mongo_ops import BaseDocument, PopulatingRepository\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\n\nclass Profile(BaseDocument):\n avatar_url: str = \"\"\n\nclass User(BaseDocument):\n username: str = \"\"\n profile: Profile | None = None # ObjectId in DB, Profile in memory\n\nengine = PopulationEngine({}) # repositories live here, keyed by collection name\n\nprofile_rule = PopulateRule(field_name=\"profile\", collection_name=\"profiles\")\n\nclass UserRepository(PopulatingRepository[User]):\n def __init__(self):\n super().__init__(\"users\", User, population_engine=engine, populate_rules=[profile_rule])\n\ndef wire() -> None:\n \"\"\"Call AFTER connect() \u2014 repositories need a live database.\"\"\"\n engine.register_repo(\"profiles\", PopulatingRepository[Profile](\"profiles\", Profile))\n engine.register_repo(\"users\", UserRepository())\n
Wiring rules:
engine._repos is keyed by collection_name as written in the rule \u2014 typo \u2192 silent None refs.
register_repo needs an already-constructed repository \u2192 call it inside the lifespan (after connect() / MongoConnectionManager.lifespan), not at module import.
The referenced repository only needs a get_by_id that returns a BaseDocument \u2014 it can be a plain BaseRepository, another PopulatingRepository, or even a cached repo (see use case 10). Population requires no extra DB index on the referenced collection's _id.
Swap at runtime: repo.set_population_engine(new_engine) and repo.set_populate_rules(new_rules) \u2014 the tests exercise both.
"},{"location":"03_use_cases/15_populating_repository_wiring/#intricacy-nested_rules-do-not-round-trip-through-depopulate","title":"\u26a0\ufe0f Intricacy \u2014 nested_rules Do NOT Round-Trip Through depopulate","text":"
nested_rules are designed for read-side deep population. If the same rules run through the write path (create / update), engine.depopulate behaves differently from plain refs \u2014 verified against the engine:
Rule shape What depopulate does to it Effect on the stored doc scalar FK, no nested_rules collapse to .id stored as ObjectId \u2713 list[ObjectId] FK, no nested_rules collapse each item to .id stored as list[ObjectId] \u2713 list FK with nested_rules items are kept as models stored as embedded dicts \u2717 \u2014 reading back raises \"contains embedded dict(s) \u2014 run repair script\" scalar FK with nested_rules the recursion runs, then the field is assigned None reference lost \u2717 \u2014 written as absent/null
In other words: a document carrying deep nested_rules (like UC 09's Author \u2192 books \u2192 publisher) cannot be created / updated as-is. If the graph must be written back, materialize references separately (save each Book to its collection first, then store list[ObjectId] without nested_rules), and keep nested_rules only on read rules you never hand back to depopulate.
Never pass a raw ObjectId-holding model to create/update: the \"was populate skipped?\" ValueError is the guard. Read through the repo so the read path can populate first.
Unsaved references are a write-once trap: they embed as dicts, and re-reading raises the \"run repair script\" ValueError. Save referenced docs to their collection first, then reference their id.
filter / projection on PopulateRule are declared but not applied by the engine \u2014 don't rely on them.
Reads cost 1 query per reference (no $lookup yet); batch-heavy endpoints should add caching (next use case).
08 \u2013 Population \u00b7 09 \u2013 Advanced population \u00b7 16 \u2013 Cached repository intricacies \u00b7 Components
"},{"location":"03_use_cases/16_cached_repository_intricacies/","title":"Use Case 16: Inside the Cached Repository \u2014 What's Actually Stored & Returned","text":"
Scenario: You want the exact contract of CachedBaseRepository \u2014 what goes into the cache, in what shape, and why a cached read occasionally looks \"wrong\" for populated models \u2014 before wiring it into a service.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#the-cache-contract","title":"\ud83d\udce6 The Cache Contract","text":"Aspect Value Key \"{key_prefix}{id}\" \u2014 default prefix \"{collection_name}:\" Value json.dumps(model_dump(by_alias=True), default=str) \u2014 a byte string of JSON get_by_id hit decode_value(cached) \u2192 self.model(**data) \u2014 no DB hit get_by_id miss DB read, then the raw doc is cached (model_dump(by_alias=True)) create creates in DB, then caches the result (model_dump) update DB update, then set new value or delete the key if the doc vanished delete DB delete, then removes the key TTL config.default_ttl (default 300 s); in-memory eviction is LRU + expiry
So the cache stores JSON snapshots of whole documents \u2014 it is a read-through cache keyed by document id, not a query cache.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#intricacy-1-objectids-become-hex-strings","title":"\u26a0\ufe0f Intricacy #1 \u2014 ObjectIds Become Hex Strings","text":"
json.dumps(..., default=str) stringifies every non-JSON value \u2014 most importantly an ObjectId in a FK field:
Python
# model in memory: User(id=..., profile=ObjectId(\"507f1f77bcf86cd799439011\"))\n# cached bytes: b'{\"_id\":\"507f1f77bcf86cd799439011\",\"profile\":\"507f1f77bcf86cd799439011\", ...}'\n
On a cache hit, self.model(**data) must therefore accept a hex string where the raw doc held an ObjectId. This is fine for: - id (typed PyObjectId, which accepts both str and ObjectId), and - created_at / updated_at (ISO strings coerce to datetime).
It is not fine for a field typed as a model.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#intricacy-2-model-typed-fk-fields-fail-on-a-cache-hit","title":"\u26a0\ufe0f Intricacy #2 \u2014 Model-Typed FK Fields Fail on a Cache Hit","text":"
If your document has a populated field, e.g.:
Python
class User(BaseDocument):\n profile: Profile | None = None # populate-ruled\n
then the base-class hit path self.model(**data) receives profile=\"507f\u2026\" and Pydantic raises a ValidationError \u2014 a string cannot coerce into a Profile. The miss path fails the same way: BaseRepository.get_by_id materializes the model from the raw doc whose profile is an ObjectId \u2014 also a ValidationError. So a plain CachedBaseRepository cannot materialize a model-typed FK field at all, hit or miss \u2014 the problem is the base classes build a typed model straight from raw docs.
The tension is structural: PopulatingRepository types the field as the model; CachedBaseRepository stores the raw (depopulated) shape. You cannot have a single typed model serve both at once. The two resolutions:
Compose \u2014 keep profile: Profile | None and cache raw, populating on read. Exact recipe in use case 10: its miss path fetches the raw doc (bypassing the base model-build) and both paths normalize str \u2192 ObjectId before populate, because the JSON round-trip hands you strings.
Type it as an id \u2014 profile: PyObjectId | None and no populate rules; then cached hits validate cleanly, but you've given up population entirely.
Do not cache a populated model object through the base class: create caches result.model_dump, so the first cache write stores the embedded-dict shape while later get_by_id misses would repopulate \u2014 inconsistent shapes for the same key, and update overwrites with yet another. Pick one canonical raw shape and stick to it.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#example-inspecting-what-gets-stored","title":"\ud83d\ude80 Example \u2014 Inspecting What Gets Stored","text":"
One shared backend instance. The repository needs it (cache_backend=cache) and the registry needs it (ModelRegistry.set_cache_backend(cache)) so initialize_cache() / shutdown_cache() manage the same object. Shutdown cancels the in-memory TTL cleanup task \u2014 forgetting it leaks an asyncio.Task at app exit.
initialize_cache() starts the backend; initialize_all() creates indexes. Both come after connect().
warm_cache([ids]) skips keys that already exist, fetches the rest from the DB, and returns how many it wrote \u2014 safe to call repeatedly.
invalidate_cache(id) deletes one key; clear_pattern(\"products:*\") wipes a collection.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#redis-differences","title":"\ud83d\udd04 Redis Differences","text":"Behaviour In-memory Redis (redis.asyncio) Key prefixing prefix baked into the stored key _full_key() applied on every op TTL heap-based, lazy eviction + cleanup task Native SETEX Invalidation broadcast n/a PUBLISH on mongo_ops:cache:invalidate on delete clear_pattern prefix match on stored keys SCAN MATCH in batches"},{"location":"03_use_cases/16_cached_repository_intricacies/#choosing-the-right-layer","title":"\ud83c\udfaf Choosing the Right Layer","text":"Need Use Scalar docs, no refs \u2014 cache-first reads CachedBaseRepository (UC 07, UC 16) Refs resolved on read, no caching PopulatingRepository (UC 08, UC 15) Refs and cache-first reads composed subclass (UC 10) Bulk warm on startup / cache-then-database failover warm_cache + CacheStats"},{"location":"03_use_cases/16_cached_repository_intricacies/#related","title":"Related","text":"
"}]}
\ No newline at end of file
+{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"\ud83e\udde9 mongo-ops \u2014 Async MongoDB Operations Layer for FastAPI","text":"
mongo-ops is a modular, high-performance MongoDB operations library for FastAPI microservices. It standardizes connection lifecycle, Pydantic v2 document models, and repository-style async CRUD \u2014 with added layers for caching, reference population, and multi-document transactions. It is built on top of Motor.
Doc model: this wiki is written for humans \u2014 how\u2011to guides, examples, and testing recipes. The authoritative API contracts live in the code (docstrings) and the machine\u2011readable bundle under docs/mcp/.
mongo-ops gives FastAPI/microservice teams a small, opinionated toolkit for talking to MongoDB asynchronously. This page builds the mental model: what the layers are, in what order they must be wired up, and which repository to reach for.
"},{"location":"01_overview/#architecture","title":"\ud83c\udfd7\ufe0f Architecture","text":"Text Only
mongo-ops has a strict startup order. Violating it raises fast, loud exceptions:
Step Call Why 1 await MongoConnectionManager.connect(uri, db_name) Without a connection, get_database()/get_client() raise RuntimeError(\"Database not connected...\"). 2 await ModelRegistry.initialize_all() Create registered indexes (idempotent via create_index). 3 await ModelRegistry.set_cache_backend(...) then await ModelRegistry.initialize_cache() Must happen after connection, before any cache-backed operation. Raises RuntimeError(\"No cache backend registered...\") if skipped. 4 Use repositories Constructed repos resolve the collection from the live database. 5 await ModelRegistry.shutdown_cache() + await MongoConnectionManager.disconnect() On shutdown (in-memory TTL task cancelled; Redis pub/sub closed).
The canonical wiring is the FastAPI lifespan context manager (see use case 01).
"},{"location":"01_overview/#which-repository-should-you-use","title":"\ud83e\udded Which Repository Should You Use?","text":"Repository Use when Adds BaseRepository[T] Plain CRUD \u2014 the default nothing extra CachedBaseRepository[T] Read\u2011heavy, low\u2011write fields (lookups by _id) cache\u2011first get_by_id, cache on create, invalidate on update/delete, warm_cachePopulatingRepository[T] You return related docs (FK references) denormalized _populate on read, _depopulate on write, FK\u2011guarded patchCRUDMixin Reusing CRUD inside an existing class raw CRUD against a collection you already have
There is no built\u2011in CachedPopulatingRepository. Use case 10 shows how to compose caching + population in a small subclass.
Generic CRUD over a Motor collection. The building block of all repositories.
Method Signature Notes createasync (data: T) -> T Dumps model (excludes id, None), stamps created_at/updated_at, inserts, returns model with assigned _id. get_by_idasync (id: str \\| ObjectId) -> Optional[T]str is accepted and converted to ObjectId. get_manyasync (filter: dict \\| None = None, skip: int = 0, limit: int = 100, sort: list[tuple] \\| None = None) -> list[T] Cursor .skip().limit().sort(...) then to_list(limit). limit=0 disables the limit clause. updateasync (id, data: dict[str, Any]) -> Optional[T]$set + refreshed updated_at via find_one_and_update. patchasync (id, data: dict[str, Any]) -> Optional[T] Same as update but intended for REST PATCH semantics. deleteasync (id) -> bool True if a document was deleted. countasync (filter: dict \\| None = None) -> intcount_documents."},{"location":"02_components/#4-baserepositoryt","title":"4. BaseRepository[T]","text":"
BaseRepository(collection_name: str, model: type[T]) \u2014 resolves the collection from MongoConnectionManager.get_database()[collection_name]. Requires an active connection at construction time. Provides everything in CRUDMixin plus collection_name.
_depopulate(document) \u2014 collapses populated FK fields back to ObjectId before create/update.
create / update accept a model T (not a dict) so depopulation can run.
patch blocks FK fields \u2014 raises ValueError(\"Cannot patch FK fields via patch(): ... Use update() to change FK fields.\").
Populate semantics (important): a PopulateRule names a field that holds either an ObjectId or a list[ObjectId] and is the same field that gets replaced with the resolved document(s). There is no separate \"ref field\" vs \"target field\". See use case 08.
"},{"location":"02_components/#6-transactionmanager","title":"6. TransactionManager","text":"Method Signature Behavior start_sessionasync ctx manager (**kwargs) -> AsyncIOMotorClientSession Yields a session with an active transaction. Pass session= to every collection call inside. execute_transactionasync (operations: list[Callable[[session], Awaitable[Any]]], **kwargs) -> list[Any] Runs each op inside one transaction and returns results in order; any exception aborts the transaction and propagates."},{"location":"02_components/#7-modelregistry","title":"7. ModelRegistry","text":"
Centralized models, indexes, and cache lifecycle for multi-collection services.
Method Signature Behavior register(collection_name: str, model: type[BaseDocument], indexes: list[Any] \\| None = None) Records model + index specs. Index specs are passed as-is to pymongo create_index \u2014 single tuples, compound lists, or dicts with keys/options. initialize_allasync (db: AsyncIOMotorDatabase \\| None = None) -> Nonecreate_index per registered spec (idempotent). Uses the manager database if db omitted. get_model(collection_name) -> type[BaseDocument] Raises KeyError if unregistered. list_collections() -> list[str] Registered collection names. set_cache_backend(backend: CacheBackend) -> None Register the single shared backend. initialize_cacheasync () -> None Starts the backend (background TTL cleanup) \u2014 raises RuntimeError if no backend registered. shutdown_cacheasync () -> None Stops the backend cleanly and clears it. get_cache_backend() -> Optional[CacheBackend] Current backend, if any."},{"location":"02_components/#8-cache-layer","title":"8. Cache Layer","text":""},{"location":"02_components/#81-cachebackend-abstract","title":"8.1 CacheBackend (abstract)","text":"
Dataclass: enabled: bool = True, backend: Literal[\"memory\", \"redis\"] = \"memory\", redis_client, default_ttl: int = 300, max_entries: int = 10000, key_prefix: str = \"\", cleanup_interval: int = 60. Raises ValueError if backend=\"redis\" without a client, and ImportError if redis is not installed.
RedisCacheBackend(redis_client, key_prefix=\"\", default_ttl=300) \u2014 setex storage, SCAN-based clear_pattern, and publish_invalidate(key) for cross-service invalidation on delete via the mongo_ops:cache:invalidate channel.
Cache keys are \"{key_prefix}{id}\" (prefix defaults to \"{collection_name}:\").
get_by_id \u2014 cache-first; cache miss reads DB and stores model_dump(by_alias=True) (JSON-encoded) for default_ttl. Honors config.enabled=False (bypass).
create \u2014 inserts then caches the result.
update/delete \u2014 refresh or remove the cache entry.
warm_cache(ids) -> int \u2014 prefetch a list of IDs, returns count warmed.
invalidate_cache(id) \u2014 manual eviction.
"},{"location":"02_components/#9-population-layer","title":"9. Population Layer","text":""},{"location":"02_components/#91-populaterule","title":"9.1 PopulateRule","text":"
Dataclass:
Python
@dataclass\nclass PopulateRule:\n field_name: str # field holding the ObjectId / list[ObjectId]; replaced in-place with the resolved doc(s)\n collection_name: str # collection the references point at\n nested_rules: list[PopulateRule] | None = None\n max_depth: int = 1\n filter: dict | None = None # DECLARED but NOT yet applied by the engine\n projection: dict | None = None # DECLARED but NOT yet applied by the engine\n
\u26a0\ufe0f filter and projection are accepted but currently ignored by PopulationEngine \u2014 do not rely on them yet.
One repository per collection. Encapsulate every query the domain needs behind repository methods; keep Mongo details ($regex, $inc, projections) inside the repository.
Keep models thin. BaseDocument for the shape; use Pydantic Field constraints for validation; never put business rules in the model.
Use services for cross-repository logic. A Service composes multiple repositories (and TransactionManager) \u2014 routes stay thin.
Expose get_many(filter=..., skip=..., limit=..., sort=...) instead of raw find for list endpoints \u2014 you get controlled pagination for free.
Connect once, in the lifespan. MongoConnectionManager.lifespan(...) (or explicit connect/disconnect) \u2014 never lazily per request.
Construct repositories after connect(). Module-level Repo() before connection raises RuntimeError(\"Database not connected...\"). Use dependencies or construct inside the lifespan/request.
Order the cache lifecycle strictly: set_cache_backend(backend) \u2192 initialize_cache() (after connect, before use) \u2192 shutdown_cache() on exit.
Register all models up front via ModelRegistry.register(...) and let initialize_all() create indexes once at startup (idempotent).
"},{"location":"04_best_practices/#data-performance","title":"\ud83d\uddc4\ufe0f Data & Performance","text":"
Declare indexes for every hot query. Single-field, composite, and optioned (unique/TTL) specs all work via ModelRegistry.register \u2014 see use case 13.
Cache only hot, low-write _id reads. Use CachedBaseRepository for lookups-by-id; invalidate (update/delete handle it) and pick a sensible default_ttl.
Populate at the repository boundary. PopulatingRepository resolves refs on read and depopulates on write; do not hand-roll joins in endpoints.
Respect the populate invariants: rules name the field that holds the reference and that is replaced; patch() cannot touch FK fields \u2014 use update().
Use transactions for multi-document writes. TransactionManager.start_session (inline) or execute_transaction (list of ops) \u2014 and pass session= to every collection call inside.
Handle the library's real exceptions at the edges: DuplicateKeyError \u2192 409, InvalidId \u2192 400, CircularReferenceError \u2192 409, ValueError guides \u2192 400/422 (see Error Handling).
Default to mock-based unit tests. Patch MongoConnectionManager.get_database, use AsyncMock collections and cursor chains \u2014 the whole suite runs without MongoDB (use case 14).
Mirror the library tests. tests/test_{repository,populating_repository,cache,registry,transactions}.py are canonical examples of every pattern above.
Use type hints end-to-end \u2014 mypy-gated CI (see pyproject) catches drift early.
What can raise, what it means, and how to map it in a FastAPI app.
"},{"location":"06_error_handling/#library-raised-exceptions","title":"\ud83d\udccb Library-Raised Exceptions","text":"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 \u2014 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 \u2014 use update() with a model. ValueError(\"...contains embedded dict(s) \u2014 run repair script\")_populate on read A FK field holds an embedded document instead of an ObjectId \u2014 migrate the data. ValueError(\"...contains ObjectId \u2014 was populate skipped?\")_depopulate on write A populate-ruled field is still an ObjectId at depopulate time \u2014 the read must have populated it first. CircularReferenceError(collection, doc_id, path)PopulationEngine.populate A (Class, id) pair was revisited \u2014 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'\")CacheConfigbackend=\"redis\" without a client. Pymongo DuplicateKeyError any insert/update Unique index violation (e.g., duplicate email). bson.errors.InvalidIdObjectId(...) 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 \u2014 they fail loudly at startup or first call instead of misbehaving silently.
from mongo_ops.repository import MongoConnectionManager \u2014 patch where it is used (mongo_ops.repository.MongoConnectionManager), matching the library's own tests.
pytest-asyncio runs as auto mode per pyproject.toml, so @pytest.mark.asyncio works without extra config.
For population, cache, registry, and transaction mockups \u2014 see the full testing guide.
"},{"location":"03_use_cases/01_basic_crud/","title":"Use Case 1: Basic FastAPI CRUD API","text":"
Scenario: A simple user management API with CRUD endpoints, index registration, and a correct connection lifecycle.
"},{"location":"03_use_cases/01_basic_crud/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description MongoConnectionManager.lifespan Connect on startup, disconnect on shutdown. BaseRepository[User] Generic CRUD \u2014 no endpoint-level Mongo code. ModelRegistry.register Declares the email index; initialize_all creates it at startup."},{"location":"03_use_cases/01_basic_crud/#example","title":"\ud83d\ude80 Example","text":"Python
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 \u2014 it accepts a partial dict like update, but PopulatingRepository rejects FK fields.
Note: this snippet omits the FastAPI lifespan connection wiring for brevity \u2014 copy it from use case 01 so ProductRepository() is created only after MongoConnectionManager.connect().
Methods that hit self.collection directly (regex search, $inc) bypass the caching and population layers. If a feature composes them \u2014 extend CachedBaseRepository or PopulatingRepository instead and add the domain methods there.
Prefer get_many(filter=...) over raw find when you want pagination/sort defaults for free.
Reuse self.model(**doc) to convert raw dicts to model instances consistently.
"},{"location":"03_use_cases/03_transactions/","title":"Use Case 3: Transaction Support for Multi-Document Operations","text":"
Scenario: Order processing must update inventory and create an order atomically. Any failure rolls both back.
"},{"location":"03_use_cases/03_transactions/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description TransactionManager.start_session Async context manager yielding a session with an active transaction. session= kwarg Pass to every insert_one / update_one / find_one inside the block."},{"location":"03_use_cases/03_transactions/#example","title":"\ud83d\ude80 Example","text":"Python
"},{"location":"03_use_cases/05_soft_deletes/","title":"Use Case 5: Soft Deletes Pattern","text":"
Scenario: Deleting a task should be recoverable. Instead of removing the document, set a tombstone flag and filter it from normal queries.
"},{"location":"03_use_cases/05_soft_deletes/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description SoftDeleteDocument Base model carrying is_deleted, deleted_at, deleted_by. SoftDeleteRepository[T] Repository-level soft delete / restore / filtering using update + get_many. permanent_delete Escapes to the real delete."},{"location":"03_use_cases/05_soft_deletes/#example","title":"\ud83d\ude80 Example","text":"Python
BaseRepository[T] is generic \u2014 subclasses must parameterize it (see SoftDeleteRepository[T] above). from mongo_ops import BaseRepository; BaseRepository[T] works out of the box.
Keep a compound index on {\"is_deleted\": 1, \"status\": 1} for active-list queries (see use case 13).
Soft-deleted documents should be excluded at the repository boundary, never re-filtered ad hoc in endpoints.
"},{"location":"03_use_cases/06_multi_model/","title":"Use Case 6: Multi-Model Service with Registration","text":"
Scenario: A social app manages users, posts, and comments. Each has its own model, repository, and indexes \u2014 registered centrally and initialized at startup.
"},{"location":"03_use_cases/06_multi_model/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description ModelRegistry.register One call per collection \u2014 model + indexes together. ModelRegistry.initialize_all Creates every registered index at startup (idempotent). Repositories One repository class per collection, all sharing the same connection."},{"location":"03_use_cases/06_multi_model/#example","title":"\ud83d\ude80 Example","text":"Python
Place user_repo = ... inside the lifespan/after connect. Module-level instantiation before connect() raises RuntimeError(\"Database not connected...\").
Model relationships here are plain ObjectId strings stored on the child docs. To resolve them on read, see use case 08 \u2013 Population.
Registering indexes on created_at/author_id/post_id keeps the common queries indexed (see use case 13).
"},{"location":"03_use_cases/07_caching/","title":"Use Case 7: Caching for High-Performance Reads","text":"
Scenario: A read-heavy API (product catalog) reduces DB load by caching documents by _id \u2014 in-memory locally, or shared via Redis.
"},{"location":"03_use_cases/07_caching/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description InMemoryCacheBackend TTL + LRU cache with a background cleanup task. RedisCacheBackend Distributed cache on redis.asyncio with pub/sub invalidation. CachedBaseRepository[T] Extends BaseRepository \u2014 cache-first get_by_id, cache on create, invalidate on update/delete, warm_cache(ids), invalidate_cache(id). Backend lifecycle The same backend instance must be both passed to the repository AND registered via ModelRegistry.set_cache_backend so initialize_cache() starts its task."},{"location":"03_use_cases/07_caching/#example","title":"\ud83d\ude80 Example","text":"Python
from contextlib import asynccontextmanager\n\nfrom fastapi import FastAPI, HTTPException\nfrom mongo_ops import BaseDocument, CachedBaseRepository, ModelRegistry, MongoConnectionManager\nfrom mongo_ops.cache import CacheConfig, InMemoryCacheBackend\n\n\nclass Product(BaseDocument):\n name: str = \"\"\n price: float = 0.0\n\n\n# One shared backend \u2014 used by both the repository and the registry lifecycle.\ncache = InMemoryCacheBackend(max_entries=10_000, default_ttl=300)\nModelRegistry.set_cache_backend(cache)\n\n\nclass ProductRepo(CachedBaseRepository[Product]):\n def __init__(self):\n super().__init__(\n collection_name=\"products\",\n model=Product,\n cache_backend=cache,\n config=CacheConfig(enabled=True, backend=\"memory\"),\n )\n\n\n@asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\", db_name=\"shop\"\n ):\n await ModelRegistry.initialize_all()\n await ModelRegistry.initialize_cache() # starts the TTL cleanup task\n yield\n await ModelRegistry.shutdown_cache() # cancels it on exit\n\n\napp = FastAPI(lifespan=lifespan)\n\n\n@app.post(\"/products/\", response_model=Product)\nasync def create_product(product: Product):\n return await ProductRepo().create(product) # created after connect()\n\n\n@app.get(\"/products/{product_id}\", response_model=Product)\nasync def get_product(product_id: str):\n product = await ProductRepo().get_by_id(product_id)\n if not product:\n raise HTTPException(status_code=404, detail=\"Product not found\")\n return product\n\n\n@app.put(\"/products/{product_id}\", response_model=Product)\nasync def update_product(product_id: str, name: str | None = None, price: float | None = None):\n data = {}\n if name is not None:\n data[\"name\"] = name\n if price is not None:\n data[\"price\"] = price\n return await ProductRepo().update(product_id, data)\n\n\n# Outside request handlers:\n# await ProductRepo().warm_cache([object_id_1, object_id_2]) # pre-load n ids -> int\n# await ProductRepo().invalidate_cache(object_id_3) # manual eviction\n
Cache keys are \"{key_prefix}{id}\" (default prefix \"products:\"). clear_pattern(\"products:*\") wipes a whole collection's entries.
Values are JSON-encoded (json.dumps(obj, default=str)) \u2014 nested models inside a cached document are stored as dicts, not objects.
Enable/disable per repository with CacheConfig(enabled=False); a disabled repo bypasses the cache entirely.
Distinguish the two initialize* calls: initialize_cache() starts the backend task; initialize_all() creates indexes. Both belong in the lifespan, after connect().
"},{"location":"03_use_cases/08_population/","title":"Use Case 8: Document Population","text":"
Scenario: An API returns a User with its related Profile embedded in one JSON payload \u2014 no second round-trip from the client, no joins.
"},{"location":"03_use_cases/08_population/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description PopulateRule Declares which field to resolve. field_name holds the ObjectId (or list[ObjectId]) and is the same field the resolved document replaces in place. collection_name says where the referenced documents live. PopulationEngine Holds repositories by collection name and resolves rules recursively, detecting cycles (CircularReferenceError). PopulatingRepository[T]get_by_id/get_many populate on read; create/update depopulate on write; patch rejects FK fields.
\u26a0\ufe0f PopulateRule does not have a separate \"ref field\" vs \"target field\" \u2014 the ref field is the populated field. filter/projection on PopulateRule are declared but not yet applied by the engine.
Patching an FK field raises ValueError \u2014 switch FK changes to update(user_id, model) instead. See use case 05 for the reasoning with soft deletes.
Missing references resolve to None, not an error.
A field holding an embedded dict (instead of an ObjectId) raises a ValueError (\"run repair script\") \u2014 migrate embedded docs to a separate collection first.
If create/update receives a field that is already an ObjectId under a populate rule, _depopulate raises ValueError(\"...was populate skipped?\") \u2014 populate-then-depopulate pairs must be balanced.
09 \u2013 Advanced population \u00b7 10 \u2013 Cache + population \u00b7 14 \u2013 Testing guide
"},{"location":"03_use_cases/09_advanced_population/","title":"Use Case 9: Nested Document Population & Circular-Ref Handling","text":"
Scenario: An Author has books (a list of references), each Book references a publisher, and an Author may reference a mentor \u2014 which is another Author (a potential cycle).
"},{"location":"03_use_cases/09_advanced_population/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description nested_rules Populates deeper levels: resolve books on Author, then publisher inside each Book. max_depth Per-rule recursion bound \u2014 the safety net for cyclic graphs. global_max_depthPopulationEngine(repos, global_max_depth=10) global cap. CircularReferenceError Raised when a (Class, id) pair is revisited; carries the visited path."},{"location":"03_use_cases/09_advanced_population/#example","title":"\ud83d\ude80 Example","text":"Python
from contextlib import asynccontextmanager\n\nfrom bson import ObjectId\nfrom fastapi import FastAPI, HTTPException\nfrom mongo_ops import BaseDocument, ModelRegistry, MongoConnectionManager, PopulatingRepository\nfrom mongo_ops.cache import CircularReferenceError\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\n\n\n# 1. Models \u2014 each ref field holds ObjectId(s) in the DB and becomes model(s) in memory.\nclass Publisher(BaseDocument):\n name: str = \"\"\n country: str = \"\"\n\n\nclass Book(BaseDocument):\n title: str = \"\"\n publisher: Publisher | None = None # ObjectId in DB, Publisher in memory\n\n\nclass Author(BaseDocument):\n name: str = \"\"\n books: list[Book] | None = None # list[ObjectId] in DB, list[Book] in memory\n mentor: \"Author\" | None = None # self-reference \u2014 potential cycle\n\n\n# 2. Engine + nested rules.\nengine = PopulationEngine({})\n\npublisher_rule = PopulateRule(\n field_name=\"publisher\",\n collection_name=\"publishers\",\n)\n\nbook_rule = PopulateRule(\n field_name=\"books\",\n collection_name=\"books\",\n nested_rules=[publisher_rule], # fetch each book, then its publisher\n max_depth=3,\n)\n\nmentor_rule = PopulateRule(\n field_name=\"mentor\",\n collection_name=\"authors\",\n max_depth=2, # stops mentor chains early \u2014 also avoids unbounded cycles\n)\n\n\nclass AuthorRepository(PopulatingRepository[Author]):\n def __init__(self):\n super().__init__(\n collection_name=\"authors\",\n model=Author,\n population_engine=engine,\n populate_rules=[book_rule, mentor_rule],\n )\n\n\n@asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\", db_name=\"library\"\n ):\n engine.register_repo(\"publishers\", PopulatingRepository[Publisher](\"publishers\", Publisher))\n engine.register_repo(\"books\", PopulatingRepository[Book](\"books\", Book))\n engine.register_repo(\"authors\", AuthorRepository())\n await ModelRegistry.initialize_all()\n yield\n\n\napp = FastAPI()\n\n\n@app.get(\"/authors/{author_id}\")\nasync def get_author(author_id: str):\n try:\n author = await AuthorRepository().get_by_id(author_id)\n except CircularReferenceError as exc:\n raise HTTPException(status_code=409, detail=f\"Circular reference: {exc}\")\n if not author:\n raise HTTPException(status_code=404, detail=\"Author not found\")\n return author\n
08 \u2013 Population \u00b7 10 \u2013 Cache + population \u00b7 Error Handling
"},{"location":"03_use_cases/10_cache_and_population/","title":"Use Case 10: Caching + Population (Read-Through, Populated on Read)","text":"
Scenario: A service needs fast reads of a User and its related Profile. We want the cache-hit path to skip the DB entirely while still returning a populated document.
"},{"location":"03_use_cases/10_cache_and_population/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description CachedBaseRepository Provides the cache layer (cache-first get_by_id, invalidation on update/delete). PopulationEngine Provides the reference resolution. Composed subclass A small CachedBaseRepository subclass adds two _populate helpers so cache hits return populated models.
\u26a0\ufe0f Known limitation: there is no built-in CachedPopulatingRepository in the library. This use case documents the composition. We cache the raw (depopulated) document \u2014 references stay ObjectIds until read time, so Profile changes are reflected on the next fetch (within TTL) and update/delete invalidation remains correct.
from contextlib import asynccontextmanager\nfrom datetime import datetime\nfrom typing import Any, Union\n\nfrom bson import ObjectId\nfrom fastapi import FastAPI\nfrom mongo_ops import BaseDocument, CachedBaseRepository, ModelRegistry, MongoConnectionManager\nfrom mongo_ops.cache import CacheConfig, InMemoryCacheBackend\nfrom mongo_ops.cache.in_memory import decode_value, encode_value\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\n\n\n# 1. Models \u2014 `profile` holds an ObjectId in DB, a Profile in memory.\nclass Profile(BaseDocument):\n avatar_url: str = \"\"\n bio: str = \"\"\n\n\nclass User(BaseDocument):\n username: str = \"\"\n email: str = \"\"\n profile: Profile | None = None\n\n\n# 2. Engine + rule.\nengine = PopulationEngine({})\nprofile_rule = PopulateRule(field_name=\"profile\", collection_name=\"profiles\")\n\n\n# 3. Composed repository.\nclass CachedUserRepository(CachedBaseRepository[User]):\n def __init__(\n self,\n cache_backend: InMemoryCacheBackend,\n population_engine: PopulationEngine,\n populate_rules: list[PopulateRule],\n config: CacheConfig | None = None,\n ):\n super().__init__(\"users\", User, cache_backend, config)\n self.population_engine = population_engine\n self._populate_rules = populate_rules\n\n async def _populate(self, data: dict[str, Any]) -> dict[str, Any]:\n \"\"\"Dict-level resolution (mirrors PopulatingRepository._populate).\"\"\"\n for rule in self._populate_rules:\n ref = data.get(rule.field_name)\n if ref is None:\n continue\n if isinstance(ref, str): # JSON round-trip turns ObjectId into hex str\n ref = ObjectId(ref)\n if isinstance(ref, list):\n resolved = []\n for item in ref:\n item = ObjectId(item) if isinstance(item, str) else item\n repo = self.population_engine._repos.get(rule.collection_name)\n doc = await repo.get_by_id(item) if repo else None\n resolved.append(doc)\n data[rule.field_name] = resolved\n elif isinstance(ref, ObjectId):\n repo = self.population_engine._repos.get(rule.collection_name)\n data[rule.field_name] = await repo.get_by_id(ref) if repo else None\n return data\n\n async def _depopulate(self, data: User) -> dict[str, Any]:\n \"\"\"Model -> raw dict, FK fields collapsed to ObjectId (mirrors _depopulate).\"\"\"\n doc = data.model_dump(exclude={\"id\"}, exclude_none=True)\n for rule in self._populate_rules:\n value = getattr(data, rule.field_name, None)\n if isinstance(value, list):\n doc[rule.field_name] = [item.id for item in value]\n elif isinstance(value, BaseDocument):\n doc[rule.field_name] = value.id\n return doc\n\n async def create(self, data: User) -> User:\n doc = await self._depopulate(data) # store FK refs as ObjectIds\n doc[\"created_at\"] = datetime.utcnow()\n doc[\"updated_at\"] = datetime.utcnow()\n result = await self.collection.insert_one(doc)\n doc[\"_id\"] = result.inserted_id\n if self._cache_config.enabled:\n await self._cache.set(self._cache_key(doc[\"_id\"]), encode_value(doc), self._cache_config.default_ttl)\n return self.model(**await self._populate(doc))\n\n async def update(self, id: Union[str, ObjectId], data: User) -> User | None:\n if isinstance(id, str):\n id = ObjectId(id)\n doc = await self._depopulate(data) # update() takes a full model here\n doc[\"updated_at\"] = datetime.utcnow()\n result = await self.collection.find_one_and_update(\n {\"_id\": id}, {\"$set\": doc}, return_document=True\n )\n key = self._cache_key(id)\n if result is None:\n if self._cache_config.enabled:\n await self._cache.delete(key)\n return None\n raw = dict(result)\n if self._cache_config.enabled:\n await self._cache.set(key, encode_value(raw), self._cache_config.default_ttl)\n return self.model(**await self._populate(raw))\n\n async def get_by_id(self, id: Union[str, ObjectId]) -> User | None:\n if not self._cache_config.enabled:\n # No cache: still a raw fetch + populate \u2014 the base get_by_id()\n # would build a typed model from the raw doc and fail on FK fields.\n raw = await self.collection.find_one({\"_id\": ObjectId(id) if isinstance(id, str) else id})\n return self.model(**await self._populate(raw)) if raw else None\n\n cached = await self._cache.get(self._cache_key(id))\n if cached is not None:\n # Cache hit: decode the RAW doc, then populate before returning.\n data = await self._populate(decode_value(cached))\n return self.model(**data)\n\n # Cache miss: one RAW DB read. The base get_by_id() would rebuild the\n # model from the raw doc, which fails for model-typed FK fields \u2014\n # so fetch the raw dict, cache it, and populate before materializing.\n key = self._cache_key(id)\n raw = await self.collection.find_one({\"_id\": ObjectId(id) if isinstance(id, str) else id})\n if raw is None:\n return None\n await self._cache.set(key, encode_value(raw), self._cache_config.default_ttl)\n return self.model(**await self._populate(raw))\n\n\n# 4. Wire-up \u2014 one backend for both the repo and the registry lifecycle.\ncache = InMemoryCacheBackend(max_entries=20_000, default_ttl=600)\nModelRegistry.set_cache_backend(cache)\n\n\n@asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\", db_name=\"app_db\"\n ):\n # Register the repositories the engine resolves refs against:\n # engine.register_repo(\"profiles\", ProfileRepo())\n engine.register_repo(\"users\", CachedUserRepository(cache, engine, [profile_rule]))\n await ModelRegistry.initialize_all()\n await ModelRegistry.initialize_cache()\n yield\n await ModelRegistry.shutdown_cache()\n\n\napp = FastAPI(lifespan=lifespan)\n
Note: replace the placeholder engine.register_repo(\"profiles\", ...) line with a repository for Profile. A matching CachedBaseRepository[Profile](\"profiles\", Profile, cache) keeps profile reads cached too \u2014 e.g. engine.register_repo(\"profiles\", CachedBaseRepository[Profile](\"profiles\", Profile, cache)).
Cache the raw doc, populate at read time. Ref fields stay ObjectIds in the cache; JSON round-trips them to hex strings, so _populate must normalize str \u2192 ObjectId (shown above).
update/create above are model-based (update(id, User)), mirroring PopulatingRepository \u2014 they depopulate before writing and cache a raw snapshot. The plain CachedBaseRepository.update(id, dict) and delete(id) keep working and invalidate the same key.
Set a sensible default_ttl \u2014 cached User entries resolve Profile on each read, so profile edits show up within the TTL (or call invalidate_cache(user_id) explicitly).
To reuse this compose logic across many collections, extract the _populate / _depopulate helpers plus the create / update / get_by_id overrides into a mixin and parameterize the rules per subclass.
"},{"location":"03_use_cases/11_cache_lifecycle/","title":"Use Case 11: Proper Cache Lifecycle in FastAPI","text":"
Scenario: A microservice uses an in-memory (or Redis) cache backend and must start the cleanup task on app start, then shut it down cleanly on termination.
"},{"location":"03_use_cases/11_cache_lifecycle/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description ModelRegistry.set_cache_backend(backend) Registers the single shared backend. ModelRegistry.initialize_cache() Starts the backend (spawns the in-memory TTL cleanup task). Raises RuntimeError if no backend was registered. ModelRegistry.shutdown_cache() Cancels the cleanup task, closes pub/sub, clears the backend. If a repo uses a backend directly Same instance must be registered so initialize_cache starts its task."},{"location":"03_use_cases/11_cache_lifecycle/#example","title":"\ud83d\ude80 Example","text":"Python
from contextlib import asynccontextmanager\n\nfrom fastapi import FastAPI\nfrom mongo_ops import ModelRegistry, MongoConnectionManager\nfrom mongo_ops.cache import InMemoryCacheBackend\n\ncache = InMemoryCacheBackend(max_entries=10_000, default_ttl=300)\nModelRegistry.set_cache_backend(cache) # before any cache-backed repo is used\n\n\n@asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\", db_name=\"mydb\"\n ):\n await ModelRegistry.initialize_all() # create indexes (idempotent)\n await ModelRegistry.initialize_cache() # start the TTL cleanup task\n yield\n await ModelRegistry.shutdown_cache() # cancel task + close cleanly\n\n\napp = FastAPI(lifespan=lifespan)\n
"},{"location":"03_use_cases/11_cache_lifecycle/#what-actually-happens","title":"\ud83d\udd01 What Actually Happens","text":"
InMemoryCacheBackend.initialize() spawns an asyncio.Task that evicts expired entries every cleanup_interval seconds. Without a shutdown, the event loop flags the dangling task on exit \u2014 shutdown_cache() cancels it and awaits it.
shutdown_cache() also NULs the registry cache backend and closes the Redis pub/sub handle (if Redis).
If initialize_cache() is called before set_cache_backend(), it raises:
Text Only
RuntimeError: No cache backend registered. Call set_cache_backend() first.\n
Per-repository toggling is independent of the lifecycle: CacheConfig(enabled=False) bypasses the cache for that repo even after initialize_cache().
The cleanup task uses cleanup_interval seconds for scans; entries also expire on access via the TTL heap (default_ttl=0 expires immediately).
Register the backend before constructing any CachedBaseRepository that references it \u2014 otherwise a repo may hold an uninitialized backend (no cleanup task, no Redis pub/sub).
"},{"location":"03_use_cases/12_transaction_helper/","title":"Use Case 12: Using TransactionManager.execute_transaction","text":"
Scenario: Perform several writes across different collections atomically \u2014 e.g., create an Order and decrement Inventory. The low-level start_session context works, but execute_transaction collects results from a list of async operations.
"},{"location":"03_use_cases/12_transaction_helper/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description TransactionManager.execute_transaction Runs a list of async callables (each receives a session) inside one transaction; returns the result of each callable, in order. Automatic rollback Any raised exception aborts the transaction and propagates to the caller. start_session The underlying async context manager async with TransactionManager.start_session() as session:."},{"location":"03_use_cases/12_transaction_helper/#example","title":"\ud83d\ude80 Example","text":"Python
Return values: each callable can return whatever you need; results are collected in the same order.
Errors: raise inside any callable \u2192 the whole transaction aborts (session rolls back) and the exception propagates.
Reads inside a transaction: pass session=session to find_one/find too.
Testing (no Mongo): monkeypatch a fake client on mongo_ops.transactions.MongoConnectionManager.get_client and stub start_session \u2014 see tests/test_transactions.py.
Repo models still carry created_at/updated_at; for raw collection inserts inside the transaction you set them manually (as shown).
"},{"location":"03_use_cases/13_index_creation/","title":"Use Case 13: Declaring Indexes (Single-Field, Composite, Unique)","text":"
Scenario: Ensure each collection has the right indexes for fast queries and data integrity \u2014 declared in one place and created at startup.
"},{"location":"03_use_cases/13_index_creation/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description ModelRegistry.register(indexes=...) Each spec is passed as-is to pymongo create_index. Supported forms: tuple (\"field\", direction), compound list [(\"a\", 1), (\"b\", -1)], or a dict with keys + options. ModelRegistry.initialize_all Creates every registered index during startup (idempotent \u2014 create_index skips existing indexes)."},{"location":"03_use_cases/13_index_creation/#example","title":"\ud83d\ude80 Example","text":"Python
from contextlib import asynccontextmanager\n\nfrom pymongo import ASCENDING, DESCENDING\nfrom mongo_ops import BaseDocument, ModelRegistry, MongoConnectionManager\n\n\nclass User(BaseDocument):\n username: str = \"\"\n email: str = \"\"\n\n\nclass BlogPost(BaseDocument):\n author_id: str = \"\"\n created_at: str = \"\"\n title: str = \"\"\n\n\nclass Passenger(BaseDocument):\n email: str = \"\"\n seat: str = \"\"\n\n\n# 1\ufe0f\u20e3 Single-field index \u2014 email lookups.\nModelRegistry.register(\n collection_name=\"users\",\n model=User,\n indexes=[(\"email\", ASCENDING)],\n)\n\n# 2\ufe0f\u20e3 Composite index \u2014 queries filtering by author + creation date.\nModelRegistry.register(\n collection_name=\"posts\",\n model=BlogPost,\n indexes=[[(\"author_id\", ASCENDING), (\"created_at\", DESCENDING)]],\n)\n\n# 3\ufe0f\u20e3 Unique index with a custom name \u2014 enforce unique emails.\nModelRegistry.register(\n collection_name=\"passengers\",\n model=Passenger,\n indexes=[\n {\n \"keys\": [(\"email\", ASCENDING)],\n \"options\": {\"unique\": True, \"name\": \"uq_passenger_email\"},\n }\n ],\n)\n\n\n# 4\ufe0f\u20e3 Everything is created on startup.\n@asynccontextmanager\nasync def lifespan(_app):\n async with MongoConnectionManager.lifespan(\n uri=\"mongodb://localhost:27017\",\n db_name=\"mydb\",\n ):\n await ModelRegistry.initialize_all()\n yield\n
Every spec passes through to collection.create_index(spec) \u2014 so MongoDB options like unique, sparse, and expireAfterSeconds (TTL) belong in the options dict.
Idempotent by construction: create_index is a no-op when a same-shape index already exists.
Verify with the Mongo shell:
JavaScript
db.<collection>.getIndexes()\n
A unique index on an already-duplicated field will fail with DuplicateKeyError on startup \u2014 clean the data first.
Scenario: Write unit tests that never touch a real MongoDB \u2014 mock the collection, the cache backend, and the client, exactly like the library's own test suite (tests/).
"},{"location":"03_use_cases/14_testing_guide/#whats-new","title":"\ud83d\udce6 What's New?","text":"Component Description AsyncMock collections Stub find_one, insert_one, find_one_and_update, cursor chains. Patching MongoConnectionManager.get_database Gives repositories a mock collection without a live connection. monkeypatch on get_client Fakes sessions for TransactionManager tests. Reference tests tests/test_populating_repository.py, tests/test_cache.py, tests/test_registry.py, tests/test_transactions.py."},{"location":"03_use_cases/14_testing_guide/#example-boilerplate","title":"\ud83d\ude80 Example Boilerplate","text":"Python
import pytest\nfrom unittest.mock import AsyncMock, MagicMock, patch\n\nfrom bson import ObjectId\nfrom mongo_ops.cache import CacheConfig, InMemoryCacheBackend\nfrom mongo_ops.cache.repository import CachedBaseRepository\nfrom mongo_ops.models import BaseDocument\nfrom mongo_ops.registry import ModelRegistry\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\nfrom mongo_ops.repository import PopulatingRepository\n\n\n# ----------------------------------------------------------------------\n# 1. Models (same shape as the library tests)\n# ----------------------------------------------------------------------\nclass Profile(BaseDocument):\n avatar_url: str = \"\"\n\n\nclass User(BaseDocument):\n name: str = \"\"\n profile: Profile | None = None # ObjectId in DB, Profile in memory\n\n\n# ----------------------------------------------------------------------\n# 2. Fixtures\n# ----------------------------------------------------------------------\n@pytest.fixture\ndef mock_collection():\n return AsyncMock()\n\n\n@pytest.fixture\ndef engine():\n profile_repo = AsyncMock()\n return PopulationEngine({\"profiles\": profile_repo})\n\n\n@pytest.fixture\ndef repo(mock_collection, engine):\n with patch(\"mongo_ops.repository.MongoConnectionManager.get_database\") as mock_db:\n mock_db.return_value.__getitem__.return_value = mock_collection\n r = PopulatingRepository(\n \"users\",\n User,\n population_engine=engine,\n populate_rules=[PopulateRule(field_name=\"profile\", collection_name=\"profiles\")],\n )\n r.collection = mock_collection\n return r\n\n\n# ----------------------------------------------------------------------\n# 3. Population \u2014 get_by_id resolves the reference\n# ----------------------------------------------------------------------\n@pytest.mark.asyncio\nasync def test_get_by_id_populates(repo, mock_collection, engine):\n uid, pid = ObjectId(), ObjectId()\n mock_collection.find_one.return_value = {\n \"_id\": uid,\n \"name\": \"Alice\",\n \"profile\": pid, # ObjectId stored in DB\n \"created_at\": \"2024-01-01T00:00:00\",\n \"updated_at\": \"2024-01-01T00:00:00\",\n }\n engine._repos[\"profiles\"].get_by_id.return_value = Profile(id=pid, avatar_url=\"pic.png\")\n\n result = await repo.get_by_id(uid)\n\n assert result is not None\n assert result.name == \"Alice\"\n assert isinstance(result.profile, Profile)\n assert result.profile.avatar_url == \"pic.png\"\n\n\n# ----------------------------------------------------------------------\n# 4. Patch FK guard\n# ----------------------------------------------------------------------\n@pytest.mark.asyncio\nasync def test_patch_rejects_fk_field(repo):\n with pytest.raises(ValueError, match=\"Cannot patch FK fields\"):\n await repo.patch(ObjectId(), {\"profile\": ObjectId()})\n\n\n# ----------------------------------------------------------------------\n# 5. Registry \u2014 index specs pass through to create_index\n# ----------------------------------------------------------------------\n@pytest.mark.asyncio\nasync def test_initialize_all_creates_indexes():\n ModelRegistry.register(\"users\", User, indexes=[(\"email\", 1)])\n\n fake_collection = AsyncMock()\n await ModelRegistry.initialize_all(db={\"users\": fake_collection})\n\n fake_collection.create_index.assert_awaited_once_with((\"email\", 1))\n\n\n# ----------------------------------------------------------------------\n# 6. Cached repository \u2014 cache-first reads + invalidation\n# ----------------------------------------------------------------------\n@pytest.mark.asyncio\nasync def test_cached_get_by_id_populates_cache():\n backend = InMemoryCacheBackend(\n max_entries=100, default_ttl=300, cleanup_interval=9999\n )\n await backend.initialize()\n try:\n with patch(\"mongo_ops.repository.MongoConnectionManager.get_database\") as mock_db:\n mock_collection = AsyncMock()\n mock_db.return_value.__getitem__.return_value = mock_collection\n repo = CachedBaseRepository(\n \"users\", User, backend, CacheConfig(enabled=True)\n )\n repo.collection = mock_collection\n\n oid = ObjectId()\n mock_collection.find_one.return_value = {\n \"_id\": oid,\n \"name\": \"cached\",\n \"created_at\": \"2024-01-01T00:00:00\",\n \"updated_at\": \"2024-01-01T00:00:00\",\n }\n\n first = await repo.get_by_id(oid)\n assert first is not None\n\n mock_collection.find_one.return_value = None # DB now \"empty\"\n second = await repo.get_by_id(oid) # served from cache\n\n assert second is not None\n assert second.name == \"cached\"\n mock_collection.find_one.assert_awaited_once() # only one DB read\n finally:\n await backend.shutdown()\n\n\n# ----------------------------------------------------------------------\n# 7. Transactions \u2014 fake the client's start_session\n# ----------------------------------------------------------------------\nfrom mongo_ops.transactions import TransactionManager\n\n\n@pytest.mark.asyncio\nasync def test_execute_transaction(monkeypatch):\n session_ctx = AsyncMock()\n session_ctx.__aenter__.return_value = AsyncMock()\n session_ctx.__aexit__.return_value = None\n\n client = MagicMock()\n client.start_session = AsyncMock(return_value=session_ctx)\n\n # NOTE: start_transaction must return a context manager, not a coroutine.\n async_session = session_ctx.__aenter__.return_value\n async_session.start_transaction = lambda **_: session_ctx\n\n monkeypatch.setattr(\n \"mongo_ops.transactions.MongoConnectionManager.get_client\",\n lambda: client,\n )\n\n async def fake_op(session):\n return \"ok\"\n\n results = await TransactionManager.execute_transaction([fake_op])\n assert results == [\"ok\"]\n
pytest-asyncio is already configured in pyproject.toml (asyncio_mode = \"auto\"), so @pytest.mark.asyncio tests work out of the box. Run with pytest (coverage reports are enabled there too).
Never hit the network. Keep the patches in fixtures (or a conftest.py) and reuse them.
Mock cursor chains with MagicMock() + .to_list = AsyncMock(...), exactly like tests/test_repository.py.
For an optional integration check (real Mongo), use MongoConnectionManager.lifespan against a local replica set and drop the test database in teardown \u2014 keep it separate from the unit suite.
The pattern works symmetrically for Redis: mock the RedisCacheBackend methods (get, set, delete) \u2014 no Redis process required.
"},{"location":"03_use_cases/15_populating_repository_wiring/","title":"Use Case 15: Inside PopulatingRepository \u2014 the Object \u21c4 ObjectId Lifecycle","text":"
Scenario: You want to see what the repository actually does, in which order, before you trust it with your data \u2014 how a Profile becomes an ObjectId for storage and comes back as a Profile on read, and where every piece is wired.
"},{"location":"03_use_cases/15_populating_repository_wiring/#two-representations-one-field","title":"\ud83d\udce6 Two Representations, One Field","text":"
A populate-ruled field is a shape shifter \u2014 the same name holds different things depending on where you look:
Place Value held in the field In the MongoDB doc ObjectId (or list[ObjectId]) In the cache/JSON hex string (see use case 16) In the app model the referenced model (or None)
PopulatingRepository.create / update depopulate (model \u2192 ObjectId) before writing; get_by_id / get_many populate (ObjectId \u2192 model) after reading:
"},{"location":"03_use_cases/15_populating_repository_wiring/#write-path-create-step-by-step","title":"\ud83d\ude80 Write Path \u2014 create() Step by Step","text":"
Given user = User(username=\"alice\", profile=saved_profile) where saved_profile.id exists:
Python
class UserRepository(PopulatingRepository[User]):\n def __init__(self):\n super().__init__(\n collection_name=\"users\",\n model=User,\n population_engine=engine,\n populate_rules=[profile_rule],\n )\n
Guard check (_depopulate, repository.py): for each rule field, if the value is already an ObjectId (or list[ObjectId]) an error is raised \u2014 see _depopulate guards below. Saved models pass.
engine.depopulate(user, rules) collapses the graph in place:
saved_profile (a BaseDocument with .id) \u2192 becomes saved_profile.id \u2192 ObjectId
a Profile without .id (unsaved) \u2192 is left as a model object, which model_dump then embeds as a dict \u2014 see intricacy #3
when nested_rules are present, the nested model is depopulated recursively first, then collapsed
model_dump(exclude={\"id\"}, exclude_none=True) produces the raw insert dict; timestamps are added.
insert_one(doc) writes {..., \"profile\": ObjectId(\"...\"), ...} to MongoDB.
On the way out, data_to_model runs _populate (the read path below) so create returns a fully populated model, not the raw one.
"},{"location":"03_use_cases/15_populating_repository_wiring/#the-_depopulate-guards-loud-failures-silent-corruption","title":"The _depopulate guards (loud failures > silent corruption)","text":"Stored/held value under a rule field Behaviour ObjectId / list[ObjectId]ValueError(\"...contains ObjectId \u2014 was populate skipped?\") \u2014 the doc was loaded raw (e.g. via BaseRepository or an unpopulated read) and handed back to create/update a saved BaseDocument (has .id) collapse to .id (ObjectId) an unsaved BaseDocument (no .id) kept as a model \u2192 embedded dict below a dict not BaseDocument \u2192 left as-is \u2192 embedded non-BaseDocument, non-dict value AttributeError raised by engine.depopulate no engine / no rules passthrough \u2014 _depopulate just does model_dump, whatever shape you gave is stored"},{"location":"03_use_cases/15_populating_repository_wiring/#read-path-get_by_id-step-by-step","title":"\ud83d\ude80 Read Path \u2014 get_by_id() Step by Step","text":"
get_by_id inherits CRUD and only changes data_to_model (repository.py:247):
_populate(data) walks each rule field in the raw dict:
ref = data.get(\"profile\") \u2192 ObjectId. If ref is None \u2192 field untouched (stays absent).
Look up the repo: repo = engine._repos.get(\"profiles\"). If it's not registered, doc = None.
doc = await repo.get_by_id(ref) \u2014 a single read on the referenced collection.
If doc is None \u2192 data[\"profile\"] = None (missing refs resolve to None, never raise).
If the rule has nested_rules \u2192 engine.populate(doc, nested_rules, depth=1) deepens the result.
data[\"profile\"] = doc \u2192 self.model(**data) builds the User with a real Profile.
Lists behave the same per item; a ref that is a dict (an embedded document) raises ValueError(\"...contains embedded dict(s) \u2014 run repair script\").
Engine is optional. Set population_engine=None and populate_rules=[], and PopulatingRepository is just a BaseRepository \u2014 FK fields come back as raw ObjectId, and your model field type must agree (that is the whole point of the guard in the WRITE path: a Profile | None-typed field populated with a raw ObjectId is a broken round-trip waiting to happen).
"},{"location":"03_use_cases/15_populating_repository_wiring/#how-its-wired","title":"\ud83d\udd0c How It's Wired","text":"Python
from mongo_ops import BaseDocument, PopulatingRepository\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\n\nclass Profile(BaseDocument):\n avatar_url: str = \"\"\n\nclass User(BaseDocument):\n username: str = \"\"\n profile: Profile | None = None # ObjectId in DB, Profile in memory\n\nengine = PopulationEngine({}) # repositories live here, keyed by collection name\n\nprofile_rule = PopulateRule(field_name=\"profile\", collection_name=\"profiles\")\n\nclass UserRepository(PopulatingRepository[User]):\n def __init__(self):\n super().__init__(\"users\", User, population_engine=engine, populate_rules=[profile_rule])\n\ndef wire() -> None:\n \"\"\"Call AFTER connect() \u2014 repositories need a live database.\"\"\"\n engine.register_repo(\"profiles\", PopulatingRepository[Profile](\"profiles\", Profile))\n engine.register_repo(\"users\", UserRepository())\n
Wiring rules:
engine._repos is keyed by collection_name as written in the rule \u2014 typo \u2192 silent None refs.
register_repo needs an already-constructed repository \u2192 call it inside the lifespan (after connect() / MongoConnectionManager.lifespan), not at module import.
The referenced repository only needs a get_by_id that returns a BaseDocument \u2014 it can be a plain BaseRepository, another PopulatingRepository, or even a cached repo (see use case 10). Population requires no extra DB index on the referenced collection's _id.
Swap at runtime: repo.set_population_engine(new_engine) and repo.set_populate_rules(new_rules) \u2014 the tests exercise both.
"},{"location":"03_use_cases/15_populating_repository_wiring/#intricacy-nested_rules-do-not-round-trip-through-depopulate","title":"\u26a0\ufe0f Intricacy \u2014 nested_rules Do NOT Round-Trip Through depopulate","text":"
nested_rules are designed for read-side deep population. If the same rules run through the write path (create / update), engine.depopulate behaves differently from plain refs \u2014 verified against the engine:
Rule shape What depopulate does to it Effect on the stored doc scalar FK, no nested_rules collapse to .id stored as ObjectId \u2713 list[ObjectId] FK, no nested_rules collapse each item to .id stored as list[ObjectId] \u2713 list FK with nested_rules items are kept as models stored as embedded dicts \u2717 \u2014 reading back raises \"contains embedded dict(s) \u2014 run repair script\" scalar FK with nested_rules the recursion runs, then the field is assigned None reference lost \u2717 \u2014 written as absent/null
In other words: a document carrying deep nested_rules (like UC 09's Author \u2192 books \u2192 publisher) cannot be created / updated as-is. If the graph must be written back, materialize references separately (save each Book to its collection first, then store list[ObjectId] without nested_rules), and keep nested_rules only on read rules you never hand back to depopulate.
Never pass a raw ObjectId-holding model to create/update: the \"was populate skipped?\" ValueError is the guard. Read through the repo so the read path can populate first.
Unsaved references are a write-once trap: they embed as dicts, and re-reading raises the \"run repair script\" ValueError. Save referenced docs to their collection first, then reference their id.
filter / projection on PopulateRule are declared but not applied by the engine \u2014 don't rely on them.
Reads cost 1 query per reference (no $lookup yet); batch-heavy endpoints should add caching (next use case).
08 \u2013 Population \u00b7 09 \u2013 Advanced population \u00b7 16 \u2013 Cached repository intricacies \u00b7 Components
"},{"location":"03_use_cases/16_cached_repository_intricacies/","title":"Use Case 16: Inside the Cached Repository \u2014 What's Actually Stored & Returned","text":"
Scenario: You want the exact contract of CachedBaseRepository \u2014 what goes into the cache, in what shape, and why a cached read occasionally looks \"wrong\" for populated models \u2014 before wiring it into a service.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#the-cache-contract","title":"\ud83d\udce6 The Cache Contract","text":"Aspect Value Key \"{key_prefix}{id}\" \u2014 default prefix \"{collection_name}:\" Value json.dumps(model_dump(by_alias=True), default=str) \u2014 a byte string of JSON get_by_id hit decode_value(cached) \u2192 self.model(**data) \u2014 no DB hit get_by_id miss DB read, then the raw doc is cached (model_dump(by_alias=True)) create creates in DB, then caches the result (model_dump) update DB update, then set new value or delete the key if the doc vanished delete DB delete, then removes the key TTL config.default_ttl (default 300 s); in-memory eviction is LRU + expiry
So the cache stores JSON snapshots of whole documents \u2014 it is a read-through cache keyed by document id, not a query cache.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#intricacy-1-objectids-become-hex-strings","title":"\u26a0\ufe0f Intricacy #1 \u2014 ObjectIds Become Hex Strings","text":"
json.dumps(..., default=str) stringifies every non-JSON value \u2014 most importantly an ObjectId in a FK field:
Python
# model in memory: User(id=..., profile=ObjectId(\"507f1f77bcf86cd799439011\"))\n# cached bytes: b'{\"_id\":\"507f1f77bcf86cd799439011\",\"profile\":\"507f1f77bcf86cd799439011\", ...}'\n
On a cache hit, self.model(**data) must therefore accept a hex string where the raw doc held an ObjectId. This is fine for: - id (typed PyObjectId, which accepts both str and ObjectId), and - created_at / updated_at (ISO strings coerce to datetime).
It is not fine for a field typed as a model.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#intricacy-2-model-typed-fk-fields-fail-on-a-cache-hit","title":"\u26a0\ufe0f Intricacy #2 \u2014 Model-Typed FK Fields Fail on a Cache Hit","text":"
If your document has a populated field, e.g.:
Python
class User(BaseDocument):\n profile: Profile | None = None # populate-ruled\n
then the base-class hit path self.model(**data) receives profile=\"507f\u2026\" and Pydantic raises a ValidationError \u2014 a string cannot coerce into a Profile. The miss path fails the same way: BaseRepository.get_by_id materializes the model from the raw doc whose profile is an ObjectId \u2014 also a ValidationError. So a plain CachedBaseRepository cannot materialize a model-typed FK field at all, hit or miss \u2014 the problem is the base classes build a typed model straight from raw docs.
The tension is structural: PopulatingRepository types the field as the model; CachedBaseRepository stores the raw (depopulated) shape. You cannot have a single typed model serve both at once. The two resolutions:
Compose \u2014 keep profile: Profile | None and cache raw, populating on read. Exact recipe in use case 10: its miss path fetches the raw doc (bypassing the base model-build) and both paths normalize str \u2192 ObjectId before populate, because the JSON round-trip hands you strings.
Type it as an id \u2014 profile: PyObjectId | None and no populate rules; then cached hits validate cleanly, but you've given up population entirely.
Do not cache a populated model object through the base class: create caches result.model_dump, so the first cache write stores the embedded-dict shape while later get_by_id misses would repopulate \u2014 inconsistent shapes for the same key, and update overwrites with yet another. Pick one canonical raw shape and stick to it.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#example-inspecting-what-gets-stored","title":"\ud83d\ude80 Example \u2014 Inspecting What Gets Stored","text":"
One shared backend instance. The repository needs it (cache_backend=cache) and the registry needs it (ModelRegistry.set_cache_backend(cache)) so initialize_cache() / shutdown_cache() manage the same object. Shutdown cancels the in-memory TTL cleanup task \u2014 forgetting it leaks an asyncio.Task at app exit.
initialize_cache() starts the backend; initialize_all() creates indexes. Both come after connect().
warm_cache([ids]) skips keys that already exist, fetches the rest from the DB, and returns how many it wrote \u2014 safe to call repeatedly.
invalidate_cache(id) deletes one key; clear_pattern(\"products:*\") wipes a collection.
"},{"location":"03_use_cases/16_cached_repository_intricacies/#redis-differences","title":"\ud83d\udd04 Redis Differences","text":"Behaviour In-memory Redis (redis.asyncio) Key prefixing prefix baked into the stored key _full_key() applied on every op TTL heap-based, lazy eviction + cleanup task Native SETEX Invalidation broadcast n/a PUBLISH on mongo_ops:cache:invalidate on delete clear_pattern prefix match on stored keys SCAN MATCH in batches"},{"location":"03_use_cases/16_cached_repository_intricacies/#choosing-the-right-layer","title":"\ud83c\udfaf Choosing the Right Layer","text":"Need Use Scalar docs, no refs \u2014 cache-first reads CachedBaseRepository (UC 07, UC 16) Refs resolved on read, no caching PopulatingRepository (UC 08, UC 15) Refs and cache-first reads composed subclass (UC 10) Bulk warm on startup / cache-then-database failover warm_cache + CacheStats"},{"location":"03_use_cases/16_cached_repository_intricacies/#related","title":"Related","text":"
"},{"location":"03_use_cases/17_populate_depopulate_examples/","title":"17. Populate & Depopulate \u2014 Worked Examples","text":"
Scenario: You want to see what populate/depopulate actually do to your documents \u2014 field by field, at every layer (Python model \u2192 depopulated \u2192 stored in Mongo \u2192 raw read back \u2192 populated).
These examples all ran against the real mongo_ops engine (mongo_ops/populate/engine.py). Timestamps and ObjectIds are abbreviated with \u2026 for readability.
"},{"location":"03_use_cases/17_populate_depopulate_examples/#models-rules-used-below","title":"Models & rules used below","text":"Python
from bson import ObjectId\nfrom mongo_ops import BaseDocument, PopulatingRepository\nfrom mongo_ops.populate import PopulateRule, PopulationEngine\n\nclass Profile(BaseDocument):\n avatar_url: str = \"\"\n bio: str = \"\"\n\nclass Publisher(BaseDocument):\n name: str = \"\"\n\nclass Book(BaseDocument):\n title: str = \"\"\n publisher: Publisher | None = None # nested ref, reached via nested_rules\n\nclass User(BaseDocument):\n name: str = \"\"\n profile: Profile | None = None # scalar FK \u2014 DB stores an ObjectId\n books: list[Book] | None = None # list FK \u2014 DB stores list[ObjectId]\n\nengine = PopulationEngine({}) # repos keyed by collection name, see UC 15\n\n# A rule WITHOUT nested_rules \u2014 the write path is safe.\nprofile_rule = PopulateRule(field_name=\"profile\", collection_name=\"profiles\")\n\n# A rule WITH nested_rules \u2014 designed for READ-side deep population.\nbooks_rule = PopulateRule(\n field_name=\"books\",\n collection_name=\"books\",\n nested_rules=[PopulateRule(field_name=\"publisher\", collection_name=\"publishers\")],\n)\n
"},{"location":"03_use_cases/17_populate_depopulate_examples/#1-write-path-depopulate-turns-models-into-references","title":"1. Write path \u2014 depopulate() turns models into references","text":"
engine.depopulate(document, rules) walks each rule field and mutates the model in place before you insert_one / find_one_and_update it. What you get on the other side:
"},{"location":"03_use_cases/17_populate_depopulate_examples/#1a-scalar-fk-no-nested_rules-stored-as-objectid","title":"1a. Scalar FK, no nested_rules \u2192 stored as ObjectId","text":"Python
async def demo():\n profile = Profile(id=ObjectId(\"656\u202604\"), avatar_url=\"a.png\", bio=\"hi\")\n user = User(id=ObjectId(\"656\u202605\"), name=\"Ada\", profile=profile)\n\n await engine.depopulate(user, [profile_rule])\n\n return user.profile # ObjectId('656\u202604') \u2014 was a Profile model\n
What Mongo receives (via model_dump, minus _id handling):
"},{"location":"03_use_cases/17_populate_depopulate_examples/#1b-scalar-fk-with-nested_rules-reference-lost-do-not-do-this","title":"1b. Scalar FK with nested_rules \u2192 reference lost (do not do this)","text":"Python
profile_rule_deep = PopulateRule( # same field, but with nested_rules\n field_name=\"profile\",\n collection_name=\"profiles\",\n nested_rules=[PopulateRule(field_name=\"bio\", collection_name=\"profiles\")], # any child field\n)\n\nasync def demo():\n user = User(id=ObjectId(\"656\u202605\"), name=\"Ada\",\n profile=Profile(id=ObjectId(\"656\u202604\"), avatar_url=\"a.png\"))\n await engine.depopulate(user, [profile_rule_deep])\n return user.profile # None \u2014 reference silently dropped!\n
Because depopulate recurses into the child but then assigns the parent field None (see engine.py \u2014 depopulated_value is never set on that branch). On save you lose the reference entirely. There is no warning \u2014 the field is just gone.
"},{"location":"03_use_cases/17_populate_depopulate_examples/#1c-list-fk-no-nested_rules-stored-as-listobjectid-the-safe-pattern","title":"1c. List FK, no nested_rules \u2192 stored as list[ObjectId] (the safe pattern)","text":"Python
"},{"location":"03_use_cases/17_populate_depopulate_examples/#1d-list-fk-with-nested_rules-stored-as-embedded-documents-do-not-do-this","title":"1d. List FK with nested_rules \u2192 stored as embedded documents (do not do this)","text":"Python
books stays a list of Book models \u2014 the nested publisher collapsed to ObjectId, but the Book itself was left embedded. Serialized to Mongo you get a nested sub-document:
\u2026and reading that back through population raises ValueError(\u2026contains embedded dict(s) \u2014 run repair script\u2026), because the engine expects books to hold references, not embedded docs.
Why UC 09's Author \u2192 books \u2192 publisher graph can't be written as-is: a rule with nested_rules breaks the write path (1b logs silently, 1d stores embedded). The safe round-trip: save referenced documents first, store list[ObjectId]/ObjectId with a non-nested rule for reads, and keep nested_rules only on rules you never hand back to depopulate.
"},{"location":"03_use_cases/17_populate_depopulate_examples/#2-read-path-populate-turns-references-back-into-models","title":"2. Read path \u2014 populate() turns references back into models","text":"
PopulatingRepository.data_to_model(raw_dict) resolves references before building the model. It first runs _populate(dict) \u2014 replacing the raw ObjectId references with full models via the registered repos \u2014 and only then constructs self.model(**data), so every FK field is a real model by construction. The two data_to_model examples below (2a/2b) show the underlying resolution _populate performs per rule. (repo below is a PopulatingRepository wired per UC 15 \u2014 books and publishers repos registered in its PopulationEngine.)
"},{"location":"03_use_cases/17_populate_depopulate_examples/#2a-scalar-fk-objectid-profile-model","title":"2a. Scalar FK \u2014 ObjectId \u2192 Profile model","text":"Python
# what came back from Mongo (raw dict, profile is an ObjectId reference):\nraw = {\"_id\": ObjectId(\"656\u202605\"), \"name\": \"Ada\", \"profile\": ObjectId(\"656\u202604\")}\n\nasync def demo():\n user = await repo.data_to_model(raw) # _populate() \u2192 model(**data)\n return user.profile # Profile(id=ObjectId('656\u202604'), avatar_url='a.png')\n
"},{"location":"03_use_cases/17_populate_depopulate_examples/#2b-list-fk-nested_rules-two-levels-deep","title":"2b. List FK + nested_rules \u2014 two levels deep","text":"Python
books_rule \u2192 field \"books\", collection \"books\"\n nested_rules[0].publisher rule \u2192 field \"publisher\", collection \"publishers\"\n1. for ObjectId('656\u202606') \u2192 repo(\"books\").get_by_id(...) \u2192 Book(\"MongoDB in Action\")\n2. book has nested rule \u2192 repo(\"publishers\").get_by_id(...) \u2192 Publisher(\"O'Reilly\")\n3. author.books[0] \u2192 Book(publisher=Publisher(\"O'Reilly\"))\n
Per-level cost: 1 query per reference (no $lookup). A hub with 20 hooks and a publisher nested rule is 40 queries \u2014 that is why the next section exists.
"},{"location":"03_use_cases/17_populate_depopulate_examples/#3-cache-round-trip-what-actually-goes-into-the-cache","title":"3. Cache round trip \u2014 what actually goes into the cache","text":"
CachedBaseRepository stores model_dump(by_alias=True) JSON-encoded with default=str (the encode_value helper in mongo_ops/cache/in_memory.py). Crucially it snapshots the raw, just-out-of-Mongo shape \u2014 FK fields still hold ObjectIds \u2014 so the cache holds references, exactly like the DB row:
Layer profile field looks like Mongo doc (raw) \u201cprofile\u201d: ObjectId(\"656\u202604\")model_dump(by_alias=True)\"profile\": ObjectId(\"656\u202604\")encode_value(...)\"...\\\"profile\\\": \\\"656\u202604\\\"...\" \u2014 hex string decode_value(...)\"profile\": \"656\u202604\" \u2014 still a hex string
Verbatim (timestamps abbreviated):
Python
from mongo_ops.cache.in_memory import decode_value, encode_value\n\nsnapshot = {\"id\": \"656\u202605\", \"profile\": ObjectId(\"656\u202604\")} # what a cached raw doc looks like\n\ncached = encode_value(snapshot) # json.dumps(snapshot, default=str).encode()\ndecoded = decode_value(cached)\n\ndecoded[\"profile\"] # '656\u202604' \u2014 hex STRING, not ObjectId, not Profile\n
Typed-FK caveat: the round trip above only works when the FK field's model type accepts the raw value (str/ObjectId typed). A field typed Profile | None cannot even be materialized from the raw Mongo doc \u2014 both hit and miss raise ValidationError (UC 16's intricacy). That is exactly why UC 10's composed repo overrides get_by_id/create/update to carry raw data through the cache and resolve refs with _populate.
Two consequences (the \"intricacies\", detailed in UC 16):
On a cache hit, decode_value gives you hex-string references. A field typed Profile | None receiving a hex string raises Pydantic ValidationError \u2014 so a plain CachedBaseRepository cannot serve model-typed FK fields, hit or miss. The composed pattern in UC 10 re-resolves them in its get_by_id: data = decode_value(await cache.get(key)), then populated = await self._populate(data), then self.model(**populated).
On a miss, BaseRepository.get_by_id reads the raw Mongo doc whose profile is an ObjectId and immediately raises the same ValidationError \u2014 which is why UC 10 overrides get_by_id to fetch raw, cache raw, then populate. (Real ObjectIds are covered by UC 16's table.)
"},{"location":"03_use_cases/17_populate_depopulate_examples/#4-when-to-use-which","title":"4. When to use which","text":"You need\u2026 Use Scalar ObjectId store, model-struct reads PopulatingRepository (UC 08) Deep nested reads (2+ levels) PopulatingRepository + nested_rules (UC 09) \u2014 read side only Same shape, but cache-first reads UC 10 composed repo (engine + cache + _populate) Scalar docs, no refs, hot get_by_idCachedBaseRepository (UC 07, UC 16) Writing a reference graph back depopulate without nested_rules \u2014 save children first, then point at their ids"},{"location":"03_use_cases/17_populate_depopulate_examples/#related","title":"Related","text":"
08 \u2013 Population \u00b7 09 \u2013 Advanced population \u00b7 10 \u2013 Cache + Population \u00b7 15 \u2013 PopulatingRepository wiring \u00b7 16 \u2013 CachedRepository intricacies