Engine

mongo_ops.populate.engine

Summary

Recursive document population engine with cycle detection.

Classes

PopulationEngine

PopulationEngine(
    repos: dict[str, Any], global_max_depth: int = 10
)

Resolves FK references across collections with cycle detection.

The engine holds a registry of repositories keyed by collection name and walks documents according to PopulateRules, resolving ObjectId references into model instances. A visited-path set guards against circular graphs.

Notes

Guarantees:

- A reference that cannot be resolved becomes None (scalar) or a
  None entry (list) rather than raising.
- Depth is capped by both per-rule max_depth and a global
  global_max_depth.

Initialize the engine.

Parameters:

Name	Type	Description	Default
repos	dict[str, Any]	Mapping of collection name to repository, used to fetch referenced documents.	required
global_max_depth	int	Hard cap on overall population recursion depth. Defaults to 10.	10

Functions

depopulate async

depopulate(document: T, rules: list[PopulateRule]) -> T

Collapse populated model references back to their ObjectIds.

This is the inverse of populate(): model-valued FK fields are reduced to stored identifiers before the document is written to MongoDB.

Parameters:

Name	Type	Description	Default
document	T	The document to depopulate in place.	required
rules	list[PopulateRule]	Rules describing which fields to collapse.	required

Returns:

Name	Type	Description
T	T	The depopulated document.

Raises:

Type	Description
AttributeError	If a list entry is not a BaseDocument where expected.

populate async

populate(
    document: T,
    rules: list[PopulateRule],
    depth: int = 0,
    _visited: set[tuple[str, str]] | None = None,
    _path: list[str] | None = None,
) -> T

Resolve FK fields on a document according to the given rules.

Parameters:

Name	Type	Description	Default
document	T	The document to populate in place.	required
rules	list[PopulateRule]	Rules describing which fields to resolve and how deep.	required
depth	int	Current recursion depth. Defaults to 0.	0
_visited	Optional[set[tuple[str, str]]]	Internal set of (class, id) pairs on the active path.	None
_path	Optional[list[str]]	Internal path labels used for cycle reporting.	None

Returns:

Name	Type	Description
T	T	The populated document.

Raises:

Type	Description
CircularReferenceError	If a cycle is detected on the active path.

register_repo

register_repo(collection_name: str, repo: Any) -> None

Register or replace the repository for a collection.

Parameters:

Name	Type	Description	Default
collection_name	str	Collection the repository manages.	required
repo	Any	Repository exposing get_by_id() used to resolve references.	required