Chapter 20: Database Access & ORMs

Objectives

• Use sqlite3 to open connections, execute queries & transactions.
• Perform CRUD operations with raw SQL.
• Define tables & run queries using SQLAlchemy Core.
• Map Python classes to tables via SQLAlchemy ORM.
• Configure and run Alembic migrations.

1. sqlite3 Basics

import sqlite3

# connect (creates file if needed)
conn = sqlite3.connect('app.db')
cur = conn.cursor()

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

# insert and 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())

# cleanup
cur.close()
conn.close()

2. Transactions & Context Managers

from sqlite3 import connect

# context manager on connection commits or rolls back
with connect('app.db') as conn:
    cur = conn.cursor()
    cur.execute("UPDATE users SET name=? WHERE id=?",
                ("Alice Smith", 1))

# connection closed automatically

Exceptions inside the with block trigger 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)

# insert and select
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 using 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. Alembic Migrations

# install alembic
pip install alembic

# initialize
alembic init migrations

# edit alembic.ini to point to sqlite:///app.db,
# then auto-generate a migration
alembic revision --autogenerate -m "create users table"

# apply migration
alembic upgrade head

Use migrations to evolve schema safely over time.

Exercises

1. Write a script to insert, update & delete records in sqlite3 and print results.
2. Recreate the same operations using SQLAlchemy Core.
3. Define a Product ORM model and query by price range.
4. Create a new Alembic revision to add a "created_at" timestamp column.