Python per Data Engineering: 3 esercizi reali su Excel, API e automazione aziendale

Nelle aziende ci si scontra ogni giorno con fiumi di dati intrappolati in fogli di calcolo disordinati, database da aggiornare e servizi web che non si parlano tra loro.

Invece di passare ore a copiare e incollare righe manualmente su Excel, possiamo lasciare che sia Python a fare il lavoro noioso e ripetitivo.

Questi due esercizi pratici affrontano scenari reali di automazione aziendale: la pulizia dei dati e l’integrazione con API esterne. Se hai già le basi del linguaggio e sai installare un pacchetto con pip, qui troverai pane per i tuoi denti per capire come si lavora davvero su un progetto di Data Analysis.

Esercizio 1 – Filtraggio e caricamento di dati da Excel a SQLite

Testo

Sei un analista in un’azienda che riceve quotidianamente un file Excel con i dati delle vendite. Il file ha le seguenti colonne:

• ID (intero)
• Prodotto (stringa)
• Quantità (intero)
• PrezzoUnitario (decimale)
• Data (data, formato YYYY-MM-DD)

Scrivi uno script Python che:

1. Legga il file Excel (vendite.xlsx).
2. Filtri le righe dove Quantità > 10.
3. Aggiunga una colonna calcolata Totale = Quantità * PrezzoUnitario.
4. Carichi i dati filtrati in una tabella SQLite chiamata vendite_high_volume (se la tabella esiste, sovrascrivila).
5. Esporti gli stessi dati filtrati in un nuovo file Excel high_volume_vendite.xlsx.

Il file vendite.xlsx di esempio è disponibile nella cartella del progetto.

In alternativa, a fini didattici, lo generiamo con il seguente snippet:

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# Genera dati di esempio per le vendite
np.random.seed(42) # per riproducibilità

# Prodotti di esempio
prodotti = ['Laptop', 'Mouse', 'Tastiera', 'Monitor', 'Stampante', 'Smartphone', 'Tablet']

# Crea un DataFrame con 100 righe di vendite casuali
num_righe = 100

# Genera ID da 1 a 100
id_vals = list(range(1, num_righe + 1))

# Seleziona prodotti casuali
prodotti_vals = np.random.choice(prodotti, num_righe)

# Quantità: valori tra 1 e 50 (alcuni >10 per il filtro)
quantita_vals = np.random.randint(1, 51, num_righe)

# Prezzi unitari tra 10 e 1000 (con due decimali)
prezzi_vals = np.round(np.random.uniform(10, 1000, num_righe), 2)

# Date casuali nel corso del 2025
start_date = datetime(2025, 1, 1)
end_date = datetime(2025, 12, 31)
delta = end_date - start_date
date_vals = [start_date + timedelta(days=np.random.randint(0, delta.days + 1)) for _ in range(num_righe)]
# Formatta le date come stringhe YYYY-MM-DD
date_vals = [d.strftime('%Y-%m-%d') for d in date_vals]

# Crea il DataFrame
df = pd.DataFrame({
'ID': id_vals,
'Prodotto': prodotti_vals,
'Quantità': quantita_vals,
'PrezzoUnitario': prezzi_vals,
'Data': date_vals
})

# Salva il file Excel
file_path = 'vendite.xlsx'
df.to_excel(file_path, index=False, engine='openpyxl')

print(f"File Excel generato: {file_path} ({len(df)} righe)")
print("Alcune righe di esempio:")
print(df.head())
print("\nStatistiche:")
print(df.describe())
print(f"\nNumero di righe con Quantità > 10: {len(df[df['Quantità'] > 10])}")

File Excel generato: vendite.xlsx (100 righe)
Alcune righe di esempio:
   ID   Prodotto  Quantità  PrezzoUnitario        Data
0   1     Tablet         8          884.45  2025-05-15
1   2    Monitor        14          331.10  2025-07-14
2   3  Stampante        23          130.87  2025-05-08
3   4     Tablet        40          362.73  2025-02-02
4   5   Tastiera        21          907.76  2025-06-25

Statistiche:
               ID    Quantità  PrezzoUnitario
count  100.000000  100.000000      100.000000
mean    50.500000   25.700000      518.140500
std     29.011492   14.014062      282.776322
min      1.000000    1.000000       10.520000
25%     25.750000   14.000000      276.067500
50%     50.500000   26.500000      494.855000
75%     75.250000   37.500000      750.267500
max    100.000000   49.000000      997.760000

Numero di righe con Quantità > 10: 80

Soluzione

Librerie utilizzate

• pandas per la manipolazione dei dati tabellari.
• sqlite3 per il database.
• openpyxl come motore di lettura/scrittura per Excel (usato da pandas).

Installa le dipendenze con:

pip install pandas openpyxl

Codice completo con commenti didattici

import pandas as pd
import sqlite3
from pathlib import Path

# 1. Leggiamo il file Excel
#    Pandas legge automaticamente la prima riga come intestazione.
file_excel = Path("vendite.xlsx")
df = pd.read_excel(file_excel, engine="openpyxl")

print(f"📊 Lette {len(df)} righe dal file Excel.")

# 2. Filtraggio: selezioniamo solo le righe con Quantità > 10
df_filtrato = df[df["Quantità"] > 10].copy()  # .copy() per evitare warning

# 3. Calcolo della colonna Totale
df_filtrato["Totale"] = df_filtrato["Quantità"] * df_filtrato["PrezzoUnitario"]

print(f"🔍 Dopo il filtraggio rimangono {len(df_filtrato)} righe.")

# 4. Caricamento in SQLite
#    Creiamo (o apriamo) il database 'azienda.db'
conn = sqlite3.connect("azienda.db")

#   Usiamo il metodo to_sql di pandas: sostituisce la tabella se esiste
df_filtrato.to_sql(
    name="vendite_high_volume",
    con=conn,
    if_exists="replace",   # sovrascrive la tabella
    index=False            # non scrivere l'indice di pandas
)

conn.close()
print("✅ Dati caricati nel database SQLite (tabella 'vendite_high_volume').")

# 5. Esportiamo in un nuovo file Excel
df_filtrato.to_excel("high_volume_vendite.xlsx", index=False)
print("📁 File 'high_volume_vendite.xlsx' creato con successo.")

Output:
📊 Lette 100 righe dal file Excel.
🔍 Dopo il filtraggio rimangono 80 righe.
✅ Dati caricati nel database SQLite (tabella 'vendite_high_volume').
📁 File 'high_volume_vendite.xlsx' creato con successo.

💡 Osservazioni

• Pandas offre un’interfaccia ad alto livello che astrae i dettagli di lettura/scrittura di Excel e di connessione al DB. In un contesto production-ready, si potrebbe aggiungere la gestione di eventuali colonne mancanti o tipi di dato errati.
• Il metodo to_sql con if_exists="replace" cancella e ricrea la tabella. Attenzione: se la tabella ha vincoli di chiave esterna, potrebbero essere violati; in tal caso si userebbe "append" o una logica più fine.

Questo esercizio non è un semplice test di programmazione, ma rappresenta il primo vero passo verso l’ingegneria dei dati (Data Pipeline).

• Obbliga lo studente a passare dal concetto di foglio di calcolo “piatto” (Excel) alla struttura rigida e relazionale di un database (SQLite).

• Il valore reale: In azienda, i dati arrivano spesso frammentati. Saper filtrare all’ingresso le informazioni (in questo caso, isolare l’alto volume di vendite) prima di storicizzarle evita di sovraccaricare i sistemi di produzione. L’uso della sovrascrittura (if_exists="replace") introduce al concetto di sovrascrittura dei dati temporanei, un pattern comune nei processi ETL (Extract, Transform, Load) leggeri.

Mini quiz / Domande di riflessione

Esercizio 2 – Chiamata API, salvataggio in DB e report Excel

Testo

La tua azienda vuole monitorare il tasso di cambio EUR/USD ogni ora. Scrivi uno script Python che:

