mongo_ops

mongo_ops

Summary

mongo-ops: A modular MongoDB operations layer for FastAPI microservices.

This package provide a standardized way to interact with MongoDB in async Python applications, particularly optimized for FastAPI. It includes:

• Connection Management: Async lifecycle management for Motor clients.
• Base Models: Pydantic v2 models for MongoDB documents.
• Repository Pattern: Generic CRUD operations and base repository classes.
• Caching: In-memory and Redis cache backends with ID-based caching.
• Population: Recursive document populating with cycle detection.
• Transactions: Helpers for multi-document ACID transactions.
• Registry: Centralized model, index, and cache lifecycle management.

Example

from mongo_ops import BaseDocument, BaseRepository, CachedBaseRepository
from mongo_ops.cache import InMemoryCacheBackend, CacheConfig

class User(BaseDocument):
    username: str

class UserRepository(BaseRepository[User]):
    def __init__(self):
        super().__init__("users", User)

# With caching:
cache = InMemoryCacheBackend()
class CachedUserRepo(CachedBaseRepository[User]):
    def __init__(self):
        super().__init__("users", User, cache)

Classes

BaseDocument

Bases: BaseModel

Base document class with common MongoDB fields.

Inherit from this class to create Pydantic models that represent MongoDB documents. It includes automatic handling of the _id field and timestamps.

Attributes:

Name	Type	Description
id	PyObjectId | None	The MongoDB document ID (aliased to _id).
created_at	datetime	Timestamp when the document was created.
updated_at	datetime	Timestamp when the document was last updated.

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"})

CachedBaseRepository

CachedBaseRepository(
    collection_name: str,
    model: type[T],
    cache_backend: CacheBackend,
    config: CacheConfig | None = None,
)

Bases: BaseRepository[T], Generic[T]

Repository that reads and writes through a cache backend.

Wraps an existing BaseRepository with an ID-keyed cache. Reads consult the backend first and fall through to MongoDB on a miss, populating the cache on success. Writes invalidate or refresh the affected key.

Notes

Guarantees:

- The cache holds the raw document shape (``model_dump``), so FK
  references round-trip as hex strings, not populated models.
- When ``config.enabled`` is False the repository behaves exactly
  like its parent with no cache access.

Initialize the cached repository.

Parameters:

Name	Type	Description	Default
collection_name	str	Name of the MongoDB collection.	required
model	type[T]	The Pydantic model class.	required
cache_backend	CacheBackend	Backend used to store and fetch entries.	required
config	Optional[CacheConfig]	Cache configuration; a default CacheConfig is used when None.	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

Insert a document and cache the raw snapshot.

Parameters:

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

Returns:

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

delete async

delete(id: str | ObjectId) -> bool

Delete a document and remove its cache entry.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID.	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

Fetch a document, reading through the cache when enabled.

Parameters:

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

Returns:

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

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)]
)

invalidate_cache async

invalidate_cache(id: str | ObjectId) -> None

Remove a single document's cache entry.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID to invalidate.	required

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) -> T | None

Update a document and refresh its cache entry.

Parameters:

Name	Type	Description	Default
id	Union[str, ObjectId]	The document ID.	required
data	dict	Fields to set via $set.	required

Returns:

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

warm_cache async

warm_cache(ids: list[str | ObjectId]) -> int

Pre-populate the cache for a set of document IDs.

Docs already present in the cache are skipped.

Parameters:

Name	Type	Description	Default
ids	list[Union[str, ObjectId]]	Document IDs to warm.	required

Returns:

Name	Type	Description
int	int	Number of entries added to the cache.

ModelRegistry

Registry for managing multiple models and their collections.

This registry allows central management of collections and their associated indexes, making it easier to perform mass initialization at application startup.

Functions

get_cache_backend classmethod

get_cache_backend() -> CacheBackend | None

Get the registered cache backend instance.

Returns:

Type	Description
CacheBackend | None	Optional[CacheBackend]: The cache backend, if registered.

get_model classmethod

get_model(collection_name: str) -> type[BaseDocument]

Retrieve a registered model by its collection name.

Parameters:

Name	Type	Description	Default
collection_name	str	The name of the collection.	required

Returns:

Type	Description
type[BaseDocument]	type[BaseDocument]: The registered model class.

Raises:

Type	Description
KeyError	If the model for the given collection is not registered.

initialize_all async classmethod

initialize_all(
    db: AsyncIOMotorDatabase | None = None,
) -> None

Initialize all registered collections and create indexes.

This method should be called during application startup to ensure all necessary indexes exist in the database.

Parameters:

Name	Type	Description	Default
db	Optional[AsyncIOMotorDatabase]	Database instance. If not provided, uses the global database from MongoConnectionManager.	None

initialize_cache async classmethod

initialize_cache() -> None

Initialize the registered cache backend.

Must be called after set_cache_backend() and before any cache-backed repository operations. Typically called right after MongoDB connection is established.

