Skip to content

Instantly share code, notes, and snippets.

@aybruhm
Last active February 7, 2023 21:50
Show Gist options
  • Save aybruhm/33c688d3e96ea6ff0f3b707d3d57e6a6 to your computer and use it in GitHub Desktop.
Save aybruhm/33c688d3e96ea6ff0f3b707d3d57e6a6 to your computer and use it in GitHub Desktop.
The below code defines a Python class named MongoDB that provides a base wrapper for connecting to a MongoDB database.
# Third-party Imports
import pymongo
import motor
class MongoDB(object):
"""Base wrapper to connect to MongoDB"""
def __init__(self):
"""Initialize the MongoDB connection"""
self.url = "mongodb+srv://<username>:<password>@<url>/<db>?retryWrites=true&w=majority"
self.client: pymongo.MongoClient = None
def start_client(self):
"""This method creates an instance of the `AsyncIOMotorClient`"""
self.client = motor.motor_asyncio.AsyncIOMotorClient(self.url)
def get_client(self) -> pymongo.MongoClient:
"""Get the Mongo client"""
if self.client is None:
self.start_client()
return self.client
def stop_client(self):
"""Gracefully close the Mongo client"""
self.client.close()
self.client = None
def get_db(self):
"""Get the <db_name> database"""
db = self.get_client().get_database("<db_name>")
if db is None:
raise Exception({"message": "Database not found"})
return db
# initialize the MongoDB connection
mongo = MongoDB()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment