The second deploy crashed: table notes already exists

Quick recapJoins· 2 min

A JOIN matches rows of two tables on a column, in one query. notes JOIN users ON notes.owner_id = users.id gives each note with its owner. An ORM writes joins like this for you when you ask for related objects.

Quick recapClasses and objects· 2 min

A class describes a kind of thing, with fields and methods; an object is one of them. A @dataclass writes __init__ for you: Note(id=1, title="Milk"). An ORM uses one class per table and one object per row.

See the problem

Our notes app keeps its table changes as SQL files, one per change. A deploy script runs them all:

1: CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT)
2: ALTER TABLE notes ADD COLUMN pinned INTEGER NOT NULL DEFAULT 0

The first deploy works. The second deploy, with only a small code fix, crashes with table notes already exists. Someone comments out the first file “just for now”. A week later, a new laptop builds an empty database from the same files, and it has no notes table at all.

Define the task

Write migrate(db, migrations): bring a database up to date.

• Input: a sqlite3 connection, and a list of (version, statements) pairs, in any order.
• Output: the versions it ran, in order.
• The rules: run each migration once, ever, oldest first. Remember what ran in a table, schema_version, inside the same database.
• The hard part: a migration can fail half-way. It must then leave no trace, so a fixed version can run cleanly.

Example: on a new database, migrate(db, MIGS) gives [1, 2]. Running it again gives [].

Pause and think

⏸ Pause and think

Before you read on: the script can’t tell which files already ran. Where should that fact live so every server, and every new laptop, sees the same answer?

Give it two minutes with pen and paper before you open a hint.

Hints open one at a time:

Hint 1 of 3

Not in a file on one server, and not in someone’s head.

Hint 2 of 3

In the database itself: the thing that was changed knows what was done to it.

Hint 3 of 3

A small table with one row per version that ran. Read it first, run the rest, and write each new version down.

Understand the idea

An ORM (Object-Relational Mapper) maps tables to classes, and rows to objects. You write Note.objects.filter(owner=user) in Django, or select(Note).where(Note.owner_id == user.id) in SQLAlchemy. The ORM writes the SQL, with parameters, sends it, and turns each row back into a Note. It saves typing and stops injection. It also hides how many queries run: reading note.owner in a loop can send one query per note. That is the N+1 problem, with its own lesson: The order list page is slow.

A migration is one numbered change to the tables. The code says what the tables should look like; the migrations say how to get there from any older version. Each tool keeps a table of what ran:

Tool	Make a migration	Run them	Its table
Django	manage.py makemigrations	manage.py migrate	django_migrations
SQLAlchemy	alembic revision --autogenerate	alembic upgrade head	alembic_version
Flask	Flask-Migrate wraps Alembic	flask db upgrade	alembic_version

Run each migration in a transaction. PostgreSQL and SQLite can undo table changes inside a transaction, so a failed migration leaves nothing behind. MySQL can’t: each ALTER TABLE saves at once. There, keep one change per migration.

Some changes break the running app. During a deploy, old code and new code run side by side for a while. So:

• Renaming a column breaks the old code. Do it in steps: add the new column, write to both, copy the old values, read the new one, and drop the old one in a later deploy. This is called expand and contract.
• Adding a NOT NULL column without a default fails if the table has rows, and on big tables may lock it while every row is rewritten.
• Dropping a column loses its data, with no way back. Stop using it in the code first, and drop it a release later.

Four deploys and a version table

⤢ Full view

Deploy 1: a new database

Parts:

• New code + migrations (the request is here)
• Migration runner
• schema_version table
• notes table

Arrows:

• New code + migrations sends requests to Migration runner
• Migration runner sends requests to schema_version table
• Migration runner sends requests to notes table

Step 1 of 7. The code ships with two migrations: 1 creates the notes table, 2 adds a pinned column.

◀ BackPlayNext ▶Reset Ask me first Show as text

★The database remembers what was done to it

A migration tool reads a version table, runs only what is missing, oldest first, and writes each version down in the same transaction as the change. Then every server, test run and laptop agrees on the schema.

Climb the ladder

Three versions of migrate. The tests deploy twice, give the list out of order, and break a migration half-way.

Okay

Run every migration, every time

The idea: Sort the migrations, and run all their statements.

It builds a new database from nothing, which is all a first deploy needs.

import sqlite3


def migrate(db: sqlite3.Connection, migrations: list[tuple[int, list[str]]]) -> list[int]:
    ran = []
    for version, statements in sorted(migrations):
        for sql in statements:
            db.execute(sql)  # runs every migration, every deploy
        ran.append(version)
    db.commit()
    return ran

Second deploy

Crashes

Failed migration

Half done

Why

Nothing records what ran, so it all runs again.

Where it breaks: the second deploy runs CREATE TABLE notes again, and the table already exists.

Better

A version table

The idea: Read the versions that ran; run only the others; write each one down after it runs.

The waste in Okay: it forgot everything between deploys.

Highlighted lines are new in this rung.

import sqlite3


def migrate(db: sqlite3.Connection, migrations: list[tuple[int, list[str]]]) -> list[int]:
    db.execute("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)")
    done = {v for (v,) in db.execute("SELECT version FROM schema_version")}
    ran = []
    for version, statements in sorted(migrations):
        if version in done:
            continue
        for sql in statements:
            db.execute(sql)
            db.commit()  # each step is saved as soon as it runs
        db.execute("INSERT INTO schema_version VALUES (?)", (version,))
        db.commit()
        ran.append(version)
    return ran

Second deploy

Runs nothingwas Crashes

Failed migration

Half done, stuckwas Half done

Why

The database now says which versions it has.

What changed from the previous rung

import sqlite3


def migrate(db: sqlite3.Connection, migrations: list[tuple[int, list[str]]]) -> list[int]:
    db.execute("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)")
    done = {v for (v,) in db.execute("SELECT version FROM schema_version")}
    ran = []
    for version, statements in sorted(migrations):
        if version in done:
            continue
        for sql in statements:
            db.execute(sql)  # runs every migration, every deploy
            db.execute(sql)
            db.commit()  # each step is saved as soon as it runs
        db.execute("INSERT INTO schema_version VALUES (?)", (version,))
        db.commit()
        ran.append(version)
    db.commit()
    return ran

Where it breaks: each statement is saved as it runs. If step 2 of a migration fails, step 1 stays, but the version isn’t written. The next try runs step 1 again and fails: “duplicate column”. Now someone fixes the database by hand.

Best

One transaction per migration

The idea: BEGIN, run the statements, write the version, COMMIT. On any error, ROLLBACK and stop.

The waste in Better: it saved each step alone, so the database could stop between them.

db.isolation_level = None turns off Python’s own transaction handling, so our BEGIN and COMMIT are the only ones.

Highlighted lines are new in this rung.

import sqlite3


def migrate(db: sqlite3.Connection, migrations: list[tuple[int, list[str]]]) -> list[int]:
    db.isolation_level = None  # we say where each transaction starts and ends
    db.execute("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)")
    done = {v for (v,) in db.execute("SELECT version FROM schema_version")}
    ran = []
    for version, statements in sorted(migrations):
        if version in done:
            continue
        db.execute("BEGIN")
        try:
            for sql in statements:
                db.execute(sql)
            db.execute("INSERT INTO schema_version VALUES (?)", (version,))
            db.execute("COMMIT")
        except Exception:
            db.execute("ROLLBACK")  # all of this migration, or none of it
            raise
        ran.append(version)
    return ran

Second deploy

Runs nothing

Failed migration

Undonewas Half done, stuck

Why

The change and its version are saved together, or not at all.

What changed from the previous rung

import sqlite3


def migrate(db: sqlite3.Connection, migrations: list[tuple[int, list[str]]]) -> list[int]:
    db.isolation_level = None  # we say where each transaction starts and ends
    db.execute("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER PRIMARY KEY)")
    done = {v for (v,) in db.execute("SELECT version FROM schema_version")}
    ran = []
    for version, statements in sorted(migrations):
        if version in done:
            continue
        for sql in statements:
            db.execute(sql)
            db.commit()  # each step is saved as soon as it runs
        db.execute("INSERT INTO schema_version VALUES (?)", (version,))
        db.commit()
        db.execute("BEGIN")
        try:
            for sql in statements:
                db.execute(sql)
            db.execute("INSERT INTO schema_version VALUES (?)", (version,))
            db.execute("COMMIT")
        except Exception:
            db.execute("ROLLBACK")  # all of this migration, or none of it
            raise
        ran.append(version)
    return ran