Raises:

Type	Description
RuntimeError	If no cache backend has been registered.

list_collections classmethod

list_collections() -> list[str]

List all registered collection names.

Returns:

Type	Description
list[str]	list[str]: A list of collection names.

register classmethod

register(
    collection_name: str,
    model: type[BaseDocument],
    indexes: list[tuple] | None = None,
) -> None

Register a model with its collection and indexes.

Parameters:

Name	Type	Description	Default
collection_name	str	Name of the MongoDB collection.	required
model	type[BaseDocument]	Document model class (subclass of BaseDocument).	required
indexes	list[Any]	List of index specifications. Each spec is passed directly to pymongo's create_index. Supported forms: - single field: ("email", 1) - compound: [("author_id", 1), ("created_at", -1)] - with options: {"keys": [("username", 1)], "options": {"unique": True}}	None

Example

ModelRegistry.register("users", UserDocument, indexes=[("email", 1)]) ModelRegistry.register( "posts", PostDocument, indexes=[[("author_id", 1), ("created_at", -1)]], )

set_cache_backend classmethod

set_cache_backend(backend: CacheBackend) -> None

Register a cache backend for all cache-enabled repositories.

Should be called after MongoDB connection is established, before cache-backed repositories are used.

Parameters:

Name	Type	Description	Default
backend	CacheBackend	A CacheBackend instance (InMemoryCacheBackend or RedisCacheBackend).	required

shutdown_cache async classmethod

shutdown_cache() -> None

Shutdown the registered cache backend gracefully.

Should be called during application shutdown to clean up background tasks (e.g., in-memory TTL cleanup).

MongoConnectionManager

Manages MongoDB connections with async lifecycle.

This class provides a singleton-like manager for the MongoDB client and database instances, ensuring they are properly initialized and closed across the application lifecycle.

Functions

connect async classmethod

connect(
    uri: str, db_name: str, **kwargs: Any
) -> AsyncIOMotorDatabase

Connect to MongoDB and initialize the shared client.

Parameters:

Name	Type	Description	Default
uri	str	MongoDB connection URI (e.g., "mongodb://localhost:27017").	required
db_name	str	Name of the database to use.	required
**kwargs	Any	Additional Motor client options (e.g., maxPoolSize).	{}

Returns:

Name	Type	Description
AsyncIOMotorDatabase	AsyncIOMotorDatabase	The initialized database instance.

disconnect async classmethod

disconnect() -> None

Close the active MongoDB connection and cleanup resources.

get_client classmethod

get_client() -> AsyncIOMotorClient

Retrieve the current client instance.

Returns:

Name	Type	Description
AsyncIOMotorClient	AsyncIOMotorClient	The active Motor client instance.

Raises:

Type	Description
RuntimeError	If connect() has not been called yet.

get_database classmethod

get_database() -> AsyncIOMotorDatabase

Retrieve the current database instance.

Returns:

Name	Type	Description
AsyncIOMotorDatabase	AsyncIOMotorDatabase	The active database instance.

Raises:

Type	Description
RuntimeError	If connect() has not been called yet.

lifespan async classmethod

lifespan(
    uri: str, db_name: str, **kwargs: Any
) -> AbstractAsyncContextManager[AsyncIOMotorDatabase]

Async context manager for managing connection lifecycle.

Designed for use with FastAPI or other frameworks supporting lifespan management.

Notes

Connects to MongoDB on entry, yields the active database instance to the context body, and disconnects on exit.

Usage

@asynccontextmanager async def lifespan(app: FastAPI): async with MongoConnectionManager.lifespan(uri, db_name): yield

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.

TransactionManager

Simplified multi-document transaction handling.

This class provides helpers for executing operations within a MongoDB transaction, ensuring ACID compliance for multi-document updates.

Functions

execute_transaction async classmethod

execute_transaction(
    operations: list[
        Callable[
            [AsyncIOMotorClientSession], Awaitable[Any]
        ]
    ],
    **kwargs: Any
) -> list[Any]

Execute multiple operations within a single transaction.

Parameters:

Name	Type	Description	Default
operations	List[Callable[[AsyncIOMotorClientSession], Awaitable[Any]]]	A list of async callables that accept a session parameter and return a result.	required
**kwargs	Any	Transaction options.	{}

Returns:

Type	Description
list[Any]	List[Any]: A list containing the results of each operation.

Example

results = await TransactionManager.execute_transaction([ lambda s: repo1.create(data1, session=s), lambda s: repo2.update(id, data2, session=s), ])

start_session async classmethod

start_session(
    **kwargs: Any,
) -> AbstractAsyncContextManager[AsyncIOMotorClientSession]

Start a transaction session as an async context manager.

Notes

Yields the active session with a started transaction; callers run their operations against the session inside the with-block.

Usage

async with TransactionManager.start_session() as session: await collection.insert_one(doc, session=session) await other_collection.update_one(filter, update, session=session)