Asset _Manager_V1.0 - Not Reviewed

.env

+MONGO_URI=mongodb://localhost:27017
+DB_NAME=userDB
+SERVICE_HOST=0.0.0.0
+SERVICE_PORT=8671

.gitignore

+.idea
\ No newline at end of file

README.md

-# Asset Manager Metaservices
\ No newline at end of file

__version__.py

+__version__ = "1.0"

app.py

+# iot manager - Arun Uday
+import uvicorn
+
+from scripts.config import PROJECT_NAME, Services
+from scripts.logging.logger import logger
+
+# starting the application
+if __name__ == "__main__":
+    try:
+        print("Api for " + PROJECT_NAME)
+        uvicorn.run("main:app", port=int(Services.PORT))
+    except Exception as e:
+        logger.error(e)
+

conf/application.conf

+[project_name]
+PROJECT_NAME = AssetManager

main.py

+"""
+
+Author: Arun Uday
+Email: arun.uday@knowledgelens.com
+
+Asset Manager Login For Normal User
+---------------------------------------------------------
+For instructions on how to run, check README.md
+
+
+"""
+if __name__ == "__main__":
+    from dotenv import load_dotenv
+
+    load_dotenv()
+
+import uvicorn
+from scripts.services import router
+from scripts.config import PROJECT_NAME, Services as ServiceConf
+from fastapi.middleware.cors import CORSMiddleware
+from scripts.logging.logger import logger
+
+# starting the application
+if __name__ == "__main__":
+    try:
+        print("Api for " + PROJECT_NAME)
+        if ServiceConf.ENABLE_CORS:
+            router.add_middleware(
+                CORSMiddleware,
+                allow_origins=ServiceConf.CORS_URLS,
+                allow_credentials=ServiceConf.CORS_ALLOW_CREDENTIALS,
+                allow_methods=ServiceConf.CORS_ALLOW_METHODS,
+                allow_headers=ServiceConf.CORS_ALLOW_HEADERS,
+            )
+        uvicorn.run(router, port=int(ServiceConf.PORT))
+    except Exception as e:
+        logger.error(e)
+
+

requirements.txt

+uvicorn~=0.21.0
+python-dotenv~=1.0.0
+pydantic~=1.10.6
+fastapi~=0.94.1
+passlib~=1.7.4
+pymongo~=4.3.3
+bcrypt~=4.0.1
\ No newline at end of file

scripts/config/__init__.py

+import configparser
+from typing import Optional, Literal
+
+from pydantic import BaseSettings, Field
+
+config = configparser.RawConfigParser()
+config.read("conf/application.conf")
+PROJECT_NAME = config.get("project_name", "PROJECT_NAME")
+
+
+class _Services(BaseSettings):
+    HOST: str = Field(default="127.0.0.1", env="service_host")
+    PORT: int = Field(default=8000, env="service_port")
+    ENABLE_CORS: bool = True
+    CORS_URLS: list[str] = ["*.ilens.io"]
+    CORS_ALLOW_CREDENTIALS: bool = True
+    CORS_ALLOW_METHODS: list[str] = ["GET", "POST", "DELETE", "PUT"]
+    CORS_ALLOW_HEADERS: list[str] = ["*"]
+    LOG_LEVEL: Literal["INFO", "DEBUG", "ERROR", "QTRACE"] = "INFO"
+    ENABLE_FILE_LOGGING: bool = False
+    SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
+    ALGORITHM = "HS256"
+    ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+
+class _Databases(BaseSettings):
+    MONGO_URI: str
+    DB_NAME: str
+
+
+Services = _Services()
+Databases = _Databases()
+
+__all__ = [
+    "PROJECT_NAME",
+    "Services",
+    "Databases",
+]

scripts/constants/__init__.py

+from pydantic import BaseSettings, Field
+
+PROJECT_NAME = "IotManager"
+
+
+class _Services(BaseSettings):
+    HOST: str = Field(default="127.0.0.1", env="service_host")
+    PORT: int = Field(default=8000, env="service_port")
+
+
+class _Databases(BaseSettings):
+    MONGO_URI: str
+
+
+Services = _Services()
+Databases = _Databases()
+
+__all__ = [
+    "PROJECT_NAME",
+    "Services",
+    "Databases",
+]

scripts/constants/api.py

+class ApiEndPoints:
+    # version
+    version = "/v1"
+
+    #  common
+    asset_manager_submit: str = "/submit"
+
+    # login-management
+    asset_manager_login: str = "/login"

