Repository

mongo_ops.repository

Summary

Repository patterns and CRUD mixins for MongoDB.

Classes

BaseRepository

BaseRepository(collection_name: str, model: type[T])

Bases: CRUDMixin[T], Generic[T]

Base repository class combining CRUD operations and collection management.

This class simplifies repository creation by automatically obtaining the database connection and collection instance.

Attributes:

Name	Type	Description
collection_name	str	The name of the collection managed by this repository.

Initialize the repository.

Parameters:

Name	Type	Description	Default
collection_name	str	The name of the MongoDB collection.	required
model	type[T]	The Pydantic model class.	required

Functions

count async

count(filter: dict[str, Any] | None = None) -> int

Count documents matching a filter.

Parameters:

Name	Type	Description	Default
filter	Optional[Dict[str, Any]]	MongoDB filter dictionary.	None

Returns:

Name	Type	Description
int	int	The number of matching documents.

create async

create(data: T) -> T

Create a new document in the collection.

Parameters:

Name	Type	Description	Default
data	T	The Pydantic model instance to insert.	required

Returns:

Name	Type	Description
T	T	The created Pydantic model instance, including the assigned ID.

delete async

delete(id: str | ObjectId) -> bool

Delete a document by its ID.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required

Returns:

Name	Type	Description
bool	bool	True if a document was deleted, False otherwise.

get_by_id async

get_by_id(id: str | ObjectId) -> T | None

Retrieve a document by its ID.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required

Returns:

Type	Description
T | None	Optional[T]: The Pydantic model instance if found, else None.

get_many async

get_many(
    filter: dict[str, Any] | None = None,
    skip: int = 0,
    limit: int = 100,
    sort: list[tuple] | None = None,
) -> list[T]

Retrieve multiple documents with filtering, pagination, and sorting.

Parameters:

Name	Type	Description	Default
filter	Optional[Dict[str, Any]]	MongoDB filter dictionary (e.g., {"is_active": True}).	None
skip	int	Number of documents to skip for pagination.	0
limit	int	Maximum number of documents to return (default 100).	100
sort	Optional[List[tuple]]	List of sort specifications [(field, direction), ...]. E.g., [("created_at", -1)] for descending.	None

Returns:

Type	Description
list[T]	List[T]: A list of Pydantic model instances.

Example

users = await repo.get_many(
    filter={"role": "admin"},
    limit=10,
    sort=[("username", 1)]
)

patch async

patch(id: str | ObjectId, data: dict[str, Any]) -> T | None

Partially update a document using $set (REST PATCH semantics).

Unlike update(), patch() takes a partial dict and applies only those fields. PopulatingRepository overrides this to prevent patching FK fields.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required
data	Dict[str, Any]	A partial dictionary of fields and values to update.	required

Returns:

Type	Description
T | None	Optional[T]: The updated Pydantic model instance if found, else None.

update async

update(
    id: str | ObjectId, data: dict[str, Any]
) -> T | None

Update a document by its ID using the $set operator.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required
data	Dict[str, Any]	A dictionary of fields and values to update.	required

Returns:

Type	Description
T | None	Optional[T]: The updated Pydantic model instance if found, else None.

Example

updated_user = await repo.update(user_id, {"email": "new@example.com"})

CRUDMixin

CRUDMixin(
    collection: AsyncIOMotorCollection, model: type[T]
)

Bases: Generic[T]

Generic CRUD operations mixin for MongoDB collections.

This mixin provides standard Create, Read, Update, and Delete operations that work with Pydantic models.

Attributes:

Name	Type	Description
collection	AsyncIOMotorCollection	The Motor collection instance.
model	type[T]	The Pydantic model class representing the document.

Initialize the CRUD mixin.

Parameters:

Name	Type	Description	Default
collection	AsyncIOMotorCollection	The Motor collection to operate on.	required
model	type[T]	The Pydantic model class (subclass of BaseDocument).	required

Functions

count async

count(filter: dict[str, Any] | None = None) -> int

Count documents matching a filter.

Parameters:

Name	Type	Description	Default
filter	Optional[Dict[str, Any]]	MongoDB filter dictionary.	None

Returns:

Name	Type	Description
int	int	The number of matching documents.

create async

create(data: T) -> T

Create a new document in the collection.

Parameters:

Name	Type	Description	Default
data	T	The Pydantic model instance to insert.	required

Returns:

Name	Type	Description
T	T	The created Pydantic model instance, including the assigned ID.

delete async

delete(id: str | ObjectId) -> bool

Delete a document by its ID.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required

Returns:

Name	Type	Description
bool	bool	True if a document was deleted, False otherwise.

get_by_id async

get_by_id(id: str | ObjectId) -> T | None

Retrieve a document by its ID.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required

Returns:

Type	Description
T | None	Optional[T]: The Pydantic model instance if found, else None.

get_many async

get_many(
    filter: dict[str, Any] | None = None,
    skip: int = 0,
    limit: int = 100,
    sort: list[tuple] | None = None,
) -> list[T]

Retrieve multiple documents with filtering, pagination, and sorting.

Parameters:

Name	Type	Description	Default
filter	Optional[Dict[str, Any]]	MongoDB filter dictionary (e.g., {"is_active": True}).	None
skip	int	Number of documents to skip for pagination.	0
limit	int	Maximum number of documents to return (default 100).	100
sort	Optional[List[tuple]]	List of sort specifications [(field, direction), ...]. E.g., [("created_at", -1)] for descending.	None

