Esercizi Transazioni MySQL Python

Codegrind Team•Jul 10 2024

Esercizi sulla gestione delle transazioni in un database MySQL utilizzando le librerie Python mysql-connector-python e pymysql. Gli esercizi includono sia approcci sequenziali che orientati agli oggetti (OOP), e si concentrano esclusivamente sulle operazioni di transazione.

Esercizio 1: Gestione Transazioni con mysql-connector-python (Sequenziale)

Gestire una transazione semplice che include l'inserimento di dati utilizzando mysql-connector-python.

import mysql.connector

def manage_transaction():
    connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)
    cursor = connection.cursor()
    try:
        cursor.execute("INSERT INTO accounts (account_name, balance) VALUES ('New Account', 500)")
        cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE account_name = 'Existing Account'")
        connection.commit()
        print("Transazione completata con successo")
    except mysql.connector.Error as error:
        print("Errore nella transazione:", error)
        connection.rollback()
    finally:
        cursor.close()
        connection.close()

manage_transaction()

Esercizio 2: Transazione Multi-Operazione con PyMySQL (OOP)

Gestire una transazione che include più operazioni di aggiornamento utilizzando PyMySQL in modalità OOP.

import pymysql

class Database:
    def __init__(self):
        self.connection = pymysql.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)

    def manage_multiple_updates(self):
        try:
            with self.connection.cursor() as cursor:
                cursor.execute("UPDATE accounts SET balance = balance - 200 WHERE account_name = 'Account A'")
                cursor.execute("UPDATE accounts SET balance = balance + 200 WHERE account_name = 'Account B'")
                self.connection.commit()
                print("Transazione completata con successo")
        except pymysql.Error as error:
            print("Errore nella transazione:", error)
            self.connection.rollback()

    def close(self):
        self.connection.close()

db = Database()
db.manage_multiple_updates()
db.close()

Esercizio 3: Rollback Transazione con mysql-connector-python (Sequenziale)

Simulare un errore e gestire il rollback di una transazione utilizzando mysql-connector-python.

import mysql.connector

def transaction_with_rollback():
    connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)
    cursor = connection.cursor()
    try:
        cursor.execute("INSERT INTO orders (order_date, customer_id) VALUES (NOW(), 1)")
        cursor.execute("UPDATE orders SET order_date = NULL WHERE customer_id = 1")  # This will cause an error intentionally
        connection.commit()
    except mysql.connector.Error as error:
        print("Errore rilevato, rollback:", error)
        connection.rollback()
    finally:
        cursor.close()
        connection.close()

transaction_with_rollback()

Esercizio 4: Transazione Avanzata con PyMySQL (OOP)

Gestire una transazione complessa che include insert e update con controllo di errori in PyMySQL.

import pymysql

class AdvancedTransaction:
    def __init__(self):
        self.connection = pymysql.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)

    def execute_transaction(self):
        try:
            with self.connection.cursor() as cursor:
                cursor.execute("INSERT INTO products (product_name, quantity) VALUES ('New Product', 10)")
                cursor.execute("UPDATE products SET quantity = quantity + 10 WHERE product_id = 1")
                self.connection.commit()
                print("Transazione eseguita con successo")
        except pymysql.Error as error:
            print("Errore nella transazione, effettuato rollback:", error)
            self.connection.rollback()

    def close(self):
        self.connection.close()

db = AdvancedTransaction()
db.execute_transaction()
db.close()

Esercizio 5: Gestione Errori in Transazione con mysql-connector-python (Sequenziale)

Scrivere un script che gestisce un errore di transazione causato da un vincolo di chiave esterna utilizzando mysql-connector-python.

import mysql.connector

def transaction_error_management():
    connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)
    cursor = connection.cursor()
    try:
        cursor.execute("INSERT INTO payments (customer_id, amount) VALUES (9999, 100)")  # Non-existent customer_id
        connection.commit()
    except mysql.connector.Error as error:
        print("Transazione fallita, rollback eseguito:", error)
        connection.rollback()
    finally:
        cursor.close()
        connection.close()

transaction_error_management()

Esercizio 6: Transazione con Controllo di Concorrenza in PyMySQL (OOP)

Implementare un controllo di concorrenza ottimistica durante una transazione utilizzando PyMySQL.

import pymysql

