Esercizi Query Avanzate MySQL Python

Edoardo Midali

Esercizi su query avanzate in MySQL. Questi esercizi coprono vari aspetti delle query avanzate, inclusi join complessi, subquery, CTE (Common Table Expressions), e funzioni di aggregazione avanzate. Gli esercizi utilizzano sia mysql-connector-python che pymysql e includono approcci sequenziali e orientati agli oggetti (OOP).

Esercizio 1: Join Complesso con mysql-connector-python (Sequenziale)

Eseguire un join complesso tra tre tabelle utilizzando mysql-connector-python.

import mysql.connector

def complex_join():
    connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb')
    cursor = connection.cursor()
    query = """
    SELECT o.order_id, c.customer_name, p.product_name
    FROM orders o
    JOIN customers c ON o.customer_id = c.customer_id
    JOIN products p ON o.product_id = p.product_id
    """
    cursor.execute(query)
    results = cursor.fetchall()
    for row in results:
        print(row)
    cursor.close()
    connection.close()

complex_join()

Esercizio 2: Subquery con PyMySQL (OOP)

Utilizzare una subquery per recuperare dati specifici con PyMySQL in modalità OOP.

import pymysql

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

    def subquery_example(self):
        try:
            with self.connection.cursor() as cursor:
                query = """
                SELECT c.customer_name, c.customer_id
                FROM customers c
                WHERE c.customer_id IN (SELECT o.customer_id FROM orders o WHERE o.total_amount > 100)
                """
                cursor.execute(query)
                results = cursor.fetchall()
                for row in results:
                    print(row)
        except pymysql.Error as error:
            print("Errore nella subquery:", error)

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

db = Database()
db.subquery_example()
db.close()

Esercizio 3: CTE con mysql-connector-python (Sequenziale)

Utilizzare un CTE (Common Table Expression) per semplificare una query complessa con mysql-connector-python.

import mysql.connector

def use_cte():
    connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb')
    cursor = connection.cursor()
    query = """
    WITH OrderCTE AS (
        SELECT customer_id, SUM(total_amount) AS total_spent
        FROM orders
        GROUP BY customer_id
    )
    SELECT c.customer_name, o.total_spent
    FROM customers c
    JOIN OrderCTE o ON c.customer_id = o.customer_id
    """
    cursor.execute(query)
    results = cursor.fetchall()
    for row in results:
        print(row)
    cursor.close()
    connection.close()

use_cte()

Esercizio 4: Funzione di Aggregazione Avanzata con PyMySQL (OOP)

Utilizzare funzioni di aggregazione avanzate come GROUP_CONCAT con PyMySQL in modalità OOP.

import pymysql

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

    def advanced_aggregation(self):
        try:
            with self.connection.cursor() as cursor:
                query = """
                SELECT c.customer_name, GROUP_CONCAT(p.product_name SEPARATOR ', ') AS products
                FROM customers c
                JOIN orders o ON c.customer_id = o.customer_id
                JOIN products p ON o.product_id = p.product_id
                GROUP BY c.customer_name
                """
                cursor.execute(query)
                results = cursor.fetchall()
                for row in results:
                    print(row)
        except pymysql.Error as error:
            print("Errore nell'aggregazione avanzata:", error)

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

db = AggregationDatabase()
db.advanced_aggregation()
db.close()

Esercizio 5: Query Finestra con mysql-connector-python (Sequenziale)

Utilizzare una query finestra (window query) per calcolare un valore cumulativo con mysql-connector-python.

import mysql.connector

def window_query():
    connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb')
    cursor = connection.cursor()
    query = """
    SELECT order_id, customer_id, total_amount,
           SUM(total_amount) OVER (PARTITION BY customer_id ORDER BY order_date) AS cumulative_total
    FROM orders
    """
    cursor.execute(query)
    results = cursor.fetchall()
    for row in results:
        print(row)
    cursor.close()
    connection.close()

window_query()

Esercizio 6: Pivot con PyMySQL (OOP)

Creare una query pivot per trasformare righe in colonne utilizzando PyMySQL in modalità OOP.

import pymysql

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

    def pivot_query(self):
        try:
            with self.connection.cursor() as cursor:
                query = """
                SELECT
                    customer_id,
                    MAX(CASE WHEN product_id = 1 THEN total_amount ELSE 0 END) AS product_1,
                    MAX(CASE WHEN product_id = 2 THEN total_amount ELSE 0 END) AS product_2,
                    MAX(CASE WHEN product_id = 3 THEN total_amount ELSE 0 END) AS product_3
                FROM orders
                GROUP BY customer_id
                """
                cursor.execute(query)
                results = cursor.fetchall()
                for row in results:
                    print(row)
        except pymysql.Error as error:
            print("Errore nella query pivot:", error)

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

db = PivotDatabase()
db.pivot_query()
db.close()

Esercizio 7: Query Ricorsiva con mysql-connector-python (Sequenziale)

Utilizzare una query ricorsiva per navigare una gerarchia di dati con mysql-connector-python.

import mysql.connector

def recursive_query():
    connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb')
    cursor = connection.cursor()
    query = """
    WITH RECURSIVE EmployeeCTE AS (
        SELECT employee_id, manager_id, employee_name
        FROM employees
        WHERE manager_id IS NULL
        UNION ALL
        SELECT e.employee_id, e.manager_id, e.employee_name
        FROM employees e
        INNER JOIN EmployeeCTE ecte ON e.manager_id = ecte.employee_id
    )
    SELECT * FROM EmployeeCTE
    """
    cursor.execute(query)
    results = cursor.fetchall()
    for row in results:
        print(row)
    cursor.close()
    connection.close()

recursive_query()

Esercizio 8: Funzione di Finestra con PyMySQL (OOP)

Utilizzare una funzione di finestra per calcolare il ranking dei record con PyMySQL in modalità OOP.

import pymysql

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

    def window_function_query(self):
        try:
            with self.connection.cursor() as cursor:
                query = """
                SELECT order_id, customer_id, total_amount,
                       RANK() OVER (PARTITION BY customer_id ORDER BY total_amount

 DESC) AS rank
                FROM orders
                """
                cursor.execute(query)
                results = cursor.fetchall()
                for row in results:
                    print(row)
        except pymysql.Error as error:
            print("Errore nella funzione di finestra:", error)

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

db = WindowFunctionDatabase()
db.window_function_query()
db.close()

Esercizio 9: Join con Funzione Aggregata con mysql-connector-python (Sequenziale)

Utilizzare una funzione aggregata in una query con join utilizzando mysql-connector-python.

import mysql.connector

def join_with_aggregate():
    connection = mysql.connector.connect(host='localhost', user='user', password='password', database='testdb')
    cursor = connection.cursor()
    query = """
    SELECT c.customer_name, SUM(o.total_amount) AS total_spent
    FROM customers c
    JOIN orders o ON c.customer_id = o.customer_id
    GROUP BY c.customer_name
    """
    cursor.execute(query)
    results = cursor.fetchall()
    for row in results:
        print(row)
    cursor.close()
    connection.close()

join_with_aggregate()

Esercizio 10: Query Parti Equi con PyMySQL (OOP)

Utilizzare una query per suddividere i record in parti eque basate su un criterio specifico con PyMySQL in modalità OOP.

import pymysql

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

    def partition_query(self):
        try:
            with self.connection.cursor() as cursor:
                query = """
                SELECT customer_id, order_id, total_amount,
                       NTILE(4) OVER (ORDER BY total_amount DESC) AS quartile
                FROM orders
                """
                cursor.execute(query)
                results = cursor.fetchall()
                for row in results:
                    print(row)
        except pymysql.Error as error:
            print("Errore nella query di partizione:", error)

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

db = EqualPartitionDatabase()
db.partition_query()
db.close()