scripts/constants/db_constants.py

+class DatabaseConstants:
+    collection_user_details = "UserDetails"
+    collection_user_roles = "UserRoles"
+    collection_user_devices = "UserDevices"

scripts/core/handlers/login_handler.py

+from __future__ import annotations
+
+from passlib.context import CryptContext
+
+from scripts.database.mongo.mongo_login import MongoUser
+from scripts.errors import ErrorMessages
+
+
+class LoginHandlers:
+    def __init__(self):
+        self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+        self.db_user_data = None
+
+    @staticmethod
+    def user_data_validation(login_data) -> dict | None:
+        if login_data.username == "" or login_data.username == "user@example.com":
+            return {"message": ErrorMessages.ERROR_INVALID_USERNAME_PASSWORD, "data": login_data.username}
+        if login_data.password == "" or login_data.password == "string":
+            return {"message": ErrorMessages.ERROR_INVALID_USERNAME_PASSWORD, "data": login_data.password}
+        return None
+
+    def db_data_validation(self, login_data):
+        self.db_user_data = MongoUser().fetch_user_details(login_data.username)
+        if not self.db_user_data:
+            return {"message": ErrorMessages.ERROR_USER_NOT_REGISTERED, "data": login_data.username}
+        return None
+
+    def db_password_matching(self, login_data):
+        if response := self.db_data_validation(login_data):
+            return response
+        if not self.pwd_context.verify(login_data.password, self.db_user_data["password"]):
+            return {"message": ErrorMessages.ERROR_PASSWORD_MISMATCH, "data": {login_data.username,
+                                                                               login_data.password}}
+        return None
+

scripts/database/mongo/__init__.py

+from scripts.config import Databases
+from scripts.utils.mongo_utils import MongoConnect
+
+mongo_obj = MongoConnect(uri=Databases.MONGO_URI)
+
+mongo_client = mongo_obj()
+
+CollectionBaseClass = mongo_obj.get_base_class()

scripts/database/mongo/mongo_login.py

+from scripts.config import Databases
+from scripts.constants.db_constants import DatabaseConstants
+from scripts.database.mongo import CollectionBaseClass, mongo_client
+
+collection_name = DatabaseConstants.collection_user_details
+
+
+class UserDetailsKeys:
+    KEY_NAME = "name"
+    KEY_EMAIL = "email"
+    KEY_PASSWORD = "password"
+    KEY_USER_ROLE = "user_role"
+    KEY_IS_ALIVE = "is_alive"
+    KEY_CREATED_AT = "created_at"
+    KEY_UPDATED_AT = "updated_at"
+
+
+class MongoUser(CollectionBaseClass):
+    key_name = UserDetailsKeys.KEY_NAME
+    key_email = UserDetailsKeys.KEY_EMAIL
+    key_password = UserDetailsKeys.KEY_PASSWORD
+    key_user_role = UserDetailsKeys.KEY_USER_ROLE
+    key_is_alive = UserDetailsKeys.KEY_IS_ALIVE
+    key_created_at = UserDetailsKeys.KEY_CREATED_AT
+    key_updated_at = UserDetailsKeys.KEY_UPDATED_AT
+
+    def __init__(self):
+        super().__init__(mongo_client, Databases.DB_NAME, collection_name)
+
+    def fetch_user_details(self, email):
+        if user := self.find_one(query={self.key_email: email}):
+            return user
+        return None

scripts/errors/__init__.py

+class ErrorMessages:
+    UNKNOWN_ERROR = "Unknown Error Occurred"
+    OP_FAILED = "Operation failed"
+    # Authorization Errors
+    ERROR_AUTH_FAILED = "Authentication Failed. Please verify token"
+    ERROR_INVALID_USERNAME_PASSWORD = "Invalid Username or Password."
+    ERROR_USER_NOT_REGISTERED = "Account is not registered in the portal."
+    ERROR_PASSWORD_MISMATCH = "Passwords Authentication Failed. Please enter the correct password"

scripts/logging/logger.py

