first commit

.idea/.gitignore

+# Default ignored files
+/shelf/
+/workspace.xml

.idea/IoTManagerTesting.iml

+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/venv" />
+    </content>
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
\ No newline at end of file

.idea/inspectionProfiles/Project_Default.xml

+<component name="InspectionProjectProfileManager">
+  <profile version="1.0">
+    <option name="myName" value="Project Default" />
+    <inspection_tool class="PyPep8NamingInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true">
+      <option name="ignoredErrors">
+        <list>
+          <option value="N806" />
+        </list>
+      </option>
+    </inspection_tool>
+    <inspection_tool class="PyUnresolvedReferencesInspection" enabled="true" level="WARNING" enabled_by_default="true">
+      <option name="ignoredIdentifiers">
+        <list>
+          <option value="scripts.config.application_config" />
+        </list>
+      </option>
+    </inspection_tool>
+  </profile>
+</component>
\ No newline at end of file

.idea/inspectionProfiles/profiles_settings.xml

+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>
\ No newline at end of file

.idea/misc.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.9 (IoTManagerTesting)" project-jdk-type="Python SDK" />
+</project>
\ No newline at end of file

.idea/modules.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/IoTManagerTesting.iml" filepath="$PROJECT_DIR$/.idea/IoTManagerTesting.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

main.py

+from scripts.core.handlers import create_users
+
+test = create_users.obj_mongo

mongo/mongo_connection.py

+from scripts.utils.mongo_utils import MongoConnect
+
+mongo_obj = MongoConnect(uri="mongodb://localhost:27017")
+
+mongo_client = mongo_obj()
+
+CollectionBaseClass = mongo_obj.get_base_class()

scripts/constants/db_constants.py

+class DatabaseConstants:
+    collection_user_details = "userDetails"
+    collection_user_roles = "userRoles"
+    collection_user_devices = "userDevices"

scripts/core/handlers/create_users.py

+import datetime
+
+import bcrypt
+from fastapi.security import OAuth2PasswordBearer
+from passlib.context import CryptContext
+from pymongo import MongoClient
+
+from scripts.constants.db_constants import DatabaseConstants
+from scripts.utils.mongo_sync import MongoCollectionBaseClass
+
+obj_mongo = MongoCollectionBaseClass(mongo_client=MongoClient(), database="userDB",
+                                     collection=DatabaseConstants.collection_user_details)
+while True:
+    name = input("Enter the name")
+    email = input("Enter email")
+    password = input("enter password")
+    salt = bcrypt.gensalt()
+    hashed_password = bcrypt.hashpw(password, salt)
+    is_alive = True
+    user_role = "super admin"
+    # Create a Python datetime object
+    dt = datetime.datetime.now()
+    created_at = datetime.datetime.now()
+    updated_at = datetime.datetime.now()
+    obj_mongo.insert_one({"name": name, "email": email, "password": password, "user_role": user_role,
+                          "is_alive": is_alive, "created_at": created_at,
+                          "updated_at": updated_at})
+    break

scripts/schemes/model.py

+from pydantic import BaseModel, EmailStr
+
+
+class UserDetails(BaseModel):
+    name: str
+    email: EmailStr
+    password: str

scripts/utils/db_name_util.py

+import logging
+from functools import lru_cache
+from typing import Tuple
+
+import ujson as json
+
+
+@lru_cache()
+def get_db_name(project_id: str, database: str, delimiter: str = "__") -> str:
+    prefix_condition, val = check_prefix_condition(project_id)
+    if prefix_condition:
+        # Get the prefix name from mongo or default to project_id
+        prefix_name = val.get("source_meta", {}).get("prefix") or project_id
+        return f"{prefix_name}{delimiter}{database}"
+    return database
+
+
+def check_prefix_condition(project_id: str) -> Tuple[bool, dict]:
+    if not project_id:
+        logging.warning("Project ID is None! Cannot check for prefix!")
+        return False, {}
+    redis_resp = project_details_db.get(project_id)
+    if redis_resp is None:
+        raise ValueError(f"Unknown Project, Project ID: {project_id} Not Found!!!")
+    val: dict = json.loads(redis_resp)
+    if not val:
+        return False, {}
+
+    # Get the prefix flag to apply project_id prefix to any db
+    prefix_condition = bool(val.get("source_meta", {}).get("add_prefix_to_database"))
+    return prefix_condition, val

scripts/utils/mongo_sync.py

