Mongo Manager
This is a detailed and didactic documentation of the MongoDBManager class, designed to make it easier for developers who will integrate this tool into their projects to understand it.
On this page
#Overview
The MongoDBManager class acts as a Wrapper (or facilitator) for the pymongo library. Its main goal is to simplify interaction with MongoDB databases, abstracting the complexity of managing connections and ensuring that common CRUD (Create, Read, Update, Delete) operations are performed in a standardized way.
It is ideal for applications that need flexible data persistence, automatically handling metadata such as creation and update dates.
#Execution Flow
The class follows a Lazy Loading logic:
- Instantiation: When the object is created, it only stores the connection URI. The database is not accessed yet.
- First Call: When any operation method (such as
save_payload) is invoked, it internally calls theconnect()method. - Persistent Connection: The
MongoClientis created and stored in theself.clientattribute. All subsequent calls reuse this same connection to save resources. - Processing: The method runs the requested logic (insert, search, etc.).
- Error Handling: If something fails, the connection is closed preventively to avoid memory leaks or "stuck" connections.
#Methods Table
| Method | Brief Description |
|---|---|
__init__ | Configures the connection URI (via argument or environment variable). |
connect | Establishes or reuses the connection to the MongoDB server. |
close_connection | Ends the active connection and clears the client. |
save_payload | Inserts a new document with a creation timestamp. |
fetch_documents | Searches documents with filters and a limit option. |
update_documents | Updates one or several records, adding an edit timestamp. |
delete_documents | Permanently removes records from the database. |
#Architecture and Insights
- State Management: The class uses a Singleton-like pattern for the connection within the instance, ensuring you do not open hundreds of unnecessary connections.
- Data Safety: In the update method, the code explicitly removes the
_idkey from the update payload (new_values.pop("_id", None)). This avoids common MongoDB errors caused by trying to overwrite a document's immutable ID. - Traceability: The automatic inclusion of
_created_atandupdated_atturns simple documents into auditable records, making later debugging and data analysis easier. - Error Handling: The use of
RuntimeErrorwraps complex network errors in messages that are readable for the application developer.
#Class Details
#MongoDBManager Class
Description
Central MongoDB persistence manager that encapsulates the connection lifecycle and CRUD operations, ensuring resource reuse and metadata standardization.
Arguments
mongo_uri(str, optional): Connection string. If omitted, it looks upos.getenv("MONGO_URI")or uses the local default.
#Methods
#1. connect
Description: Ensures that there is an active connection to the database.
Arguments: None.
Returns: MongoClient (pymongo connection object).
Raises: Exception in case of network or credentials failure.
Examples:
manager = MongoDBManager()
client = manager.connect()#2. save_payload
Description: Inserts a dictionary into the database and adds the creation date.
Arguments:
database_name(str): Name of the database.collection_name(str): Name of the collection.payload(dict): Data to be saved.
Returns: dict containing the status and the generated ID.
Raises: RuntimeError if the insert fails.
Examples:
manager.save_payload("log_db", "events", {"event": "login_success"})#3. fetch_documents
Description: Locates documents that match the given criteria.
Arguments:
database_name(str): Name of the database.collection_name(str): Name of the collection.filter(dict): Search filter (e.g.{"status": "active"}).limit(int): Maximum number of results.
Returns: List[dict] (List of documents found).
Raises: RuntimeError.
Examples:
users = manager.fetch_documents("app", "users", {"age": {"$gt": 18}}, limit=10)#4. update_documents
Description: Modifies existing documents and records the modification date.
Arguments:
filter(dict): Criterion to find the documents.new_values(dict): Data to be updated.multi(bool): IfTrue, updates all documents found; ifFalse, only the first.
Returns: dict with metrics (matched_count, modified_count).
Raises: RuntimeError.
Examples:
manager.update_documents("app", "users", {"id": 1}, {"status": "premium"})#5. delete_documents
Description: Permanently removes documents from the collection.
Arguments:
filter(dict): Deletion criterion.multi(bool): Defines whether to delete one or all the records found.
Returns: dict with deleted_count.
Raises: RuntimeError.
Examples:
manager.delete_documents("app", "temp_data", {"expired": True}, multi=True)Source: src/database/no_relational_db/mongo_manager.py