Last active
February 7, 2023 21:50
-
-
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.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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