Creating a python class to return db connection

Question

Background

After some struggle I have managed to create a cluster for Amazon DocumentDb. Now I want to write a simple python class that when instantiated returns a client connection and allows me to insert a document. Upon completion of inserting document it closes connection safely.

After some more struggle I managed to get the following to work.

MY CODE

import pymongo
import sys

client = pymongo.MongoClient(
    "mongodb://foousername:foopassword@docdb-poc-test.cluster-z7zzzzzzzztv.us-east-1.docdb.amazonaws.com:27017/?ssl=true&ssl_ca_certs=rds-combined-ca-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false"
)

##Specify the database to be used
db = client.sample_database

##Specify the collection to be used
col = db.sample_collection


col.insert_one({"id123": "Amazon DocumentDB"})


##Close the connection
client.close()

ISSUE

This code is simple it creates a connection and then I am able to specify the database and collection where i want to insert my document. The insertion part makes sense to me. I am struggling to encapsulate the whole client string in to a method. I tried the following.

def get_mongodb_connection(
    host: str,
    port: Union[str, int],
    user: str,
    password: str,
    database: str,
):
    return pymongo.MongoClient(
        host=host,
        port=int(port),
        username=user,
        password=password,
    )[database]

x = get_mongodb_connection("docdb-poc-test.cluster-z7zzzzzzzztv.us-east-1.docdb.amazonaws.com","27017","foousername","foopassword","sample_database")  

The following x.list_collection_names() resulted in ServerSelectionTimeoutError

I know my original code works and is able to get me connected and let me insert document in the db but i want to create a python class which will have a method that returns a connection as it will let me easily do more things.

I highly appreciate any input here.

Answer 1

Without seeing the rest of your code, and only using your code as closely as possible, I came up with this for you:

import pymongo
import sys

class MyMongoDBClass:
    def __init__(
        self,
        host: str,
        port: Union[str, int],
        user: str,
        password: str,
        database: str
        ):

        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.database = database


        self._client = pymongo.MongoClient(
            f'{self.user}{self.password}@{self.host}:{self.port}/?ssl=true&ssl_ca_certs=rds-combined-ca-bundle.pem&replicaSet=rs0&readPreference=secondaryPreferred&retryWrites=false'
        )
        self._db = self._client[self.database]
        self._collection = self._db.sample_collection

    def close():
        _client.close()

    def insert_one(self, key: str, value: str):
        self._collection.insert_one({key: value})

if __name__ == '__main__':
    my_db = MyMongoDBClass(
        'docdb-poc-test.cluster-z7zzzzzzzztv.us-east-1.docdb.amazonaws.com',
        '27017',
        'foousername',
        'foopassword',
        'sample_database'
    )

    my_db.insert_one('id123', 'Amazon DocumentDB')

    my_db.close()