131 lines
4.2 KiB
Python
131 lines
4.2 KiB
Python
# model/Subcontractor.py
|
|
|
|
import config
|
|
from model.Log import LogHelper
|
|
from mysql.connector import Error
|
|
|
|
class Subcontractor:
|
|
|
|
@staticmethod
|
|
def get_connection():
|
|
return config.get_db_connection()
|
|
|
|
@staticmethod
|
|
def get_all_subcontractors():
|
|
connection = Subcontractor.get_connection()
|
|
subcontractors = []
|
|
if not connection:
|
|
return None, "Database connection failed"
|
|
try:
|
|
cursor = connection.cursor()
|
|
cursor.callproc('GetAllSubcontractors')
|
|
for result in cursor.stored_results():
|
|
subcontractors = result.fetchall()
|
|
except Error as e:
|
|
print(f"Error fetching subcontractors: {e}")
|
|
return None, str(e)
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
return subcontractors, None
|
|
|
|
@staticmethod
|
|
def get_subcontractor_by_id(id):
|
|
connection = Subcontractor.get_connection()
|
|
subcontractor = None
|
|
if not connection:
|
|
return None, "Database connection failed"
|
|
try:
|
|
cursor = connection.cursor()
|
|
cursor.callproc("GetSubcontractorById", (id,))
|
|
for result in cursor.stored_results():
|
|
subcontractor = result.fetchone()
|
|
except Error as e:
|
|
print(f"Error fetching subcontractor: {e}")
|
|
return None, str(e)
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
return subcontractor, None
|
|
|
|
@staticmethod
|
|
def save_subcontractor(data):
|
|
connection = Subcontractor.get_connection()
|
|
if not connection:
|
|
return "Database connection failed"
|
|
try:
|
|
cursor = connection.cursor()
|
|
cursor.callproc('SaveContractor', (
|
|
data['Contractor_Name'],
|
|
data['Address'],
|
|
data['Mobile_No'],
|
|
data['PAN_No'],
|
|
data['Email'],
|
|
data['Gender'],
|
|
data['GST_Registration_Type'],
|
|
data['GST_No'],
|
|
data['Contractor_password']
|
|
))
|
|
connection.commit()
|
|
# Active log
|
|
LogHelper.log_action("Add Subcontractor", f"Added subcontractor '{data['Contractor_Name']}'")
|
|
except Error as e:
|
|
print(f"Error inserting subcontractor: {e}")
|
|
return str(e)
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
return None
|
|
|
|
@staticmethod
|
|
def update_subcontractor(id, data):
|
|
connection = Subcontractor.get_connection()
|
|
if not connection:
|
|
return "Database connection failed"
|
|
try:
|
|
cursor = connection.cursor()
|
|
cursor.callproc('UpdateSubcontractor', (
|
|
id,
|
|
data['Contractor_Name'],
|
|
data['Address'],
|
|
data['Mobile_No'],
|
|
data['PAN_No'],
|
|
data['Email'],
|
|
data['Gender'],
|
|
data['GST_Registration_Type'],
|
|
data['GST_No'],
|
|
data['Contractor_password']
|
|
))
|
|
connection.commit()
|
|
# Active log
|
|
LogHelper.log_action("Edit Subcontractor", f"Edited subcontractor '{id}'")
|
|
except Error as e:
|
|
print(f"Error updating subcontractor: {e}")
|
|
return str(e)
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
return None
|
|
|
|
@staticmethod
|
|
def delete_subcontractor(id):
|
|
connection = Subcontractor.get_connection()
|
|
if not connection:
|
|
return "Database connection failed", 0
|
|
affected_rows = 0
|
|
try:
|
|
cursor = connection.cursor()
|
|
cursor.callproc('DeleteSubcontractor', (id,))
|
|
connection.commit()
|
|
for result in cursor.stored_results():
|
|
row = result.fetchone()
|
|
affected_rows = row[0] if row else 0
|
|
# Active log
|
|
LogHelper.log_action("Delete Subcontractor", f"Deleted subcontractor '{id}'")
|
|
except Error as e:
|
|
print(f"Error deleting subcontractor: {e}")
|
|
return str(e), 0
|
|
finally:
|
|
cursor.close()
|
|
connection.close()
|
|
return None, affected_rows |