Esercizi Query Avanzate MySQL Python

Codegrind Team•Jul 10 2024

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