+import logging
+import pathlib
+from logging import StreamHandler
+from logging.handlers import RotatingFileHandler, SocketHandler
+
+from scripts.config import PROJECT_NAME, Services
+
+
+def read_configuration():
+    return {
+        "name": PROJECT_NAME,
+        "handlers": [
+            {"type": "RotatingFileHandler", "max_bytes": 100000000, "back_up_count": 5},
+            {"type": "StreamHandler", "name": PROJECT_NAME},
+        ],
+    }
+
+
+def init_logger():
+    logging_config = read_configuration()
+
+    """
+    Creates a rotating log
+    """
+    __logger__ = logging.getLogger(PROJECT_NAME)
+    __logger__.setLevel(Services.LOG_LEVEL)
+
+    log_formatter = "%(asctime)s - %(levelname)-6s - [%(threadName)5s:%(funcName)5s(): %(lineno)s] - %(message)s"
+    time_format = "%Y-%m-%d %H:%M:%S"
+    formatter = logging.Formatter(log_formatter, time_format)
+    for each_handler in logging_config["handlers"]:
+        if (
+                each_handler["type"] in ["RotatingFileHandler"]
+                and Services.ENABLE_FILE_LOGGING
+        ):
+            pathlib.Path("logs").mkdir(parents=True, exist_ok=True)
+            log_file = pathlib.Path("logs/", f"{PROJECT_NAME}.log")
+            temp_handler = RotatingFileHandler(
+                log_file,
+                maxBytes=each_handler["max_bytes"],
+                backupCount=each_handler["back_up_count"],
+            )
+            temp_handler.setFormatter(formatter)
+        elif each_handler["type"] in ["SocketHandler"]:
+            temp_handler = SocketHandler(each_handler["host"], each_handler["port"])
+        elif each_handler["type"] in ["StreamHandler"]:
+            temp_handler = StreamHandler()
+            temp_handler.setFormatter(formatter)
+
+        else:
+            temp_handler = None
+        __logger__.addHandler(temp_handler)
+    return __logger__
+
+
+logger = init_logger()

scripts/schemas/collection_schema.py

+from pydantic import BaseModel
+from typing import Union
+
+
+class UserDetails(BaseModel):
+    name: str
+    email: Union[str: None] = None
+    password: Union[str: None] = None
+    user_role: Union[str: None] = None
+    is_alive: Union[bool: None] = None
+    created_at: Union[str: None] = None
+    updated_at: Union[str: None] = None

scripts/schemas/default_responses.py

+from typing import Any, Optional
+
+from pydantic import BaseModel
+
+
+class DefaultResponse(BaseModel):
+    status: bool = True
+    message: Optional[str]
+    payload: Optional[Any]
+
+
+class DefaultFailureResponse(DefaultResponse):
+    status: bool = False
+    error: Any

scripts/schemas/login_schema.py

+from typing import Union
+
+from pydantic import BaseModel, EmailStr
+
+
+class NormalLogin(BaseModel):
+    username: Union[EmailStr, None] = None
+    password: Union[str, None] = None
+    project_id: Union[str, None] = None

scripts/services/__init__.py

+from fastapi import FastAPI
+from scripts.services import v1
+
+router = FastAPI()
+router.include_router(v1.router)

scripts/services/v1/__init__.py

+from fastapi import APIRouter
+from scripts.constants.api import ApiEndPoints
+from scripts.services.v1 import iot_manager_services
+
+router = APIRouter(prefix=ApiEndPoints.asset_manager_login)
+router.include_router(iot_manager_services.router)

scripts/services/v1/iot_manager_services.py

+from fastapi import APIRouter
+
+from scripts.constants.api import ApiEndPoints
+from scripts.core.handlers.login_handler import LoginHandlers
+from scripts.logging.logger import logger
+from scripts.schemas.default_responses import DefaultResponse, DefaultFailureResponse
+from scripts.schemas.login_schema import NormalLogin
+
+router = APIRouter(prefix=ApiEndPoints.version)
+obj_login_handler = LoginHandlers()
+
+
+@router.post(ApiEndPoints.asset_manager_submit)
+def login_default(login_data: NormalLogin):
+    try:
+        response = obj_login_handler.user_data_validation(login_data)
+        if response is not None:
+            return DefaultFailureResponse(error=response["message"])
+        response = obj_login_handler.db_password_matching(login_data)
+        if response is not None:
+            return DefaultFailureResponse(error=response["message"])
+        return DefaultResponse(message="Login Successful", payload={"username": login_data.username})
+    except Exception as e:
+        logger.exception(e)

scripts/utils/mongo_tools/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,
+    ) -> None:
+        self.client = mongo_client
+        self.database = database
+        self.collection = collection
+
+    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 .mongo_tools 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)