first

.env

+POSTGRES_URI =postgresql://interns:interns%40123@192.168.0.220:5432/internsb2
\ No newline at end of file

.idea/.gitignore

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

.idea/.name

+main.py
\ 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="C:\Users\ramya.r\Anaconda3" 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/postgresql.iml" filepath="$PROJECT_DIR$/.idea/postgresql.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/postgresql.iml

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

.idea/sonarlint/issuestore/index.pb

+
+A
+posgresql_file.py,6\6\667b853e3f9ccfca10dbccfd60360189bb0c13c0
\ No newline at end of file

.idea/sonarlint/securityhotspotstore/index.pb

+
+A
+posgresql_file.py,6\6\667b853e3f9ccfca10dbccfd60360189bb0c13c0
\ No newline at end of file

conf/application.conf

+[postgres]
+
+user="interns"
+password="interns@123"
+port=5432
+host=192.168.0.220
+database="internsb2"
+
+

main.py

+import uvicorn
+
+from fastapi import FastAPI
+from dotenv import load_dotenv
+
+from script.service.postgres import router
+
+load_dotenv()
+app = FastAPI(title="Employee")
+app.include_router(router)
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host='localhost', port=9000)

script/config/app_config.py

+import os
+from configparser import SafeConfigParser
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+config = SafeConfigParser()
+config.read('conf/application.conf')
+
+
+class Postgres:
+    port = int(config.getint("postgres", "port"))
+    host = config.get("postgres", "host")
+
+
+class Postgresql:
+    postgres_uri: str = os.environ.get("POSTGRES_URI")
+
+
+
+
+
+
+
+
+

script/constants/__init__.py

+class APIEndpoints:
+    delete = "/delete"
+    save = "/save"
+    employee_base = "/employee"
+    insert = "/insert"
+    update = "/update"
+    find = "/find"
\ No newline at end of file

script/core/errors/__init__.py

+class NameDoesNotExist(Exception):
+    pass
\ No newline at end of file

script/core/handlers/postgresql.py

+from sqlalchemy import MetaData
+from sqlalchemy import Column, Integer, String, MetaData, text
+from script.core.schema.postgresql import Employee, Model
+from script.utils.postgresql_connection import session
+
+
+
+class EmployeeData:
+    @staticmethod
+    def insert_data(request_data):
+        try:
+            employee_data = Model(emp_id=request_data.emp_id, emp_name=request_data.emp_name,
+                                  emp_department=request_data.emp_department,
+                                  emp_sal=request_data.emp_sal)
+            session.add(employee_data)
+            session.commit()
+            return employee_data
+        except Exception as e:
+            print(e, "Error Detected in inserting")
+
+    @staticmethod
+    def find_data(emp_id):
+        try:
+            employee_data= session.query(Model).all()
+            list_ = []
+            for data in employee_data:
+                list_.append({"emp_id": data.emp_id, "emp_name": data.emp_name,
+                              "emp_department": data.emp_department,
+                              "emp_sal": data.emp_sal})
+            return list_
+            return employee_data
+        except Exception as e:
+            print(e, "Error Detected in finding employee data")
+
+    @staticmethod
+    def update_data(request_data, emp_id):
+        try:
+            get_update = {key: value for key, value in request_data if
+                          value is not None and value != 'string' and value != 0}
+            update_data = session.query(Model).filter(
+                text(str(f"emp_id={emp_id}"))
+            ).update(
+                get_update
+            )
+            session.commit()
+            print(update_data)
+            return update_data
+        except Exception as e:
+            print(e, "Error detected in updating")
+
+    @staticmethod
+    def delete_data(emp_id):
+        try:
+            data = session.query(Model).filter_by(emp_id=emp_id).first()
+            session.delete(data)
+            session.commit()
+            return data
+        except Exception as e:
+            print(e, "Error detected in deleting")
+
+
+
+
+

script/core/schema/postgresql.py

+from typing import Optional
+
+from pydantic import BaseModel
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.orm import declarative_base
+
+from script.utils.postgresql_connection import engine
+
+
+# define a model for Item
+class Employee(BaseModel):
+    emp_id: int
+    emp_name: Optional[str]
+    emp_department: Optional[str]
+    emp_sal: Optional[int]
+
+
+Base = declarative_base()
+
+
+class Model(Base):
+    __tablename__ = 'employee'
+    emp_id = Column('emp_id', Integer, primary_key=True, index=True)
+    emp_name = Column('emp_name', String(30))
+    emp_department = Column('emp_dept', String(100))
+    emp_sal = Column('salary', Integer)
+
+Base.metadata.create_all(engine)
\ No newline at end of file

script/core/schema/response.py

+from typing import Any, Optional
+
+from pydantic import BaseModel
+
+
+class DefaultResponse(BaseModel):
+    status: str = "failed"
+    message: str
+    data: Optional[Any]
+

script/service/postgres.py

+import logging
+from fastapi import APIRouter
+from script.constants import APIEndpoints
+from script.core.handlers.postgresql import EmployeeData
+from script.core.schema.postgresql import Employee
+from script.core.schema.response import DefaultResponse
+
+router = APIRouter(prefix=APIEndpoints.employee_base)
+handler = EmployeeData()
+
+
+@router.delete(APIEndpoints.delete)
+async def delete_item(emp_id: int):
+
+    try:
+        handler.delete_data(emp_id)
+        return DefaultResponse(message="Successfully deleted", status="success")
+    except ValueError:
+        return DefaultResponse(message="Due to Value Error")
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Deletion failed due to server error")
+
+
+@router.post(APIEndpoints.update)
+async def update_item(emp_id: int, request_data: Employee):
+    try:
+        handler.update_data( request_data, emp_id)
+        return DefaultResponse(message="Successfully Updated", status="success")
+    except ValueError:
+        return DefaultResponse(message="Due to value error")
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Updation Failed due to server error")
+
+
+@router.get(APIEndpoints.find)
+async def find_item(emp_id:int):
+    try:
+        data = handler.find_data(emp_id)
+        return DefaultResponse(message="Successfully Found", status="success", data=data)
+    except ValueError:
+        return DefaultResponse(message="Due to value error")
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Finding Failed due to server error")
+
+
+@router.post(APIEndpoints.insert)
+def insert_item(request_data: Employee):
+    try:
+        op = handler.insert_data(request_data)
+        return DefaultResponse(message="Successfully Found", status="success", op=op)
+    except ValueError:
+        return DefaultResponse(message="Due to value error")
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Finding Failed due to server error")

script/utils/postgresql_connection.py

+import sqlalchemy
+from sqlalchemy import create_engine, MetaData
+from sqlalchemy.orm import sessionmaker
+
+from script.config.app_config import Postgresql
+
+engine = create_engine(Postgresql.postgres_uri)
+LocalSession = sessionmaker(bind=engine)
+session = LocalSession()
+Base = sqlalchemy.orm.declarative_base()
+Base.metadata.create_all(engine)
+metadata = MetaData()