MongoDBConnection provide the mongo_client on every call to the get method. And will close the connection when exiting the "with" statement.

Example

config.yaml

core_lib:
  mongodb:
    _target_: core_lib.connection.mongodb_connection_factory.MongoDBConnectionFactory
    config:
      url:
        protocol: mongodb+srv
        host: server.example.com
        port: 27017
        username: ${oc.env:MONGO_USERNAME}
        password: ${oc.env:MONGO_PASSWORD}

your_core_lib.py

from core_lib.connection.mongodb_connection_factory import MongoDBConnectionFactory


class YourCoreLib(CoreLib):
    
    def __init__(self, conf: DictConfig):
        self.config = conf.core_lib.mongodb
        mongodb = MongoDBConnectionFactory(self.config)
        self.user = UserDataAccess(mongodb)

user_data_access.py



class UserDataAccess(DataAccess):

    def __init__(self, db: MongoDBConnectionFactory):
        self.mongodb = db

    def get(self, user_id: int) -> pymongo.cursor.Cursor:
        with self.mongodb.get() as client:
            return client.db.collection.find_one({'_id': user_id})

    def create(self, data: dict) -> pymongo.results.InsertOneResult:
        with self.mongodb.get() as client:
            return client.db.collection.insert_one(data)

    def update(self, user_id: int, data: dict) -> pymongo.results.UpdateResult:
        with self.mongodb.get() as client:
            return client.db.collection.update_one({'_id': user_id}, {'$set': data})

    def delete(self, user_id: int) -> pymongo.results.DeleteResult:
        with self.mongodb.get() as client:
            return client.db.collection.delete_one({'id': user_id})

MongoDBConnection

core_lib.connection.mongodb_connection.MongoDBConnection [source]

This class gets the mongo_client on initialization.

class MongoDBConnection(Connection):
    def __init__(self, mongo_client: MongoClient):
        self.mongo_client = mongo_client

Arguments

  • mongo_client (MongoClient): The mongo_client provided from MongoDBConnectionFactory.

MongoDBConnectionFactory

core_lib.connection.mongodb_connection_factory.MongoDBConnectionFactory [source]

This class is responsible for initializing the mongo_client (MongoClient) from pymongo and passing mongo_client on to the MongoDBConnection in the get() method.

Functions

__init__()

core_lib.connection.mongodb_connection_factory.MongoDBConnectionFactory.__init__() [source]

Responsible for initializing mongo_client from the configuration

def __init__(self, config: DictConfig):

Arguments

  • config (DictConfig): MongoDB section config from Core-Lib's config.

client()

core_lib.connection.mongodb_connection_factory.MongoDBConnectionFactory.client() [source]

Returns the MongoClient from pymongo.

def client(self) -> MongoClient:

Returns

(MongoClient): Returns the MongoClient from pymongo

get()

core_lib.connection.mongodb_connection_factory.MongoDBConnectionFactory.get() [source]

Returns the MongoDBConnection instance and passes mongo_client as an argument to it.

def get(self, *args, **kwargs) -> MongoDBConnection:

Arguments

  • *args, **kwargs: The args and kwargs of the function.

Returns

(MongoDBConnection): Returns the MongoDBConnection instance.