Created
May 6, 2022 15:32
-
-
Save itszechs/2e4bfd1f504a3f625268b26c21baab1f to your computer and use it in GitHub Desktop.
Mongo class for CRUD operations using pymongo
This file contains hidden or 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
| from typing import Any, List | |
| import re | |
| import pymongo | |
| import urllib.parse | |
| class MongoDb: | |
| SERVER_URL_REGEX = r"^(.*)://(.*):(.*)@(.*\.mongodb\.net)" | |
| def __init__( | |
| self, | |
| server_url: str, | |
| db_name: str, | |
| collection_name: str | |
| ) -> None: | |
| self.__db_name = db_name | |
| self.__collection_name = collection_name | |
| self.__split_server_url(server_url) | |
| # Connect to MongoDB | |
| self.__client = pymongo.MongoClient( | |
| self.server_url, | |
| retryWrites=True, | |
| w="majority" | |
| ) | |
| print(self.__client.server_info()) | |
| self.__db = self.__client[self.__db_name] | |
| self.__collection = self.__db[self.__collection_name] | |
| def __split_server_url(self, server_url) -> None: | |
| """ | |
| Splits the server url into 3 parts | |
| - protocol | |
| - username | |
| - password | |
| - host | |
| """ | |
| match = re.match(self.SERVER_URL_REGEX, server_url) | |
| if match: | |
| try: | |
| self.__protocol = match.group(1) | |
| self.__username = match.group(2) | |
| self.__password = match.group(3) | |
| self.__host = match.group(4) | |
| except IndexError: | |
| print("Invalid server url") | |
| self.__get_db_url() | |
| else: | |
| raise ValueError("Invalid server url") | |
| def __get_db_url(self) -> None: | |
| """ | |
| Url encode the credentials and create the db url | |
| """ | |
| username = urllib.parse.quote_plus(self.__username) | |
| password = urllib.parse.quote_plus(self.__password) | |
| self.server_url = f"{self.__protocol}://{username}:{password}@" \ | |
| f"{self.__host}/{self.__db_name}" | |
| def upsert(self, data: dict) -> None: | |
| """ | |
| Insert data into the collection | |
| if it does not exist, otherwise update it | |
| """ | |
| self.__collection.update_one( | |
| data, | |
| {"$set": data}, | |
| upsert=True | |
| ) | |
| def insert(self, data: dict) -> Any: | |
| """ | |
| Inserts data into the collection | |
| and returns the inserted id | |
| Keep in mind, MongoDb does not have conflicts on insertion | |
| as they automatically append a unique id to the data | |
| Advised to use `upsert` function instead | |
| """ | |
| return self.__collection.insert_one(data).inserted_id | |
| def query(self, query: dict = {}) -> List: | |
| """ | |
| Reads data from the collection | |
| If no query is provided, it will return all data | |
| """ | |
| return self.__collection.find(query) | |
| def update(self, query: dict, data: dict) -> int: | |
| """ | |
| Updates data in the collection | |
| and returns modified count | |
| """ | |
| return self.__collection.update_one( | |
| query, | |
| {"$set": data} | |
| ).modified_count | |
| def delete(self, data: dict) -> int: | |
| """ | |
| Deletes data from the collection | |
| and returns deleted count | |
| """ | |
| return self.__collection.delete_one(data).deleted_count | |
| def drop(self) -> None: | |
| """ | |
| Drops the collection | |
| """ | |
| self.__collection.drop() | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment