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()
            self.cursor.close()

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

        try:
            cur.execute(
                "SELECT id FROM mat_credit WHERE financial_year=%s",
                (data["financial_year"],)
            )
            row = cur.fetchone()

            if row:
                mat_id = row[0]

                cur.execute("""
                    UPDATE mat_credit
                    SET mat_credit=%s, balance=%s
                    WHERE id=%s
                """, (data["mat_credit"], data["balance"], mat_id))

                cur.execute(
                    "DELETE FROM mat_utilization WHERE mat_credit_id=%s",
                    (mat_id,)
                )
            else:
                cur.execute("""
                    INSERT INTO mat_credit (financial_year, mat_credit, balance)
                    VALUES (%s,%s,%s)
                """, (data["financial_year"], data["mat_credit"], data["balance"]))

                mat_id = cur.lastrowid

            for u in data["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()

        except Exception:
            conn.rollback()
            raise

        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()