Skip to content

Core Components

A validated reference to the public API. Signatures match the code in mongo_ops/ exactly. See the use cases for runnable recipes.

1. MongoConnectionManager

Singleton-style owner of the Motor client and active database.

Method Signature Behavior
connect async (uri: str, db_name: str, **kwargs) -> AsyncIOMotorDatabase Creates AsyncIOMotorClient(uri, **kwargs) and selects the database. Idempotent.
disconnect async () -> None Closes the client and clears state.
get_database () -> AsyncIOMotorDatabase Returns the active database; raises RuntimeError("Database not connected. Call connect() first.").
get_client () -> AsyncIOMotorClient Returns the active client; raises RuntimeError("Client not connected. Call connect() first.").
lifespan async ctx manager (uri, db_name, **kwargs) connect() on entry, yields the database, disconnect() on exit. Designed for FastAPI lifespan.

2. BaseDocument & PyObjectId

BaseDocument(BaseModel) — inherit for every Mongo entity.

  • id: Optional[PyObjectId] — aliased to _id, serialized to str.
  • created_at: datetime — defaults to datetime.utcnow().
  • updated_at: datetime — defaults to datetime.utcnow().
  • Config.populate_by_name = True, arbitrary_types_allowed = True, json_encoders = {ObjectId: str}.

PyObjectId(ObjectId) — Pydantic v2-compatible ObjectId that accepts str or ObjectId and validates with ObjectId.is_valid.

3. CRUDMixin[T]

Generic CRUD over a Motor collection. The building block of all repositories.

Method Signature Notes
create async (data: T) -> T Dumps model (excludes id, None), stamps created_at/updated_at, inserts, returns model with assigned _id.
get_by_id async (id: str \| ObjectId) -> Optional[T] str is accepted and converted to ObjectId.
get_many async (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.
update async (id, data: dict[str, Any]) -> Optional[T] $set + refreshed updated_at via find_one_and_update.
patch async (id, data: dict[str, Any]) -> Optional[T] Same as update but intended for REST PATCH semantics.
delete async (id) -> bool True if a document was deleted.
count async (filter: dict \| None = None) -> int count_documents.

4. BaseRepository[T]

BaseRepository(collection_name: str, model: type[T]) — resolves the collection from MongoConnectionManager.get_database()[collection_name]. Requires an active connection at construction time. Provides everything in CRUDMixin plus collection_name.

5. PopulatingRepository[T]

PopulatingRepository(collection_name, model, population_engine: PopulationEngine | None = None, populate_rules: list[PopulateRule] | None = None).

  • set_population_engine(engine) / set_populate_rules(rules) — swap engine/rules at runtime.
  • data_to_model hooks _populate — get_by_id/get_many return fully populated models.
  • _depopulate(document) — 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 — 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.

6. TransactionManager

Method Signature Behavior
start_session async ctx manager (**kwargs) -> AsyncIOMotorClientSession Yields a session with an active transaction. Pass session= to every collection call inside.
execute_transaction async (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.

7. ModelRegistry

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 — single tuples, compound lists, or dicts with keys/options.
initialize_all async (db: AsyncIOMotorDatabase \| None = None) -> None create_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_cache async () -> None Starts the backend (background TTL cleanup) — raises RuntimeError if no backend registered.
shutdown_cache async () -> None Stops the backend cleanly and clears it.
get_cache_backend () -> Optional[CacheBackend] Current backend, if any.

8. Cache Layer

8.1 CacheBackend (abstract)

Async interface: get(key) -> Optional[bytes], set(key, value: bytes, ttl | None), delete(key), exists(key), clear_pattern(pattern), get_stats() -> CacheStats, initialize(), shutdown(). Values are bytes (JSON-encoded).

8.2 CacheStats

Dataclass: hits, misses, sets, deletes, current_size, max_size.

8.3 CacheConfig

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.

8.4 InMemoryCacheBackend

InMemoryCacheBackend(max_entries=10000, default_ttl=300, cleanup_interval=60) — LRU OrderedDict + TTL heap; initialize() spawns the periodic cleanup task (TTL 0 expires immediately).

8.5 RedisCacheBackend

RedisCacheBackend(redis_client, key_prefix="", default_ttl=300) — setex storage, SCAN-based clear_pattern, and publish_invalidate(key) for cross-service invalidation on delete via the mongo_ops:cache:invalidate channel.

8.6 CachedBaseRepository[T]

CachedBaseRepository(collection_name, model, cache_backend: CacheBackend, config: CacheConfig | None = None).

  • Cache keys are "{key_prefix}{id}" (prefix defaults to "{collection_name}:").
  • get_by_id — cache-first; cache miss reads DB and stores model_dump(by_alias=True) (JSON-encoded) for default_ttl. Honors config.enabled=False (bypass).
  • create — inserts then caches the result.
  • update/delete — refresh or remove the cache entry.
  • warm_cache(ids) -> int — prefetch a list of IDs, returns count warmed.
  • invalidate_cache(id) — manual eviction.

9. Population Layer

9.1 PopulateRule

Dataclass:

Python
@dataclass
class PopulateRule:
    field_name: str          # field holding the ObjectId / list[ObjectId]; replaced in-place with the resolved doc(s)
    collection_name: str     # collection the references point at
    nested_rules: list[PopulateRule] | None = None
    max_depth: int = 1
    filter: dict | None = None      # DECLARED but NOT yet applied by the engine
    projection: dict | None = None  # DECLARED but NOT yet applied by the engine

⚠️ filter and projection are accepted but currently ignored by PopulationEngine — do not rely on them yet.

9.2 PopulationEngine

PopulationEngine(repos: dict[str, Any], global_max_depth: int = 10) — repos maps collection_name → repository.

  • register_repo(collection_name, repo) — add repositories at runtime.
  • populate(document, rules, depth=0) — resolves refs recursively, replacing field_name in place; raises CircularReferenceError(collection, doc_id, path) when a (Class, id) pair is revisited.
  • depopulate(document, rules) — collapse populated docs back to IDs (for storage).
  • global_max_depth caps recursion; per-rule max_depth bounds a rule's descent.

9.3 CircularReferenceError(ValueError)

Holds collection, doc_id, and the visited path for debugging cycle messages.