RugvAI Labs

Databases with Python: SQLite, SQLAlchemy & Data Access

Every production app eventually needs a database — SQLite for desktop tools, Postgres for web services, both accessed from Python through DB-API or SQLAlchemy. Knowing parameterized queries, transactions, and connection pooling is what separates safe code from SQL-injection bait.

Learning Objectives

After this lesson, you will be able to:

Explain why databases are essential for data science beyond CSVs
Use sqlite3 to create tables, insert data, and run queries
Write parameterized queries to prevent SQL injection attacks
Use context managers for safe, automatic database connection handling
Define SQLAlchemy ORM models with relationships and run queries
Integrate Pandas with SQL databases using read_sql() and to_sql()
Build a real-world data pipeline that combines all three approaches

Why

Why This Matters

CSVs break at scale -- loading a 10 GB CSV into memory crashes your laptop. Databases let you query only the rows you need, keep data on disk, and handle concurrent access from multiple processes
Every production ML system stores data in a database -- training metadata, feature stores, model registries, experiment tracking (MLflow), and prediction logs all live in PostgreSQL, SQLite, or similar systems. CSVs are for prototyping; databases are for production
SQL is the universal data language -- whether you use Python, R, Java, or a BI tool like Tableau, SQL is how you talk to data. Learning it once gives you access to virtually every dataset in every company
SQL injection is a top security vulnerability -- if you build a web app or API that talks to a database, one missing parameterized query can let attackers read, modify, or delete your entire database. This is not theoretical; it is the most common web attack

Prerequisites

Before You Start

Review

#Why Databases for Data Science

#The Problem with CSVs

CSVs are simple and ubiquitous, but they have serious limitations for real-world data work:

pythonrunnable cell

# The CSV approach -- fine for small data
import csv

# Reading a CSV loads EVERYTHING into memory
with open("users.csv") as f:
    reader = csv.DictReader(f)
    users = list(reader)  # all rows in memory at once

# Want only active users? You still loaded ALL users first
active = [u for u in users if u["status"] == "active"]

# Want to join users with orders? Load ANOTHER file into memory
with open("orders.csv") as f:
    reader = csv.DictReader(f)
    orders = list(reader)

# Manual join -- slow, error-prone, memory-hungry
user_orders = []
for user in active:
    for order in orders:
        if order["user_id"] == user["id"]:
            user_orders.append({**user, **order})

Feature	CSV	Database (SQL)
File size limit	RAM-limited	Disk-limited (terabytes)
Query specific rows	Must load all, then filter	`WHERE` clause fetches only matching rows
Join tables	Manual, slow, error-prone	`JOIN` in SQL, optimized by query planner
Concurrent access	File locking, corruption risk	Built-in transaction support (ACID)
Data types	Everything is a string	Typed columns (INT, FLOAT, TEXT, DATE)
Indexing	None (linear scan)	B-tree indexes for O(log n) lookup
Data integrity	None	Constraints, foreign keys, unique checks

#SQL Basics in 60 Seconds

sql

-- Create a table (like defining a class)
CREATE TABLE users (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE,
    age INTEGER,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Insert data
INSERT INTO users (name, email, age) VALUES ('Alice', 'alice@example.com', 28);
INSERT INTO users (name, email, age) VALUES ('Bob', 'bob@example.com', 34);

-- Query data
SELECT name, age FROM users WHERE age > 30;          -- Filter
SELECT COUNT(*) FROM users;                           -- Count
SELECT age, COUNT(*) FROM users GROUP BY age;         -- Group
SELECT * FROM users ORDER BY created_at DESC LIMIT 5; -- Sort + limit

#sqlite3 -- Python's Built-In Database

SQLite is a serverless, file-based database that comes bundled with Python. No installation, no configuration, no server process -- just import sqlite3. It powers billions of devices: every iPhone, every Android phone, every Firefox and Chrome browser uses SQLite.

#Connecting and Creating Tables

pythonrunnable cell

import sqlite3

# Connect to a database (creates the file if it doesn't exist)
conn = sqlite3.connect("ml_experiments.db")
cursor = conn.cursor()

# Create a table for ML experiment tracking
cursor.execute("""
    CREATE TABLE IF NOT EXISTS experiments (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        model_name TEXT NOT NULL,
        dataset TEXT NOT NULL,
        accuracy REAL,
        loss REAL,
        epochs INTEGER,
        learning_rate REAL,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

conn.commit()  # save changes to disk
print("Table created successfully!")

#Inserting Data

pythonrunnable cell

import sqlite3

conn = sqlite3.connect("ml_experiments.db")
cursor = conn.cursor()

# Insert a single experiment
cursor.execute("""
    INSERT INTO experiments (model_name, dataset, accuracy, loss, epochs, learning_rate)
    VALUES (?, ?, ?, ?, ?, ?)
""", ("ResNet50", "CIFAR-10", 0.923, 0.245, 100, 0.001))

# Insert multiple experiments at once
experiments = [
    ("VGG16", "CIFAR-10", 0.891, 0.312, 80, 0.01),
    ("ResNet50", "ImageNet", 0.761, 0.892, 90, 0.001),
    ("BERT-base", "SST-2", 0.928, 0.187, 3, 0.00002),
    ("GPT-2", "WikiText", 0.0, 3.21, 1, 0.0001),
    ("ResNet18", "CIFAR-10", 0.887, 0.334, 50, 0.01),
    ("ViT-B/16", "ImageNet", 0.812, 0.723, 300, 0.001),
    ("BERT-large", "SST-2", 0.942, 0.162, 3, 0.00001),
    ("ResNet50", "CIFAR-100", 0.712, 0.987, 200, 0.001),
]

cursor.executemany("""
    INSERT INTO experiments (model_name, dataset, accuracy, loss, epochs, learning_rate)
    VALUES (?, ?, ?, ?, ?, ?)
""", experiments)

conn.commit()
print(f"Inserted {cursor.rowcount} experiments")

conn.close()

#Querying Data

pythonrunnable cell

import sqlite3

conn = sqlite3.connect("ml_experiments.db")

# Return rows as dictionaries instead of tuples
conn.row_factory = sqlite3.Row
cursor = conn.cursor()

# SELECT: fetch specific columns with a filter
cursor.execute("""
    SELECT model_name, dataset, accuracy, epochs
    FROM experiments
    WHERE accuracy > 0.9
    ORDER BY accuracy DESC
""")

print("High-accuracy experiments:")
for row in cursor.fetchall():
    print(f"  {row['model_name']} on {row['dataset']}: "
          f"{row['accuracy']:.3f} ({row['epochs']} epochs)")

# Aggregation: average accuracy per dataset
cursor.execute("""
    SELECT dataset, 
           COUNT(*) as num_experiments,
           AVG(accuracy) as avg_accuracy,
           MAX(accuracy) as best_accuracy
    FROM experiments
    GROUP BY dataset
    ORDER BY avg_accuracy DESC
""")

print("\nDataset summary:")
for row in cursor.fetchall():
    print(f"  {row['dataset']}: {row['num_experiments']} experiments, "
          f"avg={row['avg_accuracy']:.3f}, best={row['best_accuracy']:.3f}")

conn.close()

#Parameterized Queries -- Preventing SQL Injection

Watch Out

Watch out

Fix: Always use parameterized queries (? placeholders) to prevent SQL injection. Never put user input directly into SQL strings.

Building SQL queries with string formatting or f-strings

SQL injection is the most common web application vulnerability. When you embed user input directly into SQL, an attacker can terminate your query and execute arbitrary SQL.

```python

DANGEROUS -- SQL injection vulnerability!

user_input = "Alice'; DROP TABLE experiments; --" query = f"SELECT * FROM users WHERE name = '{user_input}'"

This becomes:

SELECT * FROM users WHERE name = 'Alice'; DROP TABLE experiments; --'

The attacker just deleted your entire table!

SAFE -- parameterized query

cursor.execute("SELECT * FROM users WHERE name = ?", (user_input,))

The database treats the entire input as a string value, not as SQL code

No matter what the user types, it cannot change the query structure

```

This applies to ALL SQL interfaces: sqlite3, psycopg2, MySQL connector, SQLAlchemy text(). The ? placeholder (or %s or :name depending on the library) tells the database driver to safely escape the value. Never use f-strings, .format(), or + concatenation for SQL queries.

pythonrunnable cell

import sqlite3

conn = sqlite3.connect("ml_experiments.db")
cursor = conn.cursor()

# DANGEROUS -- never do this!
# user_input = "ResNet50"
# cursor.execute(f"SELECT * FROM experiments WHERE model_name = '{user_input}'")

# SAFE -- always use parameterized queries
model_name = "ResNet50"
cursor.execute(
    "SELECT * FROM experiments WHERE model_name = ? AND accuracy > ?",
    (model_name, 0.7)
)

results = cursor.fetchall()
print(f"Found {len(results)} experiments for {model_name}")

# Named parameters (alternative style)
cursor.execute(
    "SELECT * FROM experiments WHERE dataset = :dataset AND epochs >= :min_epochs",
    {"dataset": "CIFAR-10", "min_epochs": 50}
)

results = cursor.fetchall()
print(f"Found {len(results)} CIFAR-10 experiments with 50+ epochs")

conn.close()

#UPDATE and DELETE

pythonrunnable cell

import sqlite3

conn = sqlite3.connect("ml_experiments.db")
cursor = conn.cursor()

# Update: mark experiments as "legacy" if accuracy is below threshold
cursor.execute("""
    UPDATE experiments
    SET model_name = model_name || ' (legacy)'
    WHERE accuracy < 0.8 AND dataset = ?
""", ("CIFAR-10",))
print(f"Updated {cursor.rowcount} experiments")

# Delete: remove experiments with zero accuracy (failed runs)
cursor.execute("DELETE FROM experiments WHERE accuracy = 0")
print(f"Deleted {cursor.rowcount} failed experiments")

conn.commit()
conn.close()

#Context Managers for Database Connections

Database connections are resources that must be properly closed. If your code crashes between connect() and close(), the connection leaks. Context managers solve this.

#The Basic Pattern

pythonrunnable cell

import sqlite3

# BAD: if an error occurs, conn.close() never runs
conn = sqlite3.connect("ml_experiments.db")
cursor = conn.cursor()
cursor.execute("SELECT * FROM experiments")
results = cursor.fetchall()
conn.close()  # what if an error occurs above?

# GOOD: sqlite3 connection IS a context manager
with sqlite3.connect("ml_experiments.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM experiments")
    results = cursor.fetchall()
    # conn.commit() is called automatically if no exception
    # conn.rollback() is called automatically if exception occurs
# Connection is properly closed, even if an error occurred

Important nuance: sqlite3.connect() as a context manager calls commit() or rollback() on exit, but it does NOT close the connection. For both commit/rollback AND close, write a helper.

#A Reusable Database Helper

pythonrunnable cell

import sqlite3
from contextlib import contextmanager

@contextmanager
def get_db(db_path: str = "ml_experiments.db"):
    """Context manager that connects, commits/rolls back, AND closes."""
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row  # return dicts instead of tuples
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# Usage -- clean, safe, automatic cleanup
with get_db() as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT model_name, accuracy FROM experiments ORDER BY accuracy DESC LIMIT 3")
    for row in cursor.fetchall():
        print(f"{row['model_name']}: {row['accuracy']:.3f}")

# Even if an error occurs inside the with block:
# 1. conn.rollback() undoes any partial changes
# 2. conn.close() frees the connection
# 3. The exception propagates normally

What Do You Think?

What happens when an exception is raised inside a 'with sqlite3.connect(db) as conn:' block?

#Transaction Patterns

pythonrunnable cell

import sqlite3

def batch_insert_with_transaction(records: list[tuple]) -> int:
    """Insert records in a single transaction for performance and atomicity."""
    with sqlite3.connect("ml_experiments.db") as conn:
        cursor = conn.cursor()

        # All inserts happen in one transaction
        # If ANY insert fails, ALL are rolled back
        cursor.executemany("""
            INSERT INTO experiments 
            (model_name, dataset, accuracy, loss, epochs, learning_rate)
            VALUES (?, ?, ?, ?, ?, ?)
        """, records)

        return cursor.rowcount

# Without explicit transactions, each INSERT is auto-committed
# (much slower for bulk operations)
new_experiments = [
    ("EfficientNet-B0", "CIFAR-10", 0.941, 0.198, 100, 0.001),
    ("EfficientNet-B3", "CIFAR-10", 0.958, 0.142, 100, 0.001),
    ("EfficientNet-B7", "CIFAR-10", 0.966, 0.112, 100, 0.0005),
]

count = batch_insert_with_transaction(new_experiments)
print(f"Inserted {count} experiments in one transaction")

#SQLAlchemy ORM Basics

SQLAlchemy is Python's most popular database toolkit. The ORM (Object-Relational Mapper) lets you define database tables as Python classes and interact with data as Python objects instead of writing raw SQL.

#Defining Models

pythonreference · read-only

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, relationship, Session
from datetime import datetime, timezone

# Create the base class for all models
Base = declarative_base()

# Timezone-aware UTC default — datetime.utcnow is deprecated in Python 3.12+
def utc_now() -> datetime:
    return datetime.now(timezone.utc)

class Experiment(Base):
    """An ML experiment."""
    __tablename__ = "experiments"

    id = Column(Integer, primary_key=True, autoincrement=True)
    model_name = Column(String, nullable=False)
    dataset = Column(String, nullable=False)
    accuracy = Column(Float)
    loss = Column(Float)
    epochs = Column(Integer)
    learning_rate = Column(Float)
    created_at = Column(DateTime, default=utc_now)

    # Relationship: one experiment has many metrics
    metrics = relationship("Metric", back_populates="experiment")

    def __repr__(self) -> str:
        return (f"Experiment(id={self.id}, model={self.model_name}, "
                f"dataset={self.dataset}, acc={self.accuracy})")


class Metric(Base):
    """A metric recorded during an experiment (e.g., per-epoch loss)."""
    __tablename__ = "metrics"

    id = Column(Integer, primary_key=True, autoincrement=True)
    experiment_id = Column(Integer, ForeignKey("experiments.id"), nullable=False)
    epoch = Column(Integer, nullable=False)
    train_loss = Column(Float)
    val_loss = Column(Float)
    val_accuracy = Column(Float)

    # Back-reference to parent experiment
    experiment = relationship("Experiment", back_populates="metrics")

    def __repr__(self) -> str:
        return (f"Metric(exp={self.experiment_id}, epoch={self.epoch}, "
                f"val_acc={self.val_accuracy})")


# Create the database and tables
engine = create_engine("sqlite:///ml_tracking.db", echo=False)
Base.metadata.create_all(engine)
print("Database and tables created!")

#Creating and Querying Records

pythonreference · read-only

from sqlalchemy.orm import Session

# Create a session (like a database transaction context)
with Session(engine) as session:
    # Create an experiment with metrics
    exp = Experiment(
        model_name="ResNet50",
        dataset="CIFAR-10",
        accuracy=0.923,
        loss=0.245,
        epochs=100,
        learning_rate=0.001,
    )

    # Add per-epoch metrics
    for epoch in range(1, 6):
        metric = Metric(
            epoch=epoch,
            train_loss=1.0 / epoch,
            val_loss=1.2 / epoch,
            val_accuracy=0.5 + 0.08 * epoch,
        )
        exp.metrics.append(metric)  # SQLAlchemy handles the foreign key

    session.add(exp)
    session.commit()
    print(f"Created: {exp}")
    print(f"Metrics: {exp.metrics}")

#Querying with the ORM

pythonreference · read-only

from sqlalchemy import select, func
from sqlalchemy.orm import Session

with Session(engine) as session:
    # Simple query: all experiments with accuracy > 0.9
    stmt = select(Experiment).where(Experiment.accuracy > 0.9)
    high_acc = session.execute(stmt).scalars().all()
    print("High accuracy experiments:")
    for exp in high_acc:
        print(f"  {exp.model_name}: {exp.accuracy}")

    # Aggregation: count experiments per dataset
    stmt = (
        select(Experiment.dataset, func.count().label("count"))
        .group_by(Experiment.dataset)
    )
    for row in session.execute(stmt):
        print(f"  {row.dataset}: {row.count} experiments")

    # Join: get experiments with their metrics
    stmt = (
        select(Experiment)
        .where(Experiment.model_name == "ResNet50")
    )
    exp = session.execute(stmt).scalars().first()
    if exp:
        print(f"\n{exp.model_name} metrics:")
        for m in exp.metrics:
            print(f"  Epoch {m.epoch}: val_acc={m.val_accuracy:.3f}, "
                  f"val_loss={m.val_loss:.3f}")

#Updating and Deleting

pythonreference · read-only

from sqlalchemy.orm import Session

with Session(engine) as session:
    # Update: find an experiment and modify it
    stmt = select(Experiment).where(Experiment.model_name == "ResNet50")
    exp = session.execute(stmt).scalars().first()
    if exp:
        exp.accuracy = 0.935  # just modify the Python attribute
        exp.epochs = 120
        session.commit()  # SQLAlchemy generates the UPDATE SQL
        print(f"Updated: {exp}")

    # Delete: remove experiments below threshold
    stmt = select(Experiment).where(Experiment.accuracy < 0.5)
    low_acc = session.execute(stmt).scalars().all()
    for exp in low_acc:
        session.delete(exp)
    session.commit()
    print(f"Deleted {len(low_acc)} low-accuracy experiments")

#SQLAlchemy ORM vs Core vs Raw SQL: Which to Use?

SQLAlchemy has two distinct layers. Understanding the difference prevents confusion when reading documentation or debugging.

Layer	What it is	When to use it
sqlite3 / psycopg2	Raw DB-API driver	One-off scripts, maximum control, no abstraction needed
SQLAlchemy Core	SQL expression language (constructs SQL as Python objects)	Complex queries, bulk operations, performance-critical paths
SQLAlchemy ORM	Maps Python classes to tables, manages object identity	Domain-rich applications, relationships, cleaner CRUD code

pythonreference · read-only

# SQLAlchemy CORE: SQL as Python expressions (no ORM classes needed)
from sqlalchemy import create_engine, text, Table, MetaData, Column, Integer, String

engine = create_engine("sqlite:///ml_tracking.db")

# Core style: use text() for raw SQL with parameter binding
with engine.connect() as conn:
    result = conn.execute(
        text("SELECT model_name, accuracy FROM experiments WHERE accuracy > :threshold"),
        {"threshold": 0.9}
    )
    for row in result:
        print(row.model_name, row.accuracy)

# Core style: programmatic SQL construction (safe from injection, DB-agnostic)
from sqlalchemy import select as core_select, Table, MetaData
metadata = MetaData()
experiments_table = Table("experiments", metadata, autoload_with=engine)

stmt = (
    core_select(experiments_table.c.model_name, experiments_table.c.accuracy)
    .where(experiments_table.c.accuracy > 0.9)
    .order_by(experiments_table.c.accuracy.desc())
)
with engine.connect() as conn:
    for row in conn.execute(stmt):
        print(row)

Rule of thumb: Start with raw sqlite3 to understand SQL directly. Move to SQLAlchemy ORM when you have a domain model with relationships. Use SQLAlchemy Core for complex analytical queries or bulk operations where the ORM's object overhead is unnecessary.

#Connection Pooling

For production applications that handle concurrent requests, creating a new database connection per request is expensive. SQLAlchemy's create_engine automatically manages a connection pool — a cache of open connections that are reused across requests.

pythonreference · read-only

from sqlalchemy import create_engine

# Connection pool is configured at engine creation time
engine = create_engine(
    "postgresql://user:password@localhost/mldb",
    pool_size=5,        # maintain up to 5 connections in the pool
    max_overflow=10,    # allow up to 10 extra connections under load
    pool_timeout=30,    # wait up to 30s for a connection before raising
    pool_recycle=1800,  # recycle connections after 30min (prevents stale connections)
)

# SQLite does not support concurrent writes, so use StaticPool for testing
from sqlalchemy.pool import StaticPool
test_engine = create_engine(
    "sqlite:///:memory:",
    connect_args={"check_same_thread": False},
    poolclass=StaticPool,  # share one connection across all threads (test only)
)

For SQLite in development (single process, single file), the default pool (NullPool or single connection) is fine. For PostgreSQL or MySQL in production, always configure pool_size based on your expected concurrent load and the database's max_connections setting.

Intuition

A connection pool is like a fleet of taxis, not a single taxi.

Without pooling, every database request has to hail a new taxi (open a TCP connection, authenticate, negotiate SSL). That takes 20-100ms. With a pool, a fleet of taxis waits outside your door. A request grabs the nearest available taxi, uses it, and returns it to the fleet. The next request gets it instantly.

pool_size=5 means five taxis always idling. max_overflow=10 means you can hail up to 10 extra taxis during a traffic spike, but they are released when traffic drops. pool_timeout=30 means if all 15 taxis are busy for 30 seconds, give up and raise an error rather than waiting forever.

For SQLite, pooling is less critical because SQLite is a file, not a server — there is no network handshake cost. But for PostgreSQL, a properly sized pool is the difference between a sub-millisecond query and a 100ms connection setup overhead on every request.

Watch Out

Watch out

Fix: The ORM is optimized for domain-rich CRUD with object identity and relationships. For bulk inserts, analytical aggregations, or ETL pipelines, use SQLAlchemy Core's insert() / engine.execute() or Pandas' to_sql() with method='multi'. The ORM's per-object overhead (tracking, identity map, lazy-loading) is unnecessary overhead for pure data pipelines.

Using SQLAlchemy ORM for simple bulk queries or ETL loads and wondering why it is slow.

The ORM maintains an 'identity map' — a dictionary of all loaded objects by primary key. Every ORM query checks this map, every object change is tracked, and every relationship access may trigger a lazy-load query (the 'N+1 query problem'). For CRUD on individual objects (add experiment, get experiment by ID), this is worth the cost. For a query that loads 100,000 rows to compute statistics, you are paying ORM overhead on 100,000 Python objects when you only needed the aggregated numbers.

```python

SLOW: ORM loads 100k Python objects, tracks each one

with Session(engine) as session: all_exps = session.scalars(select(Experiment)).all() # 100k objects in memory avg_acc = sum(e.accuracy for e in all_exps) / len(all_exps) # iterate in Python

FAST: push aggregation to the database, get one number back

with engine.connect() as conn: result = conn.execute(text("SELECT AVG(accuracy) FROM experiments")).scalar() avg_acc = result ```

For read-heavy analytics: use Core or Pandas read_sql(). For write-heavy ETL: use Core's insert() with executemany() or Pandas to_sql(method='multi'). Reserve the ORM for the transactional domain logic where relationships and object state matter.

#Pandas + SQL -- The Data Science Power Combo

Pandas integrates seamlessly with SQL databases. pd.read_sql() runs a query and returns a DataFrame. df.to_sql() writes a DataFrame directly to a database table. This is the bridge between data analysis (Pandas) and data storage (SQL).

#Reading SQL into DataFrames

pythonrunnable cell

import pandas as pd
import sqlite3

conn = sqlite3.connect("ml_experiments.db")

# Read an entire table
df = pd.read_sql("SELECT * FROM experiments", conn)
print(f"Loaded {len(df)} experiments")
print(df.head())

# Read with a filter (much more efficient than loading all + filtering)
df_best = pd.read_sql("""
    SELECT model_name, dataset, accuracy, epochs, learning_rate
    FROM experiments
    WHERE accuracy > 0.9
    ORDER BY accuracy DESC
""", conn)
print(f"\nBest experiments:")
print(df_best.to_string(index=False))

# Parameterized query with Pandas
df_cifar = pd.read_sql(
    "SELECT * FROM experiments WHERE dataset = ?",
    conn,
    params=("CIFAR-10",)
)
print(f"\nCIFAR-10 experiments: {len(df_cifar)}")

conn.close()

#Writing DataFrames to SQL

pythonrunnable cell

import pandas as pd
import sqlite3

# Create a DataFrame from analysis results
results = pd.DataFrame({
    "model_name": ["MLP", "CNN", "Transformer", "RNN", "GAN"],
    "task": ["classification", "image", "text", "sequence", "generation"],
    "params_millions": [0.5, 11.7, 110.0, 2.3, 8.4],
    "training_hours": [0.1, 2.5, 48.0, 1.2, 12.0],
    "gpu_memory_gb": [0.5, 4.0, 16.0, 2.0, 8.0],
})

conn = sqlite3.connect("ml_experiments.db")

# Write DataFrame to a new table
results.to_sql(
    "model_benchmarks",      # table name
    conn,
    if_exists="replace",     # replace table if it already exists
    index=False,             # don't write the DataFrame index as a column
)
print(f"Wrote {len(results)} rows to model_benchmarks table")

# Verify it was written
df_check = pd.read_sql("SELECT * FROM model_benchmarks", conn)
print(df_check)

conn.close()

#Real-World Data Pipeline

pythonrunnable cell

import pandas as pd
import sqlite3
from contextlib import contextmanager

@contextmanager
def get_db(path: str = "analytics.db"):
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

def create_schema(conn: sqlite3.Connection) -> None:
    """Create the analytics schema."""
    conn.executescript("""
        CREATE TABLE IF NOT EXISTS raw_events (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            user_id TEXT NOT NULL,
            event_type TEXT NOT NULL,
            page TEXT,
            duration_ms INTEGER,
            timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE IF NOT EXISTS daily_stats (
            date TEXT PRIMARY KEY,
            total_events INTEGER,
            unique_users INTEGER,
            avg_duration_ms REAL,
            top_page TEXT
        );

        CREATE INDEX IF NOT EXISTS idx_events_user ON raw_events(user_id);
        CREATE INDEX IF NOT EXISTS idx_events_type ON raw_events(event_type);
    """)

def ingest_events(conn: sqlite3.Connection, events: list[dict]) -> int:
    """Ingest raw event data."""
    cursor = conn.cursor()
    cursor.executemany("""
        INSERT INTO raw_events (user_id, event_type, page, duration_ms)
        VALUES (:user_id, :event_type, :page, :duration_ms)
    """, events)
    return cursor.rowcount

def compute_daily_stats(conn: sqlite3.Connection) -> pd.DataFrame:
    """Aggregate raw events into daily stats using Pandas."""
    df = pd.read_sql("""
        SELECT
            DATE(timestamp) as date,
            COUNT(*) as total_events,
            COUNT(DISTINCT user_id) as unique_users,
            AVG(duration_ms) as avg_duration_ms
        FROM raw_events
        GROUP BY DATE(timestamp)
    """, conn)

    # Find the top page per day
    top_pages = pd.read_sql("""
        SELECT DATE(timestamp) as date, page, COUNT(*) as visits
        FROM raw_events
        GROUP BY DATE(timestamp), page
        ORDER BY visits DESC
    """, conn)

    # Get the top page for each date
    top_page_per_day = top_pages.groupby("date").first().reset_index()[["date", "page"]]
    top_page_per_day.columns = ["date", "top_page"]

    # Merge
    daily = df.merge(top_page_per_day, on="date", how="left")

    # Write back to database
    daily.to_sql("daily_stats", conn, if_exists="replace", index=False)

    return daily

# Run the pipeline
with get_db() as conn:
    create_schema(conn)

    # Simulate event ingestion
    import random
    events = [
        {
            "user_id": f"usr_{random.randint(1, 50)}",
            "event_type": random.choice(["page_view", "click", "scroll"]),
            "page": random.choice(["/home", "/lessons", "/playground", "/about"]),
            "duration_ms": random.randint(100, 30000),
        }
        for _ in range(500)
    ]

    count = ingest_events(conn, events)
    print(f"Ingested {count} events")

with get_db() as conn:
    stats = compute_daily_stats(conn)
    print("\nDaily stats:")
    print(stats.to_string(index=False))

#Code Playground & Quiz

pythonplayground.py · Pyodide

# Databases with Python — Hands-On Exploration
import sqlite3

# === SETUP: Create an in-memory database ===
conn = sqlite3.connect(":memory:")  # in-memory DB (no file created)
conn.row_factory = sqlite3.Row
cursor = conn.cursor()

# Create tables
cursor.executescript("""
  CREATE TABLE students (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      name TEXT NOT NULL,
      major TEXT,
      gpa REAL CHECK(gpa >= 0.0 AND gpa &lt;= 4.0)
  );

CREATE TABLE courses (
      id INTEGER PRIMARY KEY AUTOINCREMENT,
      name TEXT NOT NULL,
      credits INTEGER NOT NULL
  );

CREATE TABLE enrollments (
      student_id INTEGER REFERENCES students(id),
      course_id INTEGER REFERENCES courses(id),
      grade TEXT,
      PRIMARY KEY (student_id, course_id)
  );
""")

# Insert sample data
students = [
  ("Alice", "Computer Science", 3.8),
  ("Bob", "Data Science", 3.2),
  ("Charlie", "Computer Science", 3.9),
  ("Diana", "Mathematics", 3.5),
  ("Eve", "Data Science", 3.7),
]
cursor.executemany(
  "INSERT INTO students (name, major, gpa) VALUES (?, ?, ?)", students
)

courses = [
  ("Machine Learning", 4),
  ("Linear Algebra", 3),
  ("Python Programming", 3),
  ("Statistics", 4),
]
cursor.executemany(
  "INSERT INTO courses (name, credits) VALUES (?, ?)", courses
)

enrollments = [
  (1, 1, "A"), (1, 2, "A-"), (1, 3, "A"),
  (2, 1, "B+"), (2, 4, "B"),
  (3, 1, "A"), (3, 2, "A"), (3, 4, "A-"),
  (4, 2, "A"), (4, 4, "A"),
  (5, 1, "A-"), (5, 3, "B+"), (5, 4, "A"),
]
cursor.executemany(
  "INSERT INTO enrollments (student_id, course_id, grade) VALUES (?, ?, ?)",
  enrollments
)
conn.commit()

# === QUERIES ===

# 1. All students with GPA > 3.5
print("=== High GPA Students ===")
cursor.execute("SELECT name, major, gpa FROM students WHERE gpa > 3.5 ORDER BY gpa DESC")
for row in cursor.fetchall():
  print(f"  {row['name']} ({row['major']}): {row['gpa']}")

# 2. Average GPA per major
print("\n=== Average GPA by Major ===")
cursor.execute("""
  SELECT major, COUNT(*) as count, ROUND(AVG(gpa), 2) as avg_gpa
  FROM students
  GROUP BY major
  ORDER BY avg_gpa DESC
""")
for row in cursor.fetchall():
  print(f"  {row['major']}: {row['avg_gpa']} ({row['count']} students)")

# 3. JOIN: Which students are in Machine Learning?
print("\n=== Machine Learning Students ===")
cursor.execute("""
  SELECT s.name, e.grade
  FROM students s
  JOIN enrollments e ON s.id = e.student_id
  JOIN courses c ON c.id = e.course_id
  WHERE c.name = 'Machine Learning'
  ORDER BY e.grade
""")
for row in cursor.fetchall():
  print(f"  {row['name']}: {row['grade']}")

# 4. How many courses is each student taking?
print("\n=== Course Counts ===")
cursor.execute("""
  SELECT s.name, COUNT(e.course_id) as num_courses
  FROM students s
  LEFT JOIN enrollments e ON s.id = e.student_id
  GROUP BY s.id
  ORDER BY num_courses DESC
""")
for row in cursor.fetchall():
  print(f"  {row['name']}: {row['num_courses']} courses")

conn.close()

# CHALLENGES:
# 1. Add a 'semester' column to enrollments and query by semester
# 2. Find students enrolled in ALL 4 courses (hint: HAVING COUNT = 4)
# 3. Use Pandas: pd.read_sql() to load the data and compute stats
# 4. Write a parameterized search function: find_students(major, min_gpa)

Tests · Build databases, write queries, and explore SQL from Python!

Interactive Lab

Write SQL queries in your browser and see results instantly — practice the same queries you learned in Python

SQL Playground

#Key Takeaways

CSVs are for prototyping; databases are for production -- databases handle large datasets, concurrent access, data integrity, and efficient querying that CSVs simply cannot
sqlite3 is built into Python and requires zero setup -- import sqlite3 gives you a full relational database. Use it for local experiments, prototypes, and single-user applications
Always use parameterized queries -- never embed user input into SQL strings with f-strings or .format(). The ? placeholder prevents SQL injection, the most common web application vulnerability
Context managers guarantee cleanup -- wrap database connections in with statements or custom context managers to ensure connections are properly closed and transactions are committed or rolled back
SQLAlchemy ORM maps Python classes to database tables -- define models as classes, create instances as rows, and query with Python instead of raw SQL. The ORM handles the translation and makes it easy to switch databases
Pandas + SQL is the data science power combo -- pd.read_sql() loads query results directly into DataFrames for analysis, and df.to_sql() writes DataFrames back to database tables. This bridges the gap between exploration and storage

Quick Check1 / 5

Why are databases better than CSVs for large datasets?

Your Reflection

Continue Learning

Discussion

Databases with Python: SQLite, SQLAlchemy & Data Access

After this lesson, you will be able to:

Before You Start

#Why Databases for Data Science

#The Problem with CSVs

#SQL Basics in 60 Seconds

#sqlite3 -- Python's Built-In Database

#Connecting and Creating Tables

#Inserting Data

#Querying Data

#Parameterized Queries -- Preventing SQL Injection

#UPDATE and DELETE

#Context Managers for Database Connections

#The Basic Pattern

#A Reusable Database Helper

#Transaction Patterns

#SQLAlchemy ORM Basics

#Defining Models

#Creating and Querying Records

#Querying with the ORM

#Updating and Deleting

#SQLAlchemy ORM vs Core vs Raw SQL: Which to Use?

#Connection Pooling

#Pandas + SQL -- The Data Science Power Combo

#Reading SQL into DataFrames

#Writing DataFrames to SQL

#Real-World Data Pipeline

Deep Dive

#Code Playground & Quiz

#Key Takeaways