Untitled

for index, source_row in source_df.iterrows():
    pk_value = source_row[pk]

    # Check if the primary key exists in the target DataFrame
    if pk_value not in target_df[pk].values:

        # ... (same code as before)

    else:
        # Fetch the corresponding row from the target DataFrame based on the primary key
        target_row = target_df[target_df[pk] == pk_value].iloc[0]

        if not source_row.equals(target_row):
            columns_to_update = []
            for column_name, source_val in source_row.items():
                target_val = target_row[column_name]
                if source_val != target_val:
                    if source_val == 'NaT':
                        columns_to_update.append(f"{column_name} = NULL")
                    elif is_datetime(source_val):
                        if db_type_ext == 'ORACLE':
                            columns_to_update.append(f"{column_name} = TO_DATE('{source_val}', 'YYYY-MM-DD HH24:MI:SS')")
                        elif db_type_ext in ('MYSQL', 'MARIA'):
                            columns_to_update.append(f"{column_name} = STR_TO_DATE('{source_val}', '%Y-%m-%d %H:%i:%s')")
                    else:
                        columns_to_update.append(f"{column_name} = '{source_val}'")

            # Generate an update query dynamically
            if columns_to_update:
                update_query = f"UPDATE {schema_ext}.{table_name} SET "
                update_query += ", ".join(columns_to_update)
                update_query += f" where {eprm_table_col_pk} = '{pk_value}'"

                try:
                    print(update_query)
                    cursor_ext.execute(update_query)
                    update_sucess_count += 1
                    # connection_ext.commit()
                except Exception as e:
                    logging.error(
                        "Error - {} . Line No - {} ".format(str(e), str(sys.exc_info()[-1].tb_lineno)))
                    update_failed_count += 1