Created
November 16, 2020 14:47
-
-
Save mass1ve-err0r/07bbc2a09bb7adfe44202bdbb37b0dfa to your computer and use it in GitHub Desktop.
(sync) driver template for 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
| import urllib.parse | |
| from pymongo import MongoClient | |
| from pymongo.collection import Collection | |
| from pymongo.database import Database | |
| from typing import Optional, Any | |
| from os import environ as env | |
| # <editor-fold desc="[Globals]"> | |
| mUser = urllib.parse.quote_plus(str(env.get('MONGO_USER'))) | |
| mPass = urllib.parse.quote_plus(str(env.get('MONGO_PASS'))) | |
| mAddr = str(env.get('MONGO_ADDR')) | |
| mAuthDB = str(env.get('MONGO_AUD')) | |
| mTargetDB = str(env.get('MONGO_MAIN_DB')) | |
| # </editor-fold> | |
| class MongoDBHandler: | |
| def __init__(self, collectionName: str) -> None: | |
| self.mainHandle: MongoClient = MongoClient("mongodb://" + mUser + ":" + mPass + "@" + mAddr + "/?authSource=" + mAuthDB) | |
| self.dbHandle: Database = self.mainHandle[mTargetDB] | |
| self.collectionHandle: Collection = self.dbHandle[collectionName] | |
| # <editor-fold desc="[Existence Operations]"> | |
| def ensureExistence(self, query: dict) -> bool: | |
| """ | |
| Check if an item exists in the Collection. | |
| :param query: The query (py-dict) to execute | |
| :return: True if exists, False if not | |
| """ | |
| occ: int = self.collectionHandle.count_documents(query, limit=1) | |
| if occ == 1: | |
| return True | |
| else: | |
| return False | |
| def ensureExistenceInAnyArray(self, qArrayName: str, query: dict) -> bool: | |
| """ | |
| Check if an Object exists inside an Array globally in the collection of Objects. | |
| :param qArrayName: The Array's name to filter for | |
| :param query: The Filter (py-dict) to scan with | |
| :return: NoneType | |
| """ | |
| scan_query = {qArrayName: query} | |
| occ: int = self.collectionHandle.count_documents(scan_query, limit=1) | |
| if occ == 1: | |
| return True | |
| else: | |
| return False | |
| # </editor-fold> | |
| # <editor-fold desc="[Get Operations]"> | |
| def getItem(self, query: dict) -> Optional[dict]: | |
| """ | |
| Retrieve Object from the Collection. | |
| :param query: The query (py-dict) to execute | |
| :return: Object from Mongo as dict | |
| """ | |
| rv = self.collectionHandle.find_one(query) | |
| return rv | |
| def getBooleanFromObject(self, qFilter: dict, qBoolName: str) -> bool: | |
| """ | |
| Get the bool directly from an Object. | |
| :param qFilter: The query Filter (py-dict) to limit by | |
| :param qBoolName: The boolean to switch | |
| :return: NoneType | |
| """ | |
| bool_query = {qBoolName: True} | |
| rv_dict = self.collectionHandle.find_one(qFilter, bool_query) | |
| try: | |
| rv_bool = rv_dict[qBoolName] | |
| return rv_bool | |
| except KeyError: | |
| return False | |
| except Exception: | |
| return False | |
| def getFieldValueFromObject(self, qFilter: dict, qFieldName: str) -> Optional[Any]: | |
| """ | |
| Get the a Python-Object directly from an Object. | |
| :param qFilter: The query Filter (py-dict) to limit by | |
| :param qFieldName: The Field to get from | |
| :return: Any-Type | |
| """ | |
| rv_dict = self.collectionHandle.find_one(qFilter) | |
| try: | |
| rv = rv_dict[qFieldName] | |
| return rv | |
| except KeyError: | |
| return None | |
| except Exception: | |
| return None | |
| def getListObjectsWithFilter(self, qFilter: dict) -> list: | |
| """ | |
| Get a list of Objects by filter. | |
| :param qFilter: The query Filter (py-dict) to limit by | |
| :return: List of Objects matching the query | |
| """ | |
| ret_list = [] | |
| rCursor = self.collectionHandle.find(qFilter) | |
| for _obj in rCursor: | |
| ret_list.append(_obj) | |
| return ret_list | |
| # </editor-fold> | |
| # <editor-fold desc="[Deletion Operations]"> | |
| def deleteObjectFromCollection(self, query: dict) -> None: | |
| """ | |
| Delete an Object from the Collection with a given key. | |
| :param query: The query (py-dict) to execute | |
| :return: NoneType | |
| """ | |
| self.collectionHandle.delete_one(query) | |
| return | |
| def removeObjectFromArray(self, qFilter: dict, qArrayName: str, qObjectFilter: dict) -> None: | |
| """ | |
| Remove an Element from an Object's Array. | |
| :param qFilter: The Filtering query (py-dict) | |
| :param qArrayName: The Array identifier to remove from | |
| :param qObjectFilter: The Object's filter (py-dict) to id and remove with | |
| :return: NoneType | |
| """ | |
| removal_query = {"$pull": {qArrayName: qObjectFilter}} | |
| self.collectionHandle.update_one(qFilter, removal_query) | |
| return | |
| # </editor-fold> | |
| # <editor-fold desc="[Add Operations]"> | |
| def addObjectToCollection(self, _object: dict) -> None: | |
| """ | |
| Add an Object to the Collection. | |
| :param _object: Object to add | |
| :return: NoneType | |
| """ | |
| self.collectionHandle.insert_one(_object) | |
| return | |
| def insertObjectIntoArray(self, qFilter: dict, qArrayName: str, qObject: dict) -> None: | |
| """ | |
| Insert an Element into an Array of the Object found by the qFilter. | |
| :param qFilter: The Filtering query (py-dict) | |
| :param qArrayName: The Array identifier to insert into | |
| :param qObject: The Object (py-dict) to add | |
| :return: NoneType | |
| """ | |
| insertion_query = {"$push": {qArrayName: qObject}} | |
| self.collectionHandle.update_one(qFilter, insertion_query) | |
| return | |
| # </editor-fold> | |
| # <editor-fold desc="[Other Operations]"> | |
| def toggleBooleanValueForObject(self, qFilter: dict, qBoolName: str, qValue: bool) -> None: | |
| """ | |
| Toggle a boolean of an Object. | |
| :param qFilter: The query Filter (py-dict) to limit by | |
| :param qBoolName: The boolean to switch | |
| :param qValue: Value to change to | |
| :return: NoneType | |
| """ | |
| toggle_query = {"$set": {qBoolName: qValue}} | |
| self.collectionHandle.update_one(qFilter, toggle_query) | |
| return | |
| # </editor-fold> |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment