trolldb.database.mongodb module

The module which handles database CRUD operations for MongoDB.

It is based on the following libraries:

Note

Some functions/methods in this module are decorated with the Pydantic @validate_call which checks the arguments during the function calls.

trolldb.database.mongodb.CoroutineLike

A simple type hint for a coroutine of any type.

alias of Coroutine[Any, Any, T]

trolldb.database.mongodb.CoroutineDocument

Coroutine type hint for document like objects.

alias of Coroutine[Any, Any, _DocumentType | None]

trolldb.database.mongodb.CoroutineStrList

Coroutine type hint for a list of strings.

alias of Coroutine[Any, Any, list[str]]

async trolldb.database.mongodb.get_id(doc: Coroutine[Any, Any, _DocumentType | None]) → str[source]

Retrieves the ID of a document as a simple flat string.

Note

The rationale behind this method is as follows. In MongoDB, each document has a unique ID which is of type bson.objectid.ObjectId. This is not suitable for purposes when a simple string is needed, hence the need for this method.

Parameters:

doc – A MongoDB document in the coroutine form. This could be e.g. the result of applying the standard find_one method from MongoDB on a collection given a filter.

Returns:

The ID of a document as a simple string. For example, when applied on a document with _id: ObjectId('000000000000000000000000'), the method returns '000000000000000000000000'.

async trolldb.database.mongodb.get_ids(docs: motor.motor_asyncio.AsyncIOMotorCommandCursor | motor.motor_asyncio.AsyncIOMotorCursor) → list[str][source]

Similar to get_id() but for a list of documents.

Parameters:

docs – A list of MongoDB documents as AsyncIOMotorCommandCursor or AsyncIOMotorCursor. This could be e.g. the result of applying the standard aggregate method from MongoDB on a collection given a pipeline.

Returns:

The list of all IDs, each as a simple string.

class trolldb.database.mongodb.MongoDB[source]

Bases: object

A wrapper class around the motor async driver for Mongo DB.

It includes convenience methods tailored to our specific needs. As such, the initialize() method returns a coroutine which needs to be awaited.

Note

This class is not meant to be instantiated! That’s why all the methods in this class are decorated with @classmethods. This choice has been made to guarantee optimal performance, i.e. for each running process there must be only a single motor client to handle all database operations. Having different clients which are constantly opened/closed degrades the performance. The expected usage is that we open a client in the beginning of the program and keep it open until the program finishes. It is okay to reopen/close the client for testing purposes when isolation is needed.

Note

The main difference between this wrapper class and the original motor driver class is that we attempt to access the database and collections during the initialization to see if we succeed or fail. This is contrary to the behaviour of the motor driver which simply creates a client object and does not attempt to access the database until some time later when an actual operation is performed on the database. This behaviour is not desired for us, we would like to fail early!

__client: ClassVar[motor.motor_asyncio.AsyncIOMotorClient | None] = None
__database_config: ClassVar[DatabaseConfig | None] = None
__main_collection: ClassVar[motor.motor_asyncio.AsyncIOMotorCollection | None] = None
__main_database: ClassVar[motor.motor_asyncio.AsyncIOMotorDatabase | None] = None
default_database_names: ClassVar[list[str]] = ['admin', 'config', 'local']

MongoDB creates these databases by default for self usage.

classmethod initialize(database_config: DatabaseConfig)

Initializes the motor client. Note that this method has to be awaited!

Parameters:

database_config – An object of type DatabaseConfig which includes the database configurations.

Warning

The timeout is given in seconds in the configurations, while the MongoDB uses milliseconds.

Returns:

On success None.

Raises:

  • ValidationError – If the method is not called with arguments of valid type.

  • SystemExit(errno.EIO) – If connection is not established, i.e. ConnectionFailure.

  • SystemExit(errno.EIO) – If the attempt times out, i.e. ServerSelectionTimeoutError.

  • SystemExit(errno.EIO) – If one attempts reinitializing the class with new (different) database configurations without calling close() first.

  • SystemExit(errno.EIO) – If the state is not consistent, i.e. the client is closed or None but the internal database configurations still exist and are different from the new ones which have been just provided.

  • SystemExit(errno.ENODATA) – If either database_config.main_database_name or database_config.main_collection_name does not exist.

classmethod is_initialized() → bool[source]

Checks if the motor client is initialized.

classmethod close() → None[source]

Closes the motor client.

classmethod list_database_names() → Coroutine[Any, Any, list[str]][source]

Lists all the database names.

classmethod main_collection() → motor.motor_asyncio.AsyncIOMotorCollection[source]

A convenience method to get the main collection.

Returns:

The main collection which resides inside the main database. Equivalent to MongoDB.client()[<main_database_name>][<main_collection_name>].

classmethod main_database() → motor.motor_asyncio.AsyncIOMotorDatabase[source]

A convenience method to get the main database.

Returns:

The main database which includes the main collection, which in turn includes the desired documents.

This is equivalent to MongoDB.client()[<main_database_name>].

classmethod get_collection(database_name: str | None = None, collection_name: str | None = None) → motor.motor_asyncio.AsyncIOMotorCollection

Gets the collection object given its name and the database name in which it resides.

Parameters:

  • database_name – The name of the parent database which includes the collection.

  • collection_name – The name of the collection which resides inside the parent database labelled by database_name.

Returns:

The database object. In case of None for both the database name and collection name, the main collection will be returned.

Raises:

  • ValidationError – If the method is not called with arguments of valid type.

  • KeyError – If the database name exists, but it does not include any collection with the given name.

  • TypeError – If only one of the database or collection names are None.

  • ... – This method relies on get_database() to check for the existence of the database which can raise exceptions. Check its documentation for more information.

classmethod get_database(database_name: str | None = None) → motor.motor_asyncio.AsyncIOMotorDatabase

Gets the database object given its name.

Parameters:

database_name – The name of the database to retrieve.

Returns:

The database object.

Raises:

  • ValidationError – If the method is not called with arguments of valid type.

  • KeyError – If the database name does not exist in the list of database names.

trolldb.database.mongodb.mongodb_context(database_config: DatabaseConfig) → AsyncGenerator[source]

An asynchronous context manager to connect to the MongoDB client.

It can be either used in PRODUCTION or in TESTING environments.

Note

Since the MongoDB is supposed to be used statically, this context manager does not yield anything! One can simply use MongoDB inside the context manager.

Parameters:

database_config – The configuration of the database.