Returns:

Type	Description
list[T]	List[T]: A list of Pydantic model instances.

Example

users = await repo.get_many(
    filter={"role": "admin"},
    limit=10,
    sort=[("username", 1)]
)

patch async

patch(id: str | ObjectId, data: dict[str, Any]) -> T | None

Partially update a document using $set (REST PATCH semantics).

Unlike update(), patch() takes a partial dict and applies only those fields. PopulatingRepository overrides this to prevent patching FK fields.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required
data	Dict[str, Any]	A partial dictionary of fields and values to update.	required

Returns:

Type	Description
T | None	Optional[T]: The updated Pydantic model instance if found, else None.

update async

update(
    id: str | ObjectId, data: dict[str, Any]
) -> T | None

Update a document by its ID using the $set operator.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required
data	Dict[str, Any]	A dictionary of fields and values to update.	required

Returns:

Type	Description
T | None	Optional[T]: The updated Pydantic model instance if found, else None.

Example

updated_user = await repo.update(user_id, {"email": "new@example.com"})

PopulatingRepository

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

Bases: BaseRepository[T], Generic[T]

Repository that auto-populates and depopulates FK references.

On read, ObjectId FK fields are resolved to model instances according to the configured PopulateRules. On write, populated model references are collapsed back to ObjectIds before hitting MongoDB.

Notes

Guarantees:

- Populate rules apply on both read (data_to_model) and write
  (create/update) paths.
- Patching FK fields via patch() is rejected.
- Class attributes `population_engine` and `_populate_rules` must
  be set (see set_population_engine/set_populate_rules) for any
  population to occur.

Initialize the repository.

Parameters:

Name	Type	Description	Default
collection_name	str	Name of the MongoDB collection.	required
model	type[T]	The Pydantic model class.	required
population_engine	Optional[PopulationEngine]	Engine used to resolve references. Defaults to None.	None
populate_rules	Optional[list[PopulateRule]]	Rules describing FK resolution. Defaults to None (no rules).	None

Functions

count async

count(filter: dict[str, Any] | None = None) -> int

Count documents matching a filter.

Parameters:

Name	Type	Description	Default
filter	Optional[Dict[str, Any]]	MongoDB filter dictionary.	None

Returns:

Name	Type	Description
int	int	The number of matching documents.

create async

create(data: T) -> T

Depopulate, insert, and re-populate a new document.

Parameters:

Name	Type	Description	Default
data	T	The model instance to insert (FK fields may hold models).	required

Returns:

Name	Type	Description
T	T	The created model instance, including its ID.

data_to_model async

data_to_model(data: dict) -> T

Convert a raw dict to a model, resolving FK references first.

Parameters:

Name	Type	Description	Default
data	dict	Raw document dictionary.	required

Returns:

Name	Type	Description
T	T	The populated model instance.

Raises:

Type	Description
ValueError	If a FK field holds an embedded dict instead of an ObjectId.

delete async

delete(id: str | ObjectId) -> bool

Delete a document by its ID.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required

Returns:

Name	Type	Description
bool	bool	True if a document was deleted, False otherwise.

get_by_id async

get_by_id(id: str | ObjectId) -> T | None

Retrieve a document by its ID.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID (string or ObjectId).	required

Returns:

Type	Description
T | None	Optional[T]: The Pydantic model instance if found, else None.

get_many async

get_many(
    filter: dict[str, Any] | None = None,
    skip: int = 0,
    limit: int = 100,
    sort: list[tuple] | None = None,
) -> list[T]

Retrieve multiple documents with filtering, pagination, and sorting.

Parameters:

Name	Type	Description	Default
filter	Optional[Dict[str, Any]]	MongoDB filter dictionary (e.g., {"is_active": True}).	None
skip	int	Number of documents to skip for pagination.	0
limit	int	Maximum number of documents to return (default 100).	100
sort	Optional[List[tuple]]	List of sort specifications [(field, direction), ...]. E.g., [("created_at", -1)] for descending.	None

Returns:

Type	Description
list[T]	List[T]: A list of Pydantic model instances.

Example

users = await repo.get_many(
    filter={"role": "admin"},
    limit=10,
    sort=[("username", 1)]
)

patch async

patch(id: str | ObjectId, data: dict[str, Any]) -> T | None

Partially update a document, rejecting FK field changes.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID.	required
data	dict[str, Any]	Partial dictionary of fields to update.	required

Returns:

Type	Description
T | None	Optional[T]: The updated model instance, or None when not found.

Raises:

Type	Description
ValueError	If any FK field is present in the patch payload.

set_populate_rules

set_populate_rules(rules: list[PopulateRule]) -> None

Set the FK resolution rules.

Parameters:

Name	Type	Description	Default
rules	list[PopulateRule]	Rules describing which fields resolve and how deep.	required

set_population_engine

set_population_engine(engine: PopulationEngine) -> None

Attach (or replace) the population engine.

Parameters:

Name	Type	Description	Default
engine	PopulationEngine	Engine used to resolve references.	required

update async

update(id: str | ObjectId, data: T) -> T | None

Depopulate and update a document by ID.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID.	required
data	T	The model instance holding the updated fields.	required

Returns:

Type	Description
T | None	Optional[T]: The updated model instance, or None when not found.