IncomeTaxSystem/AppCode/MatCreditHandler.py

from AppCode.Config import DBConfig
import mysql.connector

class MatCreditHandler:

    def __init__(self):
        # db = DBConfig()
        self.conn = DBConfig.get_db_connection()
        self.cursor = self.conn.cursor(dictionary=True)

    # get all Mat credit data
    def fetch_all(self):
        try:
           self.cursor.callproc("GetMatCedit")
           result_sets = self.cursor.stored_results()
           mat_rows = next(result_sets).fetchall()
           utilization_rows = next(result_sets).fetchall()

           return mat_rows, utilization_rows
        finally:
            self.cursor.close()

    # Save Mat credit data single row
    @staticmethod
    def save_single(data):
        conn = DBConfig.get_db_connection()
        cur = conn.cursor(dictionary=True)
        try:

            cur.callproc("SaveOrUpdateMatCredit",(
            data["financial_year"],
            data["mat_credit"],
            data["balance"]
            ))

            result = next(cur.stored_results()).fetchone()
            mat_id = result["mat_id"]

            if not mat_id:
                raise Exception("mat_id not returned from procedure")

            for u in data.get("utilization", []):
                cur.callproc(
                    "InsertMatUtilization",
                    (mat_id, u["year"], u["amount"])
                )

            conn.commit()

        except Exception as e:
            conn.rollback()
            raise e

        finally:
            cur.close()
            conn.close()
    
    # save all Mat credit data 
    # @staticmethod
    # def save_bulk(rows):
    #     conn = DBConfig.get_db_connection()
    #     cur = conn.cursor()
    #     skipped = []

    #     try:
    #         for row in rows:
    #             cur.execute(
    #                 "SELECT id FROM mat_credit WHERE financial_year=%s",
    #                 (row["financial_year"],)
    #             )
    #             if cur.fetchone():
    #                 skipped.append(row["financial_year"])
    #                 continue

    #             cur.execute("""
    #                 INSERT INTO mat_credit (financial_year, mat_credit, balance)
    #                 VALUES (%s,%s,%s)
    #             """, (row["financial_year"], row["mat_credit"], row["balance"]))

    #             mat_id = cur.lastrowid

    #             for u in row["utilization"]:
    #                 cur.execute("""
    #                     INSERT INTO mat_utilization
    #                     (mat_credit_id, utilized_year, utilized_amount)
    #                     VALUES (%s,%s,%s)
    #                 """, (mat_id, u["year"], u["amount"]))

    #         conn.commit()
    #         return skipped

    #     except Exception:
    #         conn.rollback()
    #         raise

    #     finally:
    #         cur.close()
    #         conn.close()

    # CLOSE CONNECTION
    def close(self):
        self.cursor.close()
        self.conn.close()