Boto3 e AWS: Guida pratica Python per Data Science e DevOps (con codice)

import boto3
from botocore.exceptions import ClientError, NoCredentialsError
import sys

def verifica_connessione_e_lista_bucket():
    try:
        # 'client' è l'interfaccia low-level, perfetta per operazioni amministrative
        s3_client = boto3.client('s3')
        
        print("✅ Connessione a AWS riuscita!\n")
        print("I tuoi bucket S3 disponibili:")
        print("-" * 50)
        
        response = s3_client.list_buckets()
        buckets = response['Buckets']
        
        if not buckets:
            print("Nessun bucket trovato. Crea il tuo primo bucket!")
        else:
            for i, bucket in enumerate(buckets, 1):
                nome = bucket['Name']
                # Formattiamo la data per renderla leggibile
                creazione = bucket['CreationDate'].strftime('%Y-%m-%d %H:%M:%S')
                print(f"{i:2d}. {nome}  (creato il {creazione})")
                
            # Extra: Debug della regione del primo bucket
            loc = s3_client.get_bucket_location(Bucket=buckets[0]['Name'])
            region = loc['LocationConstraint'] or 'us-east-1 (default)'
            print(f"\n📍 Regione del primo bucket: {region}")
            
    except NoCredentialsError:
        print("❌ Credenziali AWS non trovate. Hai lanciato 'aws configure'?")
    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code == 'InvalidAccessKeyId':
            print("❌ Access Key non valida.")
        elif error_code == 'SignatureDoesNotMatch':
            print("❌ Secret Key errata.")
        else:
            print(f"Errore AWS imprevisto: {e}")

if __name__ == "__main__":
    verifica_connessione_e_lista_bucket()

import boto3
from botocore.exceptions import ClientError
import os
from pathlib import Path

def upload_dataset(file_path, bucket_name, s3_key=None, extra_args=None):
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File locale non trovato: {file_path}")
    
    s3_client = boto3.client('s3')
    s3_key = s3_key or Path(file_path).name
    
    # Best Practices per produzione:
    # 1. ACL: garantisce che il proprietario del bucket abbia il controllo completo
    # 2. Encryption: Mai salvare dati in chiaro (AES256 è lo standard)
    default_args = {
        'ACL': 'bucket-owner-full-control',
        'ServerSideEncryption': 'AES256'
    }
    if extra_args:
        default_args.update(extra_args)
    
    print(f"🚀 Upload: {file_path} → s3://{bucket_name}/{s3_key}")
    
    try:
        # upload_file gestisce automaticamente il multipart upload per file grossi
        s3_client.upload_file(
            Filename=file_path,
            Bucket=bucket_name,
            Key=s3_key,
            ExtraArgs=default_args
        )
        print(f"✅ Upload completato!")
        
    except ClientError as e:
        if e.response['Error']['Code'] == 'NoSuchBucket':
            print(f"❌ Il bucket '{bucket_name}' non esiste (i nomi sono globali!).")
        elif e.response['Error']['Code'] == 'AccessDenied':
            print("❌ Accesso Negato. Controlla le policy IAM del tuo utente.")
        else:
            raise e

# Esempio d'uso (Assicurati di creare prima un bucket vuoto e un file dummy)
if __name__ == "__main__":
    # Crea un file dummy al volo per il test
    with open("test_data.csv", "w") as f: f.write("id,valore\n1,100")
    
    upload_dataset(
        file_path="test_data.csv",
        bucket_name="inserisci-qui-il-tuo-bucket-univoco", 
        s3_key="raw/sales/2024/test_data.csv",
        extra_args={'ContentType': 'text/csv'}
    )

import boto3
import hashlib
from pathlib import Path
from botocore.exceptions import ClientError

