crud operation

.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/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="N801" />
+          <option value="N803" />
+        </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.11 (task-6_PostgreSQL)" 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/task-6_PostgreSQL.iml" filepath="$PROJECT_DIR$/.idea/task-6_PostgreSQL.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/task-6_PostgreSQL.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/vcs.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>
\ No newline at end of file

conf/application.conf

+[POSTGRESQL]
+host = 127.0.0.1
+port =7999
+user = "interns"
+password = "interns@123"
\ No newline at end of file

main.py

+from dotenv import load_dotenv
+from script.config.app_confing import Services
+
+import uvicorn
+from fastapi import FastAPI
+
+from script.service.inventory import router
+load_dotenv()
+
+app = FastAPI(title="Inventor Mt")
+app.include_router(router)
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host=Services.host, port=Services.port)
\ No newline at end of file

script/config/app_confing.py

+from dotenv import load_dotenv
+import os
+from configparser import SafeConfigParser
+
+load_dotenv()
+config = SafeConfigParser()
+config.read('conf/application.conf')
+
+
+class Sql:
+
+    postgresql: str = os.environ.get("POSTGRESQL_URI")
+
+
+class Services:
+    host = config.get("POSTGRESQL", "host")
+    port = int(config.get("POSTGRESQL", "port"))

script/constants/__init__.py

+class APIEndpoints:
+    delete = "/delete"
+    save = "/save"
+    inventory_base = "/inventory"
+    insert = "/insert"
+    update = "/update"
+    find = "/find"
+    date = "/date"
\ No newline at end of file

script/core/db/postgresql_connector.py

+import sqlalchemy
+from sqlalchemy import create_engine, MetaData
+from sqlalchemy.orm import sessionmaker
+
+from script.config.app_confing import Sql
+
+engine = create_engine(Sql.postgresql)
+LocalSession = sessionmaker(bind=engine)
+session = LocalSession()
+Base = sqlalchemy.orm.declarative_base()
+metadata = MetaData()

script/core/errors/__init__.py

+class ErrorMessage(Exception):
+    ERRORS_INSERT = "Due to value error in insertion"
+    ERRORS_FETCH = "Due to value error in fetch"
+    ERRORS_UPDATE = "Due to value error in update"
+    ERRORS_DELETE  = "Due to value error in deletion"
+

script/core/handlers/inventory.py

+from sqlalchemy import Column, Integer, String, MetaData, text
+from sqlalchemy import Table
+
+from script.core.db.postgresql_connector import session
+from script.core.schema.inventory import Inventory
+
+metadata = MetaData()
+
+
+class PostgresqlCollectRecords:
+    @staticmethod
+    def create_table():
+        try:
+            inventory_table = Table('inventory_data', metadata,
+                                    Column('order_id', Integer, primary_key=True, index=True),
+                                    Column('customer_name', String),
+                                    Column('status', String),
+                                    Column('sales_order', Integer, index=True)
+                                    )
+            return inventory_table
+        except Exception as e:
+            print(e, "Error detected in creating the table")
+
+    @staticmethod
+    def insert_data(request_data):
+        try:
+            new_inventory_data = Inventory(customer_name=request_data.customer_name,
+                                           status=request_data.status, sales_order=request_data.sales_order)
+            session.add(new_inventory_data)
+            session.commit()
+            return {"order_id": new_inventory_data.order_id, "customer_name": new_inventory_data.customer_name,
+                    "status": new_inventory_data.status,
+                    "sales_order": new_inventory_data.sales_order}
+        except Exception as e:
+            print(e, "Error detected in inserting")
+
+    @staticmethod
+    def fetch_records():
+        try:
+            inventory_data = session.query(Inventory).all()
+            list_ = []
+            for data in inventory_data:
+                list_.append({"order_id": data.order_id, "customer_name": data.customer_name, "status": data.status,
+                              "sales_order": data.sales_order})
+            return list_
+        except Exception as e:
+            print(e, "Error detected in fetching the data")
+
+    @staticmethod
+    def update_records(request_data, order_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(Inventory).filter(
+                text(str(f"order_id={order_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_records(order_id):
+        try:
+            data = session.query(Inventory).filter_by(order_id=order_id).first()
+            session.delete(data)
+            session.commit()
+            return data
+        except Exception as e:
+            print(e, "Error detected in deleting")

script/core/schema/inventory.py

+from typing import Optional
+
+from pydantic import BaseModel
+from sqlalchemy import Column, Integer, String
+from sqlalchemy.ext.declarative import declarative_base
+
+from script.core.db.postgresql_connector import engine
+
+Base = declarative_base()
+
+
+# define a model for Item
+class Inventory(Base):
+    __tablename__ = 'inventory_data'
+
+    order_id = Column('order_id', Integer, primary_key=True)
+    customer_name = Column('customer_name', String, index=True)
+    status = Column('status', String)
+    sales_order = Column('sales_order', Integer, index=True)
+
+
+Base.metadata.create_all(engine)
+
+
+class InventoryData(BaseModel):
+    customer_name: str
+    status: str
+    sales_order: int
+
+
+class UpdateInventoryData(BaseModel):
+    customer_name: Optional[str] = None
+    status: Optional[str] = None
+    sales_order: Optional[int] = None

script/core/schema/responses.py

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

script/service/inventory.py

+import logging
+
+from fastapi import APIRouter
+
+from script.constants import APIEndpoints
+from script.core.errors import ErrorMessage
+from script.core.handlers.inventory import PostgresqlCollectRecords
+from script.core.schema.inventory import Inventory, InventoryData, UpdateInventoryData
+from script.core.schema.responses import DefaultResponse
+
+router = APIRouter(prefix=APIEndpoints.inventory_base)
+handler = PostgresqlCollectRecords()
+
+
+@router.post(APIEndpoints.insert)
+async def insert_item(request_data: InventoryData):
+    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="inserted Failed due to server error")
+
+
+@router.post(APIEndpoints.find)
+async def fetch_item():
+    try:
+        data = handler.fetch_records()
+        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="Finding Failed due to server error")
+
+
+@router.post(APIEndpoints.update)
+async def update_item(order_id: int, request_data: UpdateInventoryData):
+    try:
+        handler.update_records(request_data, order_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="Updation Failed due to server error")
+
+
+@router.post(APIEndpoints.delete)
+async def delete_item(order_id: int):
+    try:
+        handler.delete_records(order_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="Deletion failed due to server error")