firstcommit

.env

+POSTGRESQL_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/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.10 (postgre_crud)" 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/postgre_crud.iml" filepath="$PROJECT_DIR$/.idea/postgre_crud.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/postgre_crud.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

conf/application.conf

+[SERVICE]
+port=7999
+host=127.0.0.1
+

main.py

+from dotenv import load_dotenv
+load_dotenv()
+
+
+
+import uvicorn
+from fastapi import FastAPI
+from dotenv import load_dotenv
+from scripts.service.pos_agg import router
+
+load_dotenv()
+
+app = FastAPI(title="Student Management System")
+app.include_router(router)
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host='localhost', port=9000)

scripts/config/app_config.py

+from configparser import ConfigParser
+import os
+
+config = ConfigParser()
+config.read('conf/application.conf')
+
+
+class Service:
+    port = int(config.getint("SERVICE", "port"))
+    host = config.get("SERVICE", "host")
+
+
+class Psql:
+    postgresql: str = os.environ.get("POSTGRESQL_URI")
+

scripts/constants/__init__.py

+class APIEndpoints:
+    delete = "/delete"
+    student_base = "/student"
+    insert = "/insert"
+    update = "/update"
+    find = "/find"
+

scripts/core/db/database.py

+import sqlalchemy
+from sqlalchemy import create_engine, MetaData
+from sqlalchemy.orm import sessionmaker
+from scripts.config.app_config import Psql
+
+engine = create_engine(Psql.postgresql)
+LocalSession = sessionmaker(bind=engine)
+session = LocalSession()
+Base = sqlalchemy.orm.declarative_base()
+metadata = MetaData()

scripts/core/errors/__init__.py

+class ErrorMessage(Exception):
+    ERRORS_INSERT = "Failed to Insert"
+    ERRORS_FETCH = "Failed to Fetch"
+    ERRORS_UPDATE = "Failed to Update"
+    ERRORS_DELETE  = "Failed to Delete"

scripts/core/handler/pos_agg.py

+from sqlalchemy import Column, Integer, String, MetaData, text
+from sqlalchemy import Table
+
+from scripts.core.db.database import session
+from scripts.core.schema.pos_agg import student
+
+metadata = MetaData()
+
+
+class PostgreSQL:
+    @staticmethod
+    def create_table():
+        try:
+            student_table = Table('student_data', metadata,
+                                  Column('student_id', Integer, primary_key=True, index=True),
+                                  Column('student_name', String),
+                                  Column('description', String),
+                                  Column('address', String)
+                                  )
+            return student_table
+        except Exception as e:
+            print(e, "Failed to create table")
+
+    @staticmethod
+    def insert_data(request_data):
+        try:
+            new_student_data = student(student_name=request_data.student_name, description=request_data.description,
+                                       address=request_data.address)
+            session.add(new_student_data)
+            session.commit()
+            return {"student_id": new_student_data.student_id, "student_name": new_student_data.student_name,
+                    "description": new_student_data.desciption,
+                    "address": new_student_data.address}
+        except Exception as e:
+            print(e, "Failed to Insert data")
+
+    @staticmethod
+    def delete_data(student_id):
+        try:
+            data = session.query(student).filter_by(stuent_id=student_id).first()
+            session.delete(data)
+            session.commit()
+            return data
+        except Exception as e:
+            print(e, "Failed to delete data")
+
+    @staticmethod
+    def update_data(request_data, student_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(student).filter(
+                text(str(f"student_id={student_id}"))
+            ).update(
+                get_update
+            )
+            session.commit()
+            print(update_data)
+            return update_data
+        except Exception as e:
+            print(e, "failed to update the data")
+
+    @staticmethod
+    def fetch_data():
+        try:
+            student_data = session.query(student).all()
+            list_ = []
+            for data in student_data:
+                list_.append(
+                    {"student_id": data.student_id, "student_name": data.student_name, "desciption": data.description,
+                     "address": data.address})
+            return list_
+        except Exception as e:
+            print(e, "Failed to Find the data")

scripts/core/schema/pos_agg.py

+from typing import Optional
+
+from pydantic import BaseModel
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.ext.declarative import declarative_base
+
+from scripts.core.db.database import engine
+
+Base = declarative_base()
+
+
+# define a model for Item
+class student(Base):
+    __tablename__ = 'Students_Records'
+    student_id = Column('student_id', Integer, primary_key=True)
+    student_name = Column('student_name', String, index=True)
+    description = Column('description', String)
+    address = Column('address', String)
+
+Base.metadata.create_all(engine)
+
+
+class studentData(BaseModel):
+    student_name: str
+    description: str
+    address: int
+
+
+class studentDetails(BaseModel):
+    student_name: Optional[str] = None
+    description: Optional[str] = None
+    address: Optional[int] = None

scripts/core/schema/responses.py

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

scripts/service/pos_agg.py

+import logging
+
+from fastapi import APIRouter
+
+from scripts.constants import APIEndpoints
+from scripts.core.errors import ErrorMessage
+from scripts.core.handler.pos_agg import PostgreSQL
+from scripts.core.schema.pos_agg import studentData, studentDetails
+from scripts.core.schema.responses import DefaultResponse
+
+router = APIRouter(prefix=APIEndpoints.student_base)
+handler = PostgreSQL()
+
+
+@router.post(APIEndpoints.insert)
+async def insert_student(request_data: studentData):
+    try:
+        data = handler.insert_data(request_data)
+        return DefaultResponse(message="Successfully inserted", status="success", data=data)
+    except ValueError:
+        return DefaultResponse(message=ErrorMessage.ERRORS_INSERT)
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Failed to Insert data")
+
+
+@router.post(APIEndpoints.find)
+async def fetch_student():
+    try:
+        data = handler.fetch_data()
+        return DefaultResponse(message="Successfully Found", status="success", data=data)
+    except ValueError:
+        return DefaultResponse(message=ErrorMessage.ERRORS_FETCH)
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Failed to Find the data")
+
+
+@router.post(APIEndpoints.update)
+async def update_student(student_id: int, request_data: studentDetails):
+    try:
+        handler.update_data(request_data,student_id)
+        return DefaultResponse(message="Successfully Updated", status="success")
+    except ValueError:
+        return DefaultResponse(message=ErrorMessage.ERRORS_UPDATE)
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Failed to Update the data")
+
+
+@router.post(APIEndpoints.delete)
+async def delete_student(student_id: int):
+    try:
+        handler.delete_data(student_id)
+        return DefaultResponse(message="Successfully deleted", status="success")
+    except ValueError:
+        return DefaultResponse(message=ErrorMessage.ERRORS_DELETE)
+    except Exception as e:
+        logging.exception(e)
+        return DefaultResponse(message="Failed to Delete the data")