class ConcurrencyControlDatabase:
    def __init__(self):
        self.connection = pymysql.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)

    def update_with_concurrency_control(self):
        try:
            with self.connection.cursor() as cursor:
                # Assume 'version' is a column to control concurrency
                cursor.execute("SELECT version FROM accounts WHERE account_id = 1 FOR UPDATE")
                version = cursor.fetchone()
                cursor.execute("UPDATE accounts SET balance = balance + 100, version = %s WHERE account_id = 1 AND version = %s", (version[0] + 1, version[0]))
                self.connection.commit()
                print("Update completato con controllo di concorrenza")
        except pymysql.Error as error:
            print("Errore durante l'update:", error)
            self.connection.rollback()

    def close(self):
        self.connection.close()

db = ConcurrencyControlDatabase()
db.update_with_concurrency_control()
db.close()

Esercizio 7: Transazione Multi-Tabella con mysql-connector-python (OOP)

Gestire una transazione che coinvolge più tabelle utilizzando mysql-connector-python in modalità OOP.

import mysql.connector

class MultiTableTransaction:
    def __init__(self):
        self.connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)

    def perform_transaction(self):
        cursor = self.connection.cursor()
        try:
            cursor.execute("UPDATE accounts SET balance = balance - 100 WHERE account_id = 1")
            cursor.execute("INSERT INTO transactions (account_id, amount) VALUES (1, -100)")
            self.connection.commit()
            print("Transazione su più tabelle completata con successo")
        except mysql.connector.Error as error:
            print("Errore nella transazione:", error)
            self.connection.rollback()
        finally:
            cursor.close()

    def close(self):
        self.connection.close()

db = MultiTableTransaction()
db.perform_transaction()
db.close()

Esercizio 8: Isolamento Transazione con PyMySQL (Sequenziale)

Configurare il livello di isolamento della transazione e gestire una semplice transazione con PyMySQL.

import pymysql

def setup_transaction

_isolation():
    connection = pymysql.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)
    cursor = connection.cursor()
    cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ")
    try:
        cursor.execute("UPDATE accounts SET balance = balance + 100 WHERE account_id = 1")
        connection.commit()
        print("Transazione eseguita con successo con isolamento ripetibile")
    except pymysql.Error as error:
        print("Errore durante la transazione:", error)
        connection.rollback()
    finally:
        cursor.close()
        connection.close()

setup_transaction_isolation()

Esercizio 9: Rollback Complesso con mysql-connector-python (OOP)

Implementare una transazione complessa con multiple operazioni e gestire il rollback in caso di errore con mysql-connector-python.

import mysql.connector

class ComplexRollbackTransaction:
    def __init__(self):
        self.connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)

    def execute_complex_transaction(self):
        cursor = self.connection.cursor()
        try:
            cursor.execute("DELETE FROM orders WHERE order_id = 10")
            cursor.execute("UPDATE inventory SET quantity = quantity - 10 WHERE product_id = 1")  # This might cause an error if not enough inventory
            self.connection.commit()
            print("Transazione complessa eseguita con successo")
        except mysql.connector.Error as error:
            print("Errore rilevato, rollback eseguito:", error)
            self.connection.rollback()
        finally:
            cursor.close()

    def close(self):
        self.connection.close()

db = ComplexRollbackTransaction()
db.execute_complex_transaction()
db.close()

Esercizio 10: Simulazione di Fallimento Transazionale con PyMySQL (OOP)

Simulare un fallimento transazionale e gestire il rollback appropriato con PyMySQL in modalità OOP.

import pymysql

class TransactionFailureSimulation:
    def __init__(self):
        self.connection = pymysql.connect(host='localhost', user='user', password='password', database='testdb', autocommit=False)

    def fail_transaction(self):
        try:
            with self.connection.cursor() as cursor:
                cursor.execute("INSERT INTO payments (customer_id, amount) VALUES (1, 100)")
                cursor.execute("UPDATE customers SET credit = credit - 100 WHERE customer_id = 1")  # Fail if credit goes below zero
                self.connection.commit()
                print("Transazione dovrebbe fallire qui")
        except pymysql.Error as error:
            print("Transazione fallita, rollback eseguito:", error)
            self.connection.rollback()

    def close(self):
        self.connection.close()

db = TransactionFailureSimulation()
db.fail_transaction()
db.close()