Skip to content

Use Case 10: Caching + Population (Read-Through, Populated on Read)

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.

📦 What's New?

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.

⚠️ Known limitation: there is no built-in CachedPopulatingRepository in the library. This use case documents the composition. We cache the raw (depopulated) document — references stay ObjectIds until read time, so Profile changes are reflected on the next fetch (within TTL) and update/delete invalidation remains correct.

🚀 Example

Python
from contextlib import asynccontextmanager
from datetime import datetime
from typing import Any, Union

from bson import ObjectId
from fastapi import FastAPI
from mongo_ops import BaseDocument, CachedBaseRepository, ModelRegistry, MongoConnectionManager
from mongo_ops.cache import CacheConfig, InMemoryCacheBackend
from mongo_ops.cache.in_memory import decode_value, encode_value
from mongo_ops.populate import PopulateRule, PopulationEngine


# 1. Models — `profile` holds an ObjectId in DB, a Profile in memory.
class Profile(BaseDocument):
    avatar_url: str = ""
    bio: str = ""


class User(BaseDocument):
    username: str = ""
    email: str = ""
    profile: Profile | None = None


# 2. Engine + rule.
engine = PopulationEngine({})
profile_rule = PopulateRule(field_name="profile", collection_name="profiles")


# 3. Composed repository.
class CachedUserRepository(CachedBaseRepository[User]):
    def __init__(
        self,
        cache_backend: InMemoryCacheBackend,
        population_engine: PopulationEngine,
        populate_rules: list[PopulateRule],
        config: CacheConfig | None = None,
    ):
        super().__init__("users", User, cache_backend, config)
        self.population_engine = population_engine
        self._populate_rules = populate_rules

    async def _populate(self, data: dict[str, Any]) -> dict[str, Any]:
        """Dict-level resolution (mirrors PopulatingRepository._populate)."""
        for rule in self._populate_rules:
            ref = data.get(rule.field_name)
            if ref is None:
                continue
            if isinstance(ref, str):  # JSON round-trip turns ObjectId into hex str
                ref = ObjectId(ref)
            if isinstance(ref, list):
                resolved = []
                for item in ref:
                    item = ObjectId(item) if isinstance(item, str) else item
                    repo = self.population_engine._repos.get(rule.collection_name)
                    doc = await repo.get_by_id(item) if repo else None
                    resolved.append(doc)
                data[rule.field_name] = resolved
            elif isinstance(ref, ObjectId):
                repo = self.population_engine._repos.get(rule.collection_name)
                data[rule.field_name] = await repo.get_by_id(ref) if repo else None
        return data

    async def _depopulate(self, data: User) -> dict[str, Any]:
        """Model -> raw dict, FK fields collapsed to ObjectId (mirrors _depopulate)."""
        doc = data.model_dump(exclude={"id"}, exclude_none=True)
        for rule in self._populate_rules:
            value = getattr(data, rule.field_name, None)
            if isinstance(value, list):
                doc[rule.field_name] = [item.id for item in value]
            elif isinstance(value, BaseDocument):
                doc[rule.field_name] = value.id
        return doc

    async def create(self, data: User) -> User:
        doc = await self._depopulate(data)  # store FK refs as ObjectIds
        doc["created_at"] = datetime.utcnow()
        doc["updated_at"] = datetime.utcnow()
        result = await self.collection.insert_one(doc)
        doc["_id"] = result.inserted_id
        if self._cache_config.enabled:
            await self._cache.set(self._cache_key(doc["_id"]), encode_value(doc), self._cache_config.default_ttl)
        return self.model(**await self._populate(doc))

    async def update(self, id: Union[str, ObjectId], data: User) -> User | None:
        if isinstance(id, str):
            id = ObjectId(id)
        doc = await self._depopulate(data)  # update() takes a full model here
        doc["updated_at"] = datetime.utcnow()
        result = await self.collection.find_one_and_update(
            {"_id": id}, {"$set": doc}, return_document=True
        )
        key = self._cache_key(id)
        if result is None:
            if self._cache_config.enabled:
                await self._cache.delete(key)
            return None
        raw = dict(result)
        if self._cache_config.enabled:
            await self._cache.set(key, encode_value(raw), self._cache_config.default_ttl)
        return self.model(**await self._populate(raw))

    async def get_by_id(self, id: Union[str, ObjectId]) -> User | None:
        if not self._cache_config.enabled:
            # No cache: still a raw fetch + populate — the base get_by_id()
            # would build a typed model from the raw doc and fail on FK fields.
            raw = await self.collection.find_one({"_id": ObjectId(id) if isinstance(id, str) else id})
            return self.model(**await self._populate(raw)) if raw else None

        cached = await self._cache.get(self._cache_key(id))
        if cached is not None:
            # Cache hit: decode the RAW doc, then populate before returning.
            data = await self._populate(decode_value(cached))
            return self.model(**data)

        # Cache miss: one RAW DB read. The base get_by_id() would rebuild the
        # model from the raw doc, which fails for model-typed FK fields —
        # so fetch the raw dict, cache it, and populate before materializing.
        key = self._cache_key(id)
        raw = await self.collection.find_one({"_id": ObjectId(id) if isinstance(id, str) else id})
        if raw is None:
            return None
        await self._cache.set(key, encode_value(raw), self._cache_config.default_ttl)
        return self.model(**await self._populate(raw))


# 4. Wire-up — one backend for both the repo and the registry lifecycle.
cache = InMemoryCacheBackend(max_entries=20_000, default_ttl=600)
ModelRegistry.set_cache_backend(cache)


@asynccontextmanager
async def lifespan(_app: FastAPI):
    async with MongoConnectionManager.lifespan(
        uri="mongodb://localhost:27017", db_name="app_db"
    ):
        # Register the repositories the engine resolves refs against:
        # engine.register_repo("profiles", ProfileRepo())
        engine.register_repo("users", CachedUserRepository(cache, engine, [profile_rule]))
        await ModelRegistry.initialize_all()
        await ModelRegistry.initialize_cache()
        yield
        await ModelRegistry.shutdown_cache()


app = FastAPI(lifespan=lifespan)

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 — e.g. engine.register_repo("profiles", CachedBaseRepository[Profile]("profiles", Profile, cache)).

💡 Tips

  • 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 → ObjectId (shown above).
  • update/create above are model-based (update(id, User)), mirroring PopulatingRepository — 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 — 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.