+from __future__ import annotations
+
+import logging
+from typing import Any, Dict, List, Mapping, Optional, Sequence, Tuple, Union
+
+from pymongo import MongoClient
+from pymongo.command_cursor import CommandCursor
+from pymongo.cursor import Cursor
+from pymongo.results import (
+    DeleteResult,
+    InsertManyResult,
+    InsertOneResult,
+    UpdateResult,
+)
+
+
+class MongoCollectionBaseClass:
+    def __init__(
+        self,
+        mongo_client: MongoClient,
+        database: str,
+        collection: str,
+        # project_id: Optional[str],
+    ) -> None:
+        self.client = mongo_client
+        self.database = database
+        self.collection = collection
+        # if project_id:
+        #     self.database = get_db_name(project_id=project_id, database=self.database)
+
+    def __repr__(self) -> str:
+        return f"{self.__class__.__name__}(database={self.database}, collection={self.collection})"
+
+    def insert_one(self, data: Dict) -> InsertOneResult:
+        """
+        The function is used to inserting a document to a collection in a Mongo Database.
+        :param data: Data to be inserted
+        :return: Insert ID
+        """
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.insert_one(data)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def insert_many(self, data: List) -> InsertManyResult:
+        """
+        The function is used to inserting documents to a collection in a Mongo Database.
+        :param data: List of Data to be inserted
+        :return: Insert IDs
+        """
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.insert_many(data)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def find(
+        self,
+        query: dict,
+        filter_dict: Optional[dict] = None,
+        sort: Union[
+            None, str, Sequence[Tuple[str, Union[int, str, Mapping[str, Any]]]]
+        ] = None,
+        skip: int = 0,
+        limit: Optional[int] = None,
+    ) -> Cursor:
+        """
+        The function is used to query documents from a given collection in a Mongo Database
+        :param query: Query Dictionary
+        :param filter_dict: Filter Dictionary
+        :param sort: List of tuple with key and direction. [(key, -1), ...]
+        :param skip: Skip Number
+        :param limit: Limit Number
+        :return: List of Documents
+        """
+        if sort is None:
+            sort = []
+        if filter_dict is None:
+            filter_dict = {"_id": 0}
+        database_name = self.database
+        collection_name = self.collection
+        try:
+            db = self.client[database_name]
+            collection = db[collection_name]
+            if len(sort) > 0:
+                cursor = (
+                    collection.find(
+                        query,
+                        filter_dict,
+                    )
+                    .sort(sort)
+                    .skip(skip)
+                )
+            else:
+                cursor = collection.find(
+                    query,
+                    filter_dict,
+                ).skip(skip)
+            if limit:
+                cursor = cursor.limit(limit)
+            return cursor
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def find_one(self, query: dict, filter_dict: Optional[dict] = None) -> dict | None:
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            if filter_dict is None:
+                filter_dict = {"_id": 0}
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.find_one(query, filter_dict)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def update_one(
+        self,
+        query: dict,
+        data: dict,
+        upsert: bool = False,
+        strategy: str = "$set",
+    ) -> UpdateResult:
+        """
+
+        :param strategy:
+        :param upsert:
+        :param query:
+        :param data:
+        :return:
+        """
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.update_one(query, {strategy: data}, upsert=upsert)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def update_many(
+        self, query: dict, data: dict, upsert: bool = False
+    ) -> UpdateResult:
+        """
+
+        :param upsert:
+        :param query:
+        :param data:
+        :return:
+        """
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.update_many(query, {"$set": data}, upsert=upsert)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def delete_many(self, query: dict) -> DeleteResult:
+        """
+        :param query:
+        :return:
+        """
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.delete_many(query)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def delete_one(self, query: dict) -> DeleteResult:
+        """
+        :param query:
+        :return:
+        """
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.delete_one(query)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def distinct(self, query_key: str, filter_json: Optional[dict] = None) -> list:
+        """
+        :param query_key:
+        :param filter_json:
+        :return:
+        """
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.distinct(query_key, filter_json)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def aggregate(
+        self,
+        pipelines: list,
+    ) -> CommandCursor:
+        try:
+            database_name = self.database
+            collection_name = self.collection
+            db = self.client[database_name]
+            collection = db[collection_name]
+            return collection.aggregate(pipelines)
+        except Exception as e:
+            logging.exception(e)
+            raise

scripts/utils/mongo_utils.py

+""" Mongo DB utility
+All definitions related to mongo db is defined in this module
+"""
+
+import logging
+
+from pymongo import MongoClient
+
+from scripts.utils import mongo_sync
+
+
+class MongoConnect:
+    def __init__(self, uri):
+        try:
+            self.uri = uri
+            self.client = MongoClient(uri, connect=False)
+        except Exception as e:
+            logging.exception(e)
+            raise
+
+    def __call__(self, *args, **kwargs):
+        return self.client
+
+    def get_client(self):
+        return self.client
+
+    def __repr__(self):
+        return f"Mongo Client(uri:{self.uri}, server_info={self.client.server_info()})"
+
+    @staticmethod
+    def get_base_class():
+        return mongo_sync.MongoCollectionBaseClass
+
+
+class MongoStageCreator:
+    @staticmethod
+    def add_stage(stage_name: str, stage: dict) -> dict:
+        return {stage_name: stage}
+
+    def projection_stage(self, stage: dict) -> dict:
+        return self.add_stage("$project", stage)
+
+    def match_stage(self, stage: dict) -> dict:
+        return self.add_stage("$match", stage)
+
+    def lookup_stage(self, stage: dict) -> dict:
+        return self.add_stage("$lookup", stage)
+
+    def unwind_stage(self, stage: dict) -> dict:
+        return self.add_stage("$unwind", stage)
+
+    def group_stage(self, stage: dict) -> dict:
+        return self.add_stage("$group", stage)
+
+    def add_fields(self, stage: dict) -> dict:
+        return self.add_stage("$addFields", stage)
+
+    def sort_stage(self, stage: dict) -> dict:
+        return self.add_stage("$sort", stage)