1. Chiami l’API pubblica di ExchangeRate-API (https://api.exchangerate-api.com/v4/latest/EUR) per ottenere il tasso di cambio corrente.
2. Memorizzi in un database SQLite (tabella tassi_cambio) il timestamp della richiesta e il tasso EUR/USD.
3. Generi un report Excel giornaliero (report_cambio.xlsx) contenente il tasso medio, massimo e minimo delle ultime 24 ore (rispetto all’orario di esecuzione).
4. Gestisca eventuali errori di rete con un meccanismo di riprova (almeno 3 tentativi).

Soluzione passo passo

Librerie utilizzate

• requests per effettuare la chiamata HTTP.
• pandas per l’elaborazione e l’esportazione.
• sqlite3 per il database.
• datetime per la gestione delle date.
• time per i ritardi nei retry.

Installa requests e pandas se non lo hai già fatto.

Codice completo

import requests
import sqlite3
import pandas as pd
from datetime import datetime, timedelta
import time
import logging

# Configuriamo un logger base (utile in produzione)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def ottieni_tasso_con_retry(max_tentativi=3, delay=2):
    """Chiama l'API per il tasso EUR/USD con tentativi di riprova."""
    url = "https://api.exchangerate-api.com/v4/latest/EUR"
    for tentativo in range(1, max_tentativi + 1):
        try:
            response = requests.get(url, timeout=10)
            response.raise_for_status()  # solleva eccezione per status 4xx/5xx
            data = response.json()
            tasso_usd = data['rates']['USD']
            logging.info(f"Tasso ottenuto: {tasso_usd} (tentativo {tentativo})")
            return tasso_usd
        except (requests.RequestException, KeyError) as e:
            logging.warning(f"Tentativo {tentativo} fallito: {e}")
            if tentativo == max_tentativi:
                raise  # rilancia l'eccezione dopo l'ultimo tentativo
            time.sleep(delay)  # aspetta prima di riprovare

def salva_tasso_in_db(tasso):
    """Inserisce il tasso con timestamp corrente nel DB."""
    conn = sqlite3.connect("cambio.db")
    cursor = conn.cursor()
    # Creiamo la tabella se non esiste (con la data di default)
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS tassi_cambio (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            timestamp TEXT NOT NULL,
            tasso REAL NOT NULL
        )
    ''')
    ora = datetime.now().isoformat()
    cursor.execute("INSERT INTO tassi_cambio (timestamp, tasso) VALUES (?, ?)", (ora, tasso))
    conn.commit()
    conn.close()
    logging.info("Tasso salvato nel database.")

def genera_report_24h():
    """Legge i dati delle ultime 24 ore e crea un Excel con statistiche."""
    conn = sqlite3.connect("cambio.db")
    # Calcoliamo il timestamp di 24 ore fa
    soglia = (datetime.now() - timedelta(hours=24)).isoformat()
    query = """
        SELECT tasso, timestamp
        FROM tassi_cambio
        WHERE timestamp >= ?
    """
    df = pd.read_sql_query(query, conn, params=(soglia,))
    conn.close()

    if df.empty:
        logging.warning("Nessun dato nelle ultime 24 ore. Report non generato.")
        return

    # Calcolo statistiche
    media = df['tasso'].mean()
    massimo = df['tasso'].max()
    minimo = df['tasso'].min()
    conteggio = len(df)

    # Creiamo un DataFrame di riepilogo
    report_df = pd.DataFrame({
        'Statistica': ['Media', 'Massimo', 'Minimo', 'Numero campioni'],
        'Valore': [media, massimo, minimo, conteggio]
    })

    # Aggiungiamo anche l'andamento storico (tutti i dati) in un secondo foglio
    with pd.ExcelWriter("report_cambio.xlsx", engine="openpyxl") as writer:
        report_df.to_excel(writer, sheet_name="Riepilogo", index=False)
        df.to_excel(writer, sheet_name="Dettaglio", index=False)

    logging.info("Report Excel 'report_cambio.xlsx' generato.")

if __name__ == "__main__":
    try:
        tasso = ottieni_tasso_con_retry()
        salva_tasso_in_db(tasso)
        genera_report_24h()
    except Exception as e:
        logging.error(f"Script terminato con errore: {e}")

💡 Osservazioni

• La funzione ottieni_tasso_con_retry implementa un pattern di retry con backoff (sebbene qui il ritardo sia fisso; in produzione si usa spesso un backoff esponenziale).
• La tabella tassi_cambio salva i timestamp come stringhe ISO. Per query più efficienti, si potrebbe usare il tipo DATETIME di SQLite (che però non esiste nativamente, si può usare TEXT con formati coerenti).
• Il report Excel contiene due fogli: un riepilogo e il dettaglio di tutti i dati delle ultime 24 ore.

Questo secondo scenario alza l’asticella, introducendo dinamiche di rete e persistenza temporale.

• Il fattore “Mondo Reale” (Errori di rete): Le API falliscono. La rete cade. Un server può andare in timeout. Inserire un meccanismo di retry con gestione delle eccezioni insegna che il codice non gira in un ambiente perfetto. Questo distingue uno script scolastico da un software pronto per la produzione.

• La logica del Report Manageriale: Estrarre dati, calcolare metriche aggregate (media, massimo, minimo) e formattarle in un file Excel multi-foglio simula esattamente la richiesta del reparto Finance di qualunque azienda. Mostra come Python possa fare da ponte tra i dati grezzi del web e gli strumenti decisionali usati dal management.

Mini quiz / Domande di riflessione

Risposte alle domande di riflessione

Di seguito le risposte dettagliate per ogni esercizio.

Esercizio 1

Perché usare pandas per leggere Excel invece di openpyxl direttamente?
pandas fornisce un’interfaccia tabellare ad alto livello (DataFrame) che semplifica operazioni di filtraggio, trasformazione e aggregazione. Inoltre, leggere un Excel con pandas richiede solo una riga di codice, mentre con openpyxl bisogna gestire celle, righe e fogli manualmente. Pandas è più conciso e potente per l’analisi dei dati.

Cosa succede se il file Excel contiene celle vuote o dati non numerici in PrezzoUnitario?
Pandas converte automaticamente le colonne con dati misti in tipo object; se si prova a moltiplicare una stringa per un numero, si ottiene un errore (TypeError). In un contesto production-ready, si dovrebbero validare i tipi: ad esempio, usando pd.to_numeric(..., errors='coerce') per convertire in NaN i valori non numerici, e poi decidere come gestire i NaN (es. eliminarli o sostituirli con un valore default).

Se il file avesse 1 milione di righe, il caricamento in memoria con pandas sarebbe efficiente?
Pandas carica tutto in memoria, quindi con 1 milione di righe potrebbe causare un elevato consumo di RAM. In tal caso si potrebbe usare il parametro chunksize in pd.read_excel (se supportato) per leggere a blocchi, oppure passare a librerie come dask o a lettura diretta con openpyxl iterando sulle righe senza caricare tutto il DataFrame.

Esercizio 2

Perché è importante gestire le eccezioni durante le chiamate API?
Senza gestione, un errore di rete o un timeout farebbe terminare lo script, interrompendo il flusso di lavoro. Gestendo le eccezioni, possiamo riprovare (come fatto) e, se persistono, loggare l’errore e notificare gli amministratori, garantendo che il sistema continui a funzionare (o almeno venga segnalato il problema).

Come potresti rendere lo script eseguibile automaticamente ogni ora?
Si può usare il pianificatore di sistema: su Linux, un cron job (0 * * * * /usr/bin/python3 script.py); su Windows, Task Scheduler. In alternativa, si può avviare un loop con time.sleep(3600) all’interno dello script stesso, ma in produzione si preferisce il sistema operativo per una gestione più robusta.

Quale strategia adotteresti per evitare di generare report se non ci sono abbastanza dati?
Si può controllare il numero di campioni (es. len(df)) e, se inferiore a una soglia (es. 5), saltare la generazione del report e loggare un avviso. Inoltre, si potrebbe impostare un periodo minimo di raccolta dati prima di produrre statistiche significative.

import os
import smtplib
import logging
from pathlib import Path
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.base import MIMEBase
from email import encoders

# Configurazione del logger
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# ==========================================
# GESTIONE SICURA DELLE CREDENZIALI
# ==========================================
EMAIL_MITTENTE = os.environ.get("AZIENDA_EMAIL")
EMAIL_PASSWORD = os.environ.get("AZIENDA_PASSWORD")
EMAIL_DESTINATARIO = "[email protected]"

# Politica "Fail-Fast": blocchiamo lo script all'avvio se mancano i parametri vitali
if not EMAIL_MITTENTE or not EMAIL_PASSWORD:
    raise ValueError(
        "❌ ERRORE CRITICO: Le variabili d'ambiente 'AZIENDA_EMAIL' o 'AZIENDA_PASSWORD' "
        "non sono configurate nel sistema. Impossibile avviare lo script."
    )

SMTP_SERVER = "smtp.gmail.com"  # Configurazione per Gmail (modificabile in base al provider)
SMTP_PORT = 587                 # Porta standard per STARTTLS
SOGLIA_CRITICA = 1.12  

def invia_email_alert(file_allegato, tasso_rilevato):
    """Costruisce e invia un'email di alert con allegato, gestendo le eccezioni di rete."""
    
    # 1. Controllo preventivo dell'esistenza del file
    path_allegato = Path(file_allegato)
    if not path_allegato.exists():
        logging.error(f"❌ Invio email annullato: Il file '{file_allegato}' non esiste sul disco.")
        return False

    # 2. Creazione della struttura del messaggio (MIME)
    msg = MIMEMultipart()
    msg['From'] = EMAIL_MITTENTE
    msg['To'] = EMAIL_DESTINATARIO
    msg['Subject'] = f"🚨 ALERT: Tasso EUR/USD critico rilevato ({tasso_rilevato})"

    # Corpo del messaggio formattato in HTML per una presentazione professionale
    corpo_html = f"""
    <html>
      <body style="font-family: Arial, sans-serif; color: #333;">
        <h2 style="color: #d9534f;">Rilevata Anomalia di Mercato</h2>
        <p>Il tasso di cambio EUR/USD ha superato la soglia critica aziendale di <b>{SOGLIA_CRITICA}</b>.</p>
        <p>Ultimo valore rilevato: <span style="font-size: 18px; color: #d9534f;"><b>{tasso_rilevato}</b></span></p>
        <p>In allegato trovi il report completo con lo storico delle ultime 24 ore.</p>
        <hr style="border: 0; border-top: 1px solid #ccc;">
        <p style="font-size: 11px; color: #777;">Messaggio automatico generato dal sistema di monitoraggio valute.</p>
      </body>
    </html>
    """
    msg.attach(MIMEText(corpo_html, 'html'))

    # 3. Preparazione e codifica dell'allegato Excel
    try:
        with open(path_allegato, "rb") as attachment:
            part = MIMEBase("application", "octet-stream")
            part.set_payload(attachment.read())
            
        # Codifica in Base64 per il trasferimento sicuro via email
        encoders.encode_base64(part)
        part.add_header(
            "Content-Disposition",
            f"attachment; filename={path_allegato.name}",
        )
        msg.attach(part)
        
        # 4. Connessione al server e invio
        logging.info(f"Tentativo di connessione a {SMTP_SERVER}:{SMTP_PORT} in corso...")
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=15) as server:
            server.starttls()  # Attiva la crittografia della sessione
            server.login(EMAIL_MITTENTE, EMAIL_PASSWORD)
            server.send_message(msg)
            
        logging.info(f"📧 Email di alert inviata con successo a {EMAIL_DESTINATARIO}")
        return True
        
    except smtplib.SMTPAuthenticationError:
        logging.error("❌ Errore SMTP: Credenziali errate o App Password di Google non configurata.")
    except smtplib.SMTPConnectError:
        logging.error("❌ Errore SMTP: Impossibile connettersi al server. Controlla la rete o i blocchi del firewall.")
    except Exception as e:
        logging.error(f"❌ Errore imprevisto durante l'invio dell'email: {e}")
    return False

# ==========================================
# FLUSSO DI ESECUZIONE ORCHESTRATO
# ==========================================
if __name__ == "__main__":
    try:
        # Recupero ed elaborazione dati (funzioni dell'Esercizio 2)
        tasso = ottieni_tasso_con_retry()
        salva_tasso_in_db(tasso)
        
        # Memorizziamo se il report è stato creato davvero (ritorna True o False)
        report_generato = genera_report_24h() 
        
        # Logica decisionale basata sulla soglia finanziaria
        if tasso > SOGLIA_CRITICA:
            logging.warning(f"⚠️ Tasso sopra la soglia limite ({tasso} > {SOGLIA_CRITICA}).")
            
            # Controllo incrociato delle dipendenze: l'alert parte solo se c'è il file
            if report_generato:
                invia_email_alert("report_cambio.xlsx", tasso)
            else:
                logging.error("🚨 Alert bloccato: Tasso critico, ma mancano i dati storici delle ultime 24 ore.")
        else:
            logging.info(f"📈 Tasso nella norma ({tasso}). Nessun alert inviato.")
            
    except ValueError as ve:
        logging.error(f"Errore di configurazione iniziale: {ve}")
    except Exception as e:
        logging.error(f"Script interrotto per un errore imprevisto: {e}")

2026-06-17 09:30:01,102 - INFO - Tasso ottenuto: 1.1345 (tentativo 1)
2026-06-17 09:30:01,155 - INFO - Tasso salvato nel database.
2026-06-17 09:30:01,210 - INFO - Report Excel 'report_cambio.xlsx' generato.
2026-06-17 09:30:01,212 - WARNING - ⚠️ Tasso sopra la soglia limite (1.1345 > 1.12).
2026-06-17 09:30:01,213 - INFO - Tentativo di connessione a smtp.gmail.com:587 in corso...
2026-06-17 09:30:02,890 - INFO - 📧 Email di alert inviata con successo a [email protected]

2026-06-17 10:30:00,850 - INFO - Tasso ottenuto: 1.0921 (tentativo 1)
2026-06-17 10:30:00,892 - INFO - Tasso salvato nel database.
2026-06-17 10:30:00,945 - INFO - Report Excel 'report_cambio.xlsx' generato.
2026-06-17 10:30:00,947 - INFO - 📈 Tasso nella norma (1.0921). Nessun alert inviato.

Traceback (most recent call last):
  File "monitor_valute.py", line 17, in <module>
    raise ValueError(
ValueError: ❌ ERRORE CRITICO: Le variabili d'ambiente 'AZIENDA_EMAIL' o 'AZIENDA_PASSWORD' non sono configurate nel sistema. Impossibile avviare lo script.

Process finished with exit code 1

2026-06-17 11:15:02,310 - INFO - Tasso ottenuto: 1.1250 (tentativo 1)
2026-06-17 11:15:02,340 - INFO - Tasso salvato nel database.
2026-06-17 11:15:02,345 - WARNING - Nessun dato nelle ultime 24 ore. Report non generato.
2026-06-17 11:15:02,346 - WARNING - ⚠️ Tasso sopra la soglia limite (1.125 > 1.12).
2026-06-17 11:15:02,347 - ERROR - 🚨 Alert bloccato: Tasso critico, ma mancano i dati storici delle ultime 24 ore.

2026-06-17 12:00:01,520 - INFO - Tasso ottenuto: 1.1410 (tentativo 1)
2026-06-17 12:00:01,570 - INFO - Tasso salvato nel database.
2026-06-17 12:00:01,630 - INFO - Report Excel 'report_cambio.xlsx' generato.
2026-06-17 12:00:01,632 - WARNING - ⚠️ Tasso sopra la soglia limite (1.141 > 1.12).
2026-06-17 12:00:01,633 - INFO - Tentativo di connessione a smtp.gmail.com:587 in corso...
2026-06-17 12:00:02,410 - ERROR - ❌ Errore SMTP: Credenziali errate o App Password di Google non configurata.