Capitolo 20: Accesso a Database & ORM

Obiettivi

• Usare sqlite3 per aprire connessioni, eseguire query e transazioni.
• Eseguire operazioni CRUD con SQL grezzo.
• Definire tabelle ed eseguire query con SQLAlchemy Core.
• Associare classi Python a tabelle tramite SQLAlchemy ORM.
• Configurare ed eseguire migrazioni con Alembic.

1. Basi di sqlite3

import sqlite3

# connessione (crea file se mancante)
conn = sqlite3.connect('app.db')
cur = conn.cursor()

# crea tabella
cur.execute('''
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        email TEXT UNIQUE
    )
''')
conn.commit()

# inserimento e query
cur.execute("INSERT INTO users (name,email) VALUES (?,?)",
            ("Alice","alice@example.com"))
conn.commit()
cur.execute("SELECT id,name FROM users")
print(cur.fetchall())

# chiusura
cur.close()
conn.close()

2. Transazioni & Context Manager

from sqlite3 import connect

# context manager committa o rollbacka
with connect('app.db') as conn:
    cur = conn.cursor()
    cur.execute("UPDATE users SET name=? WHERE id=?",
                ("Alice Smith", 1))

# connessione chiusa automaticamente

Eccezioni nel blocco with eseguono rollback.

3. SQLAlchemy Core

from sqlalchemy import (
    create_engine, MetaData, Table, Column, Integer, String, select
)

engine = create_engine('sqlite:///app.db', echo=True)
metadata = MetaData()

users = Table(
    'users', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    Column('email', String, unique=True)
)
metadata.create_all(engine)

# inserimento e selezione
with engine.connect() as conn:
    conn.execute(users.insert(), [
        {"name":"Bob","email":"bob@example.com"}
    ])
    result = conn.execute(select(users)).fetchall()
    print(result)

4. SQLAlchemy ORM

from sqlalchemy.orm import (
    declarative_base, sessionmaker
)
from sqlalchemy import Column, Integer, String, create_engine

Base = declarative_base()
engine = create_engine('sqlite:///app.db', echo=False)
Session = sessionmaker(bind=engine)

class User(Base):
    __tablename__ = 'users'
    id    = Column(Integer, primary_key=True)
    name  = Column(String, nullable=False)
    email = Column(String, unique=True)

Base.metadata.create_all(engine)

# CRUD con Session
session = Session()
new_user = User(name="Carol", email="carol@example.com")
session.add(new_user)
session.commit()

users = session.query(User).filter_by(name="Carol").all()
print(users)
session.close()

5. Migrazioni con Alembic

# installa alembic
pip install alembic

# inizializza
alembic init migrations

# modifica alembic.ini per puntare a sqlite:///app.db,
# genera migrazione automatica
alembic revision --autogenerate -m "create users table"

# applica migrazione
alembic upgrade head

Usa le migrazioni per evolvere lo schema in modo sicuro.

Esercizi

1. Scrivi uno script per inserire, aggiornare e cancellare record con sqlite3 e stampare i risultati.
2. Ripeti le stesse operazioni con SQLAlchemy Core.
3. Definisci un modello ORM Product e fai query per fascia di prezzo.
4. Crea una nuova migrazione Alembic per aggiungere una colonna timestamp "created_at".