def download_if_changed(bucket, key, local_path):
    s3 = boto3.client('s3')
    local_path = Path(local_path)
    
    # 1. Calcoliamo l'hash MD5 locale (se il file esiste)
    local_etag = None
    if local_path.exists():
        hasher = hashlib.md5()
        with open(local_path, 'rb') as f:
            # Leggiamo a blocchi per non saturare la RAM
            for chunk in iter(lambda: f.read(4096), b""):
                hasher.update(chunk)
        local_etag = f'"{hasher.hexdigest()}"' # S3 formatta l'ETag tra virgolette
    
    try:
        # 2. Chiediamo a S3 i metadati SENZA scaricare il file (head_object)
        head = s3.head_object(Bucket=bucket, Key=key)
        remote_etag = head['ETag']
        
        # Nota: Per file caricati in multipart, l'ETag S3 non è un semplice MD5.
        # In quel caso servirebbe una logica di confronto più complessa.
        
        if local_etag == remote_etag:
            print("😎 File già aggiornato localmente. Risparmiato download.")
            return str(local_path)
            
        print(f"🔄 Nuova versione trovata (ETag {remote_etag}). Download...")
        local_path.parent.mkdir(parents=True, exist_ok=True)
        s3.download_file(bucket, key, str(local_path))
        print(f"✅ Download completato: {local_path}")
        return str(local_path)
        
    except ClientError as e:
        print("❌ Errore nel download (File non trovato o permessi mancanti).")
        raise

# Esempio (Usa i parametri dell'esercizio precedente)
# download_if_changed("tuo-bucket", "raw/sales/2024/test_data.csv", "cache/data.csv")

import boto3
import time

ec2 = boto3.client('ec2')
INSTANCE_ID = 'i-xxxxxxxxxxxxxxxxx'  # Inserisci l'ID della tua istanza di test

def gestisci_ciclo_vita():
    print(f"Avvio istanza {INSTANCE_ID}...")
    ec2.start_instances(InstanceIds=[INSTANCE_ID])
    
    # I Waiter sono fondamentali: bloccano lo script finché lo stato non è raggiunto
    waiter = ec2.get_waiter('instance_running')
    waiter.wait(InstanceIds=[INSTANCE_ID])
    print("✅ Istanza RUNNING. Ora potresti connetterti via SSH ed eseguire il training.")
    
    # Simulazione del lavoro
    print("⏳ Esecuzione task in corso (simulato 5 sec)...")
    time.sleep(5)
    
    print("🛑 Task finito. Arresto istanza per risparmiare costi...")
    ec2.stop_instances(InstanceIds=[INSTANCE_ID])
    ec2.get_waiter('instance_stopped').wait(InstanceIds=[INSTANCE_ID])
    print("✅ Istanza STOPPED. Non stai più pagando per la compute capacity.")

# if __name__ == "__main__":
#    gestisci_ciclo_vita() 

import boto3
from datetime import datetime
from decimal import Decimal
import json

# Usa 'resource' per DynamoDB: è più pythonico e facile da usare del 'client'
dynamodb = boto3.resource('dynamodb')
# Assicurati di aver creato la tabella 'ML-Experiments' con Partition Key: ExperimentID (String)
table = dynamodb.Table('ML-Experiments')

def salva_esperimento(experiment_id, params, metrics):
    # Timestamp ISO 8601
    now = datetime.utcnow().isoformat()
    
    item = {
        'ExperimentID': experiment_id,
        'Timestamp': now,
        'Parameters': params,   # DynamoDB accetta dizionari annidati (Mappe)
        'Metrics': metrics,
        'Status': 'COMPLETED'
    }
    
    # Trick: DynamoDB richiede Decimal invece di float python nativi
    item_ready = json.loads(json.dumps(item), parse_float=Decimal)
    
    print(f"💾 Salvataggio esperimento {experiment_id}...")
    table.put_item(Item=item_ready)
    print("✅ Salvato.")

def leggi_esperimento(experiment_id):
    response = table.get_item(Key={'ExperimentID': experiment_id})
    if 'Item' in response:
        print(f"\n🔍 Dettagli esperimento {experiment_id}:")
        print(response['Item'])
    else:
        print("❌ Esperimento non trovato.")

# Esempio d'uso
if __name__ == "__main__":
    # Nota come Parameters e Metrics siano strutture flessibili
    salva_esperimento(
        experiment_id="exp-ResNet-042",
        params={"lr": 0.001, "batch": 64, "optimizer": "adam"},
        metrics={"accuracy": 0.945, "loss": 0.12}
    )
    
    leggi_esperimento("exp-ResNet-042")