Databases with SQLite & SQLAlchemy

Databases with SQLite & SQLAlchemy

Most real-world applications need to store data permanently. This chapter covers working with SQLite (a lightweight file-based database) directly, and then using SQLAlchemy — the most popular Python ORM — to interact with databases using Python objects.

Why This Chapter Matters

Whether you are building a web app, a CLI tool, or a data pipeline, databases are the backbone of persistent storage. Knowing both raw SQL and an ORM gives you flexibility to choose the right tool for each situation.

SQL Basics Refresher

SQL (Structured Query Language) is used to interact with relational databases.

-- Create a table
CREATE TABLE users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    email TEXT UNIQUE NOT NULL,
    score REAL DEFAULT 0
);

-- Insert data
INSERT INTO users (name, email) VALUES ('Asha', 'asha@example.com');

-- Query data
SELECT * FROM users WHERE score > 80 ORDER BY score DESC;

-- Update
UPDATE users SET score = 95 WHERE name = 'Asha';

-- Delete
DELETE FROM users WHERE id = 1;

SQLite with Python's Built-in sqlite3

SQLite is a lightweight, file-based database that requires no server. Python ships with sqlite3 built-in.

import sqlite3

# Connect (creates file if it doesn't exist)
conn = sqlite3.connect("students.db")
cursor = conn.cursor()

# Create table
cursor.execute("""
    CREATE TABLE IF NOT EXISTS students (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        score REAL NOT NULL
    )
""")

# Insert data
cursor.execute("INSERT INTO students (name, score) VALUES (?, ?)", ("Asha", 95))
cursor.execute("INSERT INTO students (name, score) VALUES (?, ?)", ("Leo", 88))

# Commit the transaction
conn.commit()

# Query
cursor.execute("SELECT * FROM students ORDER BY score DESC")
rows = cursor.fetchall()

for row in rows:
    print(row)

# Close connection
conn.close()

Using with for Safe Connections

with sqlite3.connect("students.db") as conn:
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM students")
    for row in cursor.fetchall():
        print(row)
# connection is committed and closed automatically

Using Row Factory for Dict-like Access

with sqlite3.connect("students.db") as conn:
    conn.row_factory = sqlite3.Row   # rows behave like dicts
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM students")
    for row in cursor.fetchall():
        print(row["name"], row["score"])

Parameterized Queries (Prevent SQL Injection)

Always use ? placeholders, never format strings directly:

# BAD — vulnerable to SQL injection
cursor.execute(f"SELECT * FROM users WHERE name = '{name}'")

# GOOD — parameterized
cursor.execute("SELECT * FROM users WHERE name = ?", (name,))

SQLAlchemy — The Python ORM

SQLAlchemy lets you interact with databases using Python objects instead of raw SQL. It supports SQLite, PostgreSQL, MySQL, and more.

Installing

pip install sqlalchemy

Defining Models

from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.orm import DeclarativeBase, Session

# Create engine (SQLite for this example)
engine = create_engine("sqlite:///school.db", echo=True)

class Base(DeclarativeBase):
    pass

class Student(Base):
    __tablename__ = "students"

    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(100), nullable=False)
    email = Column(String(200), unique=True, nullable=False)
    score = Column(Float, default=0.0)

    def __repr__(self):
        return f"Student(name={self.name!r}, score={self.score})"

# Create all tables
Base.metadata.create_all(engine)

Creating Records

with Session(engine) as session:
    asha = Student(name="Asha", email="asha@example.com", score=95)
    leo = Student(name="Leo", email="leo@example.com", score=88)
    
    session.add(asha)
    session.add(leo)
    session.add_all([
        Student(name="Mina", email="mina@example.com", score=92)
    ])
    session.commit()

Querying Records

with Session(engine) as session:
    # Get all
    students = session.query(Student).all()

    # Filter
    top_students = session.query(Student).filter(Student.score >= 90).all()

    # Order
    sorted_students = session.query(Student).order_by(Student.score.desc()).all()

    # Single record
    asha = session.query(Student).filter_by(name="Asha").first()
    by_id = session.get(Student, 1)

    for student in top_students:
        print(student)

Updating Records

with Session(engine) as session:
    student = session.query(Student).filter_by(name="Leo").first()
    student.score = 95
    session.commit()

Deleting Records

with Session(engine) as session:
    student = session.query(Student).filter_by(name="Mina").first()
    session.delete(student)
    session.commit()

Relationships (Foreign Keys)

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship

class Course(Base):
    __tablename__ = "courses"
    id = Column(Integer, primary_key=True)
    title = Column(String, nullable=False)
    students = relationship("Enrollment", back_populates="course")

class Enrollment(Base):
    __tablename__ = "enrollments"
    id = Column(Integer, primary_key=True)
    student_id = Column(Integer, ForeignKey("students.id"))
    course_id = Column(Integer, ForeignKey("courses.id"))
    student = relationship("Student")
    course = relationship("Course", back_populates="students")

Environment Variables for Database URLs

Never hard-code database credentials. Use environment variables:

import os
from dotenv import load_dotenv
from sqlalchemy import create_engine

load_dotenv()

DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///default.db")
engine = create_engine(DATABASE_URL)

.env file:

DATABASE_URL=postgresql://user:password@localhost:5432/mydb

Common Database Operations Pattern

def get_student(student_id: int):
    with Session(engine) as session:
        return session.get(Student, student_id)

def create_student(name: str, email: str, score: float = 0.0):
    with Session(engine) as session:
        student = Student(name=name, email=email, score=score)
        session.add(student)
        session.commit()
        session.refresh(student)
        return student

def update_score(student_id: int, new_score: float):
    with Session(engine) as session:
        student = session.get(Student, student_id)
        if not student:
            raise ValueError(f"Student {student_id} not found")
        student.score = new_score
        session.commit()

Common Mistakes

  • forgetting to commit() after INSERT/UPDATE/DELETE
  • using session outside of a with block (risk of unclosed sessions)
  • constructing SQL strings with f-strings (SQL injection risk)
  • not using foreign keys and missing cascading behavior
  • running migrations manually without a migration tool (use Alembic for SQLAlchemy)

Mini Exercises

  1. Create a SQLite database with sqlite3 that stores books (title, author, year).
  2. Insert 5 books and query all books published after 2000.
  3. Using SQLAlchemy, define a Product model with name, price, and category.
  4. Write functions to create, read, update, and delete products (CRUD).
  5. Add a Category model and set up a one-to-many relationship with Product.

Review Questions

  1. What is the difference between SQLite and PostgreSQL?
  2. Why must you always use parameterized queries instead of string formatting?
  3. What is an ORM and what problem does it solve?
  4. What does session.commit() do in SQLAlchemy?
  5. How do you define a relationship between two models in SQLAlchemy?

Reference Checklist

  • I can create and query SQLite databases using sqlite3
  • I use parameterized queries to prevent SQL injection
  • I can define SQLAlchemy models with columns and relationships
  • I can perform CRUD operations with SQLAlchemy Sessions
  • I know how to load database URLs from environment variables
PreviousNext Chapter