When Best isn’t best: use your framework’s tool. Django and Alembic also store the code for each change, work out the next migration from your models, and can go back a version. With several servers, run migrations once, as a deploy step, not on every server’s start.

Write it yourself

Write migrate(db, migrations) from the task. Press Run tests to check your code.

• Create schema_version (version INTEGER PRIMARY KEY) if it is missing, and read what ran.
• Go through the migrations by version. Skip the ones that ran.
• Run each one, and write its version, in one transaction. If anything fails, roll back and raise the error again.

Run each migration once, all or nothing

Ctrl + Enter runs

import sqlite3 def migrate(db: sqlite3.Connection, migrations: list[tuple[int, list[str]]]) -> list[int]: ran = [] for version, statements in sorted(migrations): for sql in statements: db.execute(sql) ran.append(version) db.commit() return ran

Run testsReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: A new database, migrations 1 and 2 → [1, 2]
2. ○Not run yet: Deploy twice: the second run → []
3. ○Not run yet: The list given out of order → [1, 2]
4. ○Not run yet: A database already at version 1 → [2]
5. ○Not run yet: At version 1, migration 2 fails at its second step: what is left? → [['id', 'body'], [1]]
6. ○Not run yet: Extra test
7. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

migrate()Run onceVisualize

Compare

Way	Second deploy	A migration fails half-way	Good	Not so good	Use it when
Okay: run everything	Crashes	Half done	Simple	Works once	A throwaway database you rebuild each time
Better: version table	Runs nothing	Half done, every retry fails	Each runs once	Needs fixing by hand after a failure	A database that can’t roll back table changes (MySQL), with one change per migration
Best: a transaction each	Runs nothing	Undone	Safe to retry	A little more code; the tools do it for you	PostgreSQL and SQLite: always

Common mistakes and edge cases

!Never edit a migration that already ran somewhere

Its version is written down on every server that ran it, so your change never runs there. Production and your laptop then disagree. Write a new migration instead.

• Two branches, two migrations with the same number. Django spots this and asks for a merge migration; Alembic shows two heads. The first practice problem catches it.
• Migrations that import your models break when the model changes later. Django gives each migration a frozen copy (apps.get_model); use it in data migrations.
• Big tables. Adding an index or rewriting every row can lock a table for minutes. In PostgreSQL, use CREATE INDEX CONCURRENTLY, and copy data in batches.
• Running migrations on every server’s start. Two servers race to run the same one. Run them once, as a step of the deploy.
• SELECT * in hand-written code. The column order changes with migrations. Name the columns, as the second practice problem shows.

Say it in the interview

Questions to ask first

• Which database: does it roll back table changes?
• How big are the tables we’re changing, and can we take downtime?
• Do old and new code run side by side during a deploy?

A 60-second walkthrough

“An ORM maps tables to classes and rows to objects, and builds parameterized SQL for me, though I watch the query count for N+1. Schema changes go in migrations: numbered files, run in order, once each. The tool keeps a version table in the database, so every environment agrees. I’d run each migration in a transaction where the database allows it, so a failure leaves nothing half done. For changes that break running code, like a rename, I use expand and contract. Add the new column, write both, backfill in batches, switch reads, and drop the old column a release later.”

Likely follow-ups

• ORM or raw SQL? ORM for most work: it is safe and quick to write. Raw SQL, with parameters, for a report the ORM can’t say well. Read the SQL the ORM sends either way.
• How do you add a NOT NULL column to a big table? Add it as nullable or with a default, backfill in batches, then add the constraint.
• How do you roll back a bad migration? Write a new one that undoes it. A down migration helps in development, but in production data may already depend on the change.
• What is lazy loading? The ORM fetches related rows only when you first touch them. Useful, and the cause of N+1.

Practice

Three problems, from easy to medium. Each has hints, tests you can run here, and its own Okay, Better and Best answers.

1. Which migrations are left? (easy)

Write pending(applied, available). Both are lists of names like "0002_pinned", which sort in order. Give back the available ones not yet applied, in order. Raise ValueError in two cases:

• The database has a migration the code doesn’t know. Old code is running against a newer database. Name the missing ones in the message.
• A pending one is older than the newest applied one. Two branches added migrations side by side. Say “older” in the message.

Example: pending(["0001_initial"], ["0002_pinned", "0001_initial"]) gives ["0002_pinned"].

Hint 1 of 3

Sets make “not yet applied” and “unknown” one line each.

Hint 2 of 3

sorted() gives the order, because the names start with a number padded with zeros.

Hint 3 of 3

After sorting, compare the first pending name with max(applied).

Which migrations are left?

Ctrl + Enter runs

def pending(applied: list[str], available: list[str]) -> list[str]: return [name for name in available if name not in applied]

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: pending(['0001_initial'], ['0001_initial', '0002_pinned', '0003_tags']) → ['0002_pinned', '0003_tags']
2. ○Not run yet: Given out of order → ['0001_initial', '0002_pinned']
3. ○Not run yet: The database is ahead of the code → raises ValueError
4. ○Not run yet: 0002 arrived after 0003 ran → raises ValueError
5. ○Not run yet: Extra test
6. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

pending()Run onceVisualize

Show the Okay, Better and Best answers

Okay

Not applied yet

The idea: Keep the available names that aren’t in applied.

Right when the list comes in order and nobody branches.

def pending(applied: list[str], available: list[str]) -> list[str]:
    return [name for name in available if name not in applied]

Order

As given

Catches

Nothing

Why

It answers the question, if everything else is right.

Where it breaks: files read from a folder come in any order. And name not in applied scans a list each time.

Better

In order, with a set

The idea: Put applied in a set, and sort what is left.

The waste in Okay: it trusted the order it was given.

Highlighted lines are new in this rung.

def pending(applied: list[str], available: list[str]) -> list[str]:
    done = set(applied)
    return sorted(name for name in available if name not in done)

Order

Sortedwas As given

Catches

Nothing

Why

Migrations always run oldest first.

What changed from the previous rung

def pending(applied: list[str], available: list[str]) -> list[str]:
    return [name for name in available if name not in applied]
    done = set(applied)
    return sorted(name for name in available if name not in done)

Where it breaks: old code against a newer database, or two branches’ migrations, go through silently. The schema ends up different from what any code expects.

Best

Sorted, and refuse a history that doesn’t fit

The idea: Raise if applied has names the code lacks, or if a pending one is older than the newest applied.

The waste in Better: it assumed the history made sense.

Highlighted lines are new in this rung.

def pending(applied: list[str], available: list[str]) -> list[str]:
    done = set(applied)
    unknown = done - set(available)
    if unknown:
        raise ValueError(f"the database has migrations this code lacks: {sorted(unknown)}")
    todo = sorted(name for name in available if name not in done)
    if todo and done and todo[0] < max(done):
        raise ValueError(f"{todo[0]} is older than {max(done)}, which already ran")
    return todo

Order

Sorted

Catches

Old code, crossed brancheswas Nothing

Why

It stops before running anything, with a message that says what is wrong.

What changed from the previous rung

def pending(applied: list[str], available: list[str]) -> list[str]:
    done = set(applied)
    return sorted(name for name in available if name not in done)
    unknown = done - set(available)
    if unknown:
        raise ValueError(f"the database has migrations this code lacks: {sorted(unknown)}")
    todo = sorted(name for name in available if name not in done)
    if todo and done and todo[0] < max(done):
        raise ValueError(f"{todo[0]} is older than {max(done)}, which already ran")
    return todo

When Best isn’t best: Django and Alembic store which migration each one follows, not just a number, so they can merge two branches. Use theirs; the fix is a merge migration.

2. Rows into objects (medium)

Write load_notes(db), a tiny ORM. Read every row of notes in id order and give back a Note for each. Match columns by name, not by position, because a migration may have put them in another order. pinned is stored as 0 or 1; make it a real bool.

Example: a row id=1, title="Milk", pinned=1 becomes Note(id=1, title="Milk", pinned=True).

Hint 1 of 3

Name the columns in the SELECT, so their order is yours.

Hint 2 of 3

cursor.description lists the columns; column[0] is each name.

Hint 3 of 3

dict(zip(names, row)) gives fields by name; Note(**fields) builds the object.

Rows into objects

Ctrl + Enter runs

import sqlite3 from dataclasses import dataclass @dataclass class Note: id: int title: str pinned: bool def load_notes(db: sqlite3.Connection) -> list[Note]: rows = db.execute("SELECT * FROM notes ORDER BY id").fetchall() return [Note(row[0], row[1], row[2]) for row in rows]

Run testsReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: Columns id, title, pinned → [[1, 'Milk', True], [2, 'Eggs', False]]
2. ○Not run yet: Columns id, pinned, title → [[1, 'Milk', True], [2, 'Eggs', False]]
3. ○Not run yet: The type of pinned → ['bool', 'bool']
4. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

load_notes()Run onceVisualize

Show the Okay, Better and Best answers

Okay

SELECT * by position

The idea: Read every column, and pass row[0], row[1], row[2].

It works on the table it was written for.

import sqlite3
from dataclasses import dataclass


@dataclass
class Note:
    id: int
    title: str
    pinned: bool


def load_notes(db: sqlite3.Connection) -> list[Note]:
    rows = db.execute("SELECT * FROM notes ORDER BY id").fetchall()
    return [Note(row[0], row[1], row[2]) for row in rows]  # by position

Columns moved

Wrong fields

pinned

0 or 1

Why

It assumes the table’s column order never changes.

Where it breaks: in a table built as id, pinned, title, the title becomes 1. No error, just wrong data.

Better

Name the columns

The idea: SELECT id, title, pinned, so the order is ours.

The waste in Okay: it let the table decide the order.

Highlighted lines are new in this rung.

import sqlite3
from dataclasses import dataclass


@dataclass
class Note:
    id: int
    title: str
    pinned: bool


def load_notes(db: sqlite3.Connection) -> list[Note]:
    rows = db.execute("SELECT id, title, pinned FROM notes ORDER BY id").fetchall()
    return [Note(*row) for row in rows]  # by name in the SQL, by position in Note

Columns moved

Rightwas Wrong fields

pinned

0 or 1

Why

The SQL now says which column goes where.

What changed from the previous rung

import sqlite3
from dataclasses import dataclass


@dataclass
class Note:
    id: int
    title: str
    pinned: bool


def load_notes(db: sqlite3.Connection) -> list[Note]:
    rows = db.execute("SELECT * FROM notes ORDER BY id").fetchall()
    return [Note(row[0], row[1], row[2]) for row in rows]  # by position
    rows = db.execute("SELECT id, title, pinned FROM notes ORDER BY id").fetchall()
    return [Note(*row) for row in rows]  # by name in the SQL, by position in Note

Where it breaks: pinned is an int. Code like if note.pinned is True fails, and JSON shows 1 instead of true.

Best

Match by name, and convert types

The idea: Zip the column names from cursor.description with each row, turn pinned into a bool, and build Note(**fields).

The waste in Better: it moved values, but not their meaning.

Highlighted lines are new in this rung.

import sqlite3
from dataclasses import dataclass


@dataclass
class Note:
    id: int
    title: str
    pinned: bool


def load_notes(db: sqlite3.Connection) -> list[Note]:
    cursor = db.execute("SELECT id, title, pinned FROM notes ORDER BY id")
    names = [col[0] for col in cursor.description]
    notes = []
    for row in cursor:
        fields = dict(zip(names, row, strict=True))
        fields["pinned"] = bool(fields["pinned"])  # SQLite stores booleans as 0 and 1
        notes.append(Note(**fields))
    return notes

Columns moved

Right

pinned

True or Falsewas 0 or 1

Why

Names link columns to fields, and each field gets its Python type: what an ORM does for you.

What changed from the previous rung

import sqlite3
from dataclasses import dataclass


@dataclass
class Note:
    id: int
    title: str
    pinned: bool


def load_notes(db: sqlite3.Connection) -> list[Note]:
    rows = db.execute("SELECT id, title, pinned FROM notes ORDER BY id").fetchall()
    return [Note(*row) for row in rows]  # by name in the SQL, by position in Note
    cursor = db.execute("SELECT id, title, pinned FROM notes ORDER BY id")
    names = [col[0] for col in cursor.description]
    notes = []
    for row in cursor:
        fields = dict(zip(names, row, strict=True))
        fields["pinned"] = bool(fields["pinned"])  # SQLite stores booleans as 0 and 1
        notes.append(Note(**fields))
    return notes

When Best isn’t best: this is the start of an ORM, so use a real one. Django’s BooleanField and SQLAlchemy’s Boolean do the conversion, and much more. For read-only reports, sqlite3.Row gives rows by name with no class at all.

3. Flag a risky migration (medium)

Write risks(sql). It gives back the risks found in one migration statement, in this order:

• "drops data" for DROP COLUMN or DROP TABLE.
• "renames" for any RENAME.
• "not null without default" for ADD COLUMN ... NOT NULL with no DEFAULT.

Ignore case and extra spaces. A column named dropped_at is not a drop, and NOT NULL in a CREATE TABLE is fine.

Example: risks("ALTER TABLE notes ADD COLUMN owner INTEGER NOT NULL") gives ["not null without default"].

Hint 1 of 3

" ".join(sql.upper().split()) makes case and spacing the same everywhere.

Hint 2 of 3

In a regular expression, \b marks the edge of a word, so \bDROP COLUMN\b won’t match DROPPED_AT.

Hint 3 of 3

Only flag NOT NULL when the statement adds a column and has no DEFAULT.

Flag a risky migration

Ctrl + Enter runs

import re def risks(sql: str) -> list[str]: return []

Run testsVisualizeReset codeLoads about 6 MB the first time. Best on Wi-Fi.

1. ○Not run yet: risks('ALTER TABLE notes ADD COLUMN owner INTEGER NOT NULL') → ['not null without default']
2. ○Not run yet: risks('ALTER TABLE notes ADD COLUMN pinned INTEGER NOT NULL DEFAULT 0') → []
3. ○Not run yet: risks('ALTER TABLE notes DROP COLUMN body') → ['drops data']
4. ○Not run yet: A column named dropped_at → []
5. ○Not run yet: Extra test
6. ○Not run yet: Extra test
7. ○Not run yet: Extra test

Extra tests check tricky cases. You see their input only if they fail.

Try your own input

Arguments, separated by commas as in Python

risks()Run onceVisualize

Show the Okay, Better and Best answers

Okay

Look for “drop”

The idea: Flag any statement that holds the letters drop.

It catches DROP COLUMN.

def risks(sql: str) -> list[str]:
    found = []
    if "drop" in sql.lower():
        found.append("drops data")
    return found

False alarms

dropped_at

Missed

Renames, NOT NULL

Why

It knows one risk, by its letters.

Where it breaks: a column named dropped_at sets it off, and renames and NOT NULL get through.

Better

Three phrases

The idea: Look for DROP COLUMN or TABLE, RENAME, and NOT NULL in upper case.

The waste in Okay: it looked at letters, not phrases.

Highlighted lines are new in this rung.

def risks(sql: str) -> list[str]:
    text = sql.upper()
    found = []
    if "DROP COLUMN" in text or "DROP TABLE" in text:
        found.append("drops data")
    if "RENAME" in text:
        found.append("renames")
    if "NOT NULL" in text:
        found.append("not null without default")
    return found

False alarms

Any NOT NULLwas dropped_at

Missed

Odd spacingwas Renames, NOT NULL

Why

It checks all three risks.

What changed from the previous rung

def risks(sql: str) -> list[str]:
    text = sql.upper()
    found = []
    if "drop" in sql.lower():
    if "DROP COLUMN" in text or "DROP TABLE" in text:
        found.append("drops data")
    if "RENAME" in text:
        found.append("renames")
    if "NOT NULL" in text:
        found.append("not null without default")
    return found

Where it breaks: NOT NULL DEFAULT 0 is safe, and so is NOT NULL in a new table, but both are flagged. And drop column with extra spaces gets through.

Best

Normalize, then match whole words

The idea: Upper-case and squash the spaces, then use word-edge patterns, and only flag NOT NULL on ADD COLUMN without DEFAULT.

The waste in Better: each rule was wider or narrower than the risk.

Highlighted lines are new in this rung.

import re


def risks(sql: str) -> list[str]:
    text = " ".join(sql.upper().split())  # one space between words
    found = []
    if re.search(r"\bDROP (COLUMN|TABLE)\b", text):
        found.append("drops data")
    if re.search(r"\bRENAME\b", text):
        found.append("renames")
    if re.search(r"\bADD COLUMN\b.*\bNOT NULL\b", text) and "DEFAULT" not in text:
        found.append("not null without default")
    return found

False alarms

None in the taskwas Any NOT NULL

Missed

None in the taskwas Odd spacing

Why

Each rule says exactly what makes the change risky.

What changed from the previous rung

import re


def risks(sql: str) -> list[str]:
    text = sql.upper()
    text = " ".join(sql.upper().split())  # one space between words
    found = []
    if "DROP COLUMN" in text or "DROP TABLE" in text:
    if re.search(r"\bDROP (COLUMN|TABLE)\b", text):
        found.append("drops data")
    if "RENAME" in text:
    if re.search(r"\bRENAME\b", text):
        found.append("renames")
    if "NOT NULL" in text:
    if re.search(r"\bADD COLUMN\b.*\bNOT NULL\b", text) and "DEFAULT" not in text:
        found.append("not null without default")
    return found

When Best isn’t best: text rules can’t read SQL fully: a comment or a string can fool them. Tools like django-migration-linter and squawk parse the SQL, and know each database’s locks. Run one in CI.

Check yourself

A few quick questions. Ones you miss are remembered, so you can find your weak spots later.

1. Multiple choiceHow does Django know which migrations already ran on a database?

   It compares the models with the tablesIt reads a file on the serverIt reads the django_migrations table in that databaseIt runs them all and ignores errors

   Check answer
2. Spot the bugA migration ran in production last week. What is wrong with this fix?

   # Edit 0004_add_owner.py, which already ran, to add a default:
   migrations.AddField("note", "owner", models.IntegerField(default=0))

   Editing a migration that ran: production never runs it againIntegerField can’t have a defaultMigrations can’t add fieldsNothing

   Check answer
3. Pick the patternYou must rename the column body to text while old and new code both run during a deploy. What is safest?

   One migration: RENAME COLUMN body TO textDrop body and add textRename it by hand at nightAdd text, write both, copy the data, read text, and drop body a release later

   Check answer
4. Multiple choiceA migration has two steps, and the second fails. With one transaction per migration, what is left?

   The first step, and the version written downNothing: the first step is undone tooThe first step, but no versionA locked table

   Check answer
5. Fill in the answerAn ORM sends one query for the list, then one more for each row’s related object. This is the N+____ problem.

   Your answerCheck answer

Flashcards

Add these to your review deck. Each card comes back a little later every time you get it right.

• Card 1What does an ORM do?Show answer
• Card 2What is a migration?Show answer
• Card 3How does a migration tool know what ran?Show answer
• Card 4Why one transaction per migration?Show answer
• Card 5What is expand and contract?Show answer
• Card 6How do you add a NOT NULL column to a big table?Show answer

Add 6 cards to my review deck

Cheat sheet

• ORM: classes are tables, objects are rows; it writes parameterized SQL. Watch for N+1.
• Migrations: numbered, run in order, once each; a version table in the database says what ran.
• One transaction per migration (PostgreSQL, SQLite). MySQL can’t roll back ALTER TABLE.
• Never edit a migration that ran; write a new one.
• Rename and drop in steps (expand and contract); add NOT NULL with a default or after a backfill.

Go deeper

• Django: migrations and making queries.
• SQLAlchemy: the ORM quick start. Alembic: tutorial.
• Next on the Frameworks roadmap: Flask, where these pieces meet in a small app.

Can you do these now?

If you’re not sure of any of these, scroll back to that part before moving on.

• Explain what an ORM does, and what SQL it sends for you
• Explain how a migration tool knows what has already run
• Run each migration once, in order, inside a transaction
• Spot risky schema changes, and make them safely in steps

✓Mark as completeI already know this

Status: Not started