Importing Content

Content Migration

Export the old PHP CMS database, transform it to the new schema, import it into FastAPI, and verify every route resolves correctly

Chapter 9 of 10 SSH Tunnel Idempotent Script Alembic

Chapter Milestone

By the end of this chapter all content from the live osztromok.com MySQL database is in the new FastAPI schema. A verification script hits every URL in the site and reports any 404s or errors. The new app is content-complete and ready for Chapter 10's deployment.

1 Two types of migration — content vs schema

Before writing any code, it's worth being precise about terminology because "migration" means two different things in the FastAPI/SQLAlchemy world:

This chapter

Content migration

Moving rows of data from the old PHP CMS database into the new FastAPI database. A one-time Python script.

Section 8 (intro)

Schema migration

Evolving the database structure (adding columns, renaming tables) after the app is live. Handled by Alembic.

Chapter 10

Deployment

Switching the live server from Apache + PHP to nginx + FastAPI. The final step after migration is verified.

This chapter focuses exclusively on content migration. The plan: connect to the old database (via SSH tunnel if needed), read the existing data, transform column names and generate URL slugs where missing, then insert everything into the new schema using the SQLAlchemy ORM you built in Chapter 2.

2 Inspect the old database schema first

Run this against your old database before writing any migration code. It shows you the exact column names and types — your migration script must match them precisely.

MySQL — discover the schema

## Connect to the old database (replace with your real credentials)
mysql -u your_user -p your_old_db

## List all tables
SHOW TABLES;

## Inspect each table's columns — run these one by one
DESCRIBE subjects;
DESCRIBE subtopics;
DESCRIBE pages;
DESCRIBE page_content;

## See a sample of actual rows to spot encoding or quirk issues
SELECT * FROM subjects ORDER BY sort_order LIMIT 5;
SELECT * FROM page_content LIMIT 3\G    -- \G formats columns vertically

Map what you find to the new schema. The column names in the old PHP CMS are likely different. Here is the mapping for a typical PHP CMS alongside what the FastAPI schema expects:

Old column (typical PHP CMS)	New column (FastAPI schema)	Notes
subjects.id	subjects.id	Same — used for FK lookups during migration
subjects.name / title	subjects.name	Map directly
subjects.description	subjects.description	Map directly
subjects.sort_order / order_num	subjects.display_order	Rename
subjects.slug (may not exist)	subjects.slug	Generate with slugify(name) if missing
subtopics.subjects_id / subject_id	subtopics.subject_id	Rename if needed
pages.subtopics_id / subtopic_id	pages.subtopic_id	Rename if needed
page_content.html_content / content	page_content.content	Copy as-is — stored HTML
page_content.section_order / sort_order	page_content.display_order	Rename

Update the column names in the migration script below to match what DESCRIBE actually showed you. The script uses comment markers # ADAPT: to flag every place where you need to substitute a real column name from your old database.

3 Accessing the old database — SSH tunnel

The old MySQL server almost certainly does not accept connections from the internet (port 3306 should be firewalled). The safest way to reach it from your laptop is an SSH tunnel: you SSH into the server, and MySQL traffic travels through that encrypted connection. To your local machine it looks like a MySQL server running on localhost:3307.

terminal — open the SSH tunnel (keep this running)

# -L 3307:localhost:3306 means:
#   local port 3307 → tunnel → server's localhost:3306
# -N = don't execute a remote command, just forward ports
# -f = run in background (omit -f if you want it in the foreground)
ssh -L 3307:localhost:3306 your_user@osztromok.com -N -f

# Verify the tunnel is up — you should see MySQL's greeting
mysql -h 127.0.0.1 -P 3307 -u db_user -p old_db_name -e "SHOW TABLES;"

Two terminal windows. Keep the SSH tunnel command running in one terminal (or backgrounded with -f) and run the migration script in another. When you close the tunnel, the migration script will lose its database connection. If you are running the migration directly on the server (e.g. via ssh user@osztromok.com and then running Python there), no tunnel is needed — connect directly to localhost:3306.

.env — add old DB connection (tunnel version) env

# New DB — the FastAPI app's database
DB_HOST=localhost
DB_PORT=3306
DB_USER=new_db_user
DB_PASS=new_db_password
DB_NAME=osztromok_new

# Old DB — accessed via SSH tunnel on port 3307
OLD_DB_HOST=127.0.0.1
OLD_DB_PORT=3307
OLD_DB_USER=old_db_user
OLD_DB_PASS=old_db_password
OLD_DB_NAME=old_db_name

4 scripts/migrate.py — the full migration script

Create a scripts/ folder at the project root. This script is standalone — it imports the FastAPI app's models and database setup, but it does not start the FastAPI server. Run it once from the command line.

The script is idempotent: if you run it twice, it won't create duplicate records. It checks whether a subject (or subtopic, page, etc.) with the same slug already exists before inserting.

scripts/migrate.py python

"""
Content migration script: old PHP CMS → FastAPI schema
═══════════════════════════════════════════════════════
Run once from the project root:
    python -m scripts.migrate

Prerequisites:
  - SSH tunnel open if old DB is remote (see Chapter 9 notes)
  - .env has OLD_DB_* and DB_* variables set
  - New DB tables already created (python -m scripts.create_tables)

The script is idempotent — safe to re-run. It skips records whose
slug already exists in the new database.
"""

import os
import sys
from pathlib import Path

# Ensure project root is on sys.path so we can import app.*
sys.path.insert(0, str(Path(__file__).parent.parent))

import pymysql
from dotenv import load_dotenv
from sqlalchemy import select

load_dotenv()

# ── New DB: import the FastAPI app's database and models ─────────────────
from app.database import Base, engine, SessionLocal, slugify
from app.models   import Subject, Subtopic, Page, PageContent


# ── Old DB: raw PyMySQL connection ───────────────────────────────────────
def old_db_conn():
    return pymysql.connect(
        host    = os.getenv("OLD_DB_HOST", "127.0.0.1"),
        port    = int(os.getenv("OLD_DB_PORT", "3306")),
        user    = os.getenv("OLD_DB_USER"),
        password= os.getenv("OLD_DB_PASS"),
        database= os.getenv("OLD_DB_NAME"),
        charset = "utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,  # rows as dicts
    )


# ── Helpers ──────────────────────────────────────────────────────────────
def get_slug(row: dict, name_col: str = "name") -> str:
    """Use the row's slug column if it exists, otherwise generate one."""
    return row.get("slug") or slugify(row[name_col])


def log(msg: str):
    print(msg, flush=True)


# ═══════════════════════════════════════════════════════════════════════════
# MIGRATE SUBJECTS
# ═══════════════════════════════════════════════════════════════════════════
def migrate_subjects(old, new_db) -> dict[int, int]:
    """Returns a mapping: old subject id → new subject id"""
    log("\n── Subjects ─────────────────────────────")
    id_map = {}

    with old.cursor() as cur:
        # ADAPT: change column names to match your old subjects table
        cur.execute("""
            SELECT
                id,
                name,                           -- ADAPT if different
                description,                    -- ADAPT: may be NULL
                sort_order  AS display_order,   -- ADAPT: rename here
                slug                            -- ADAPT: remove if no slug col
            FROM subjects
            ORDER BY sort_order
        """)
        rows = cur.fetchall()

    for row in rows:
        slug = get_slug(row)

        # Idempotent: skip if already migrated
        existing = new_db.execute(
            select(Subject).where(Subject.slug == slug)
        ).scalar_one_or_none()

        if existing:
            log(f"  SKIP  subject '{slug}' (already exists, id={existing.id})")
            id_map[row["id"]] = existing.id
        else:
            obj = Subject(
                name          = row["name"],
                slug          = slug,
                description   = row.get("description", "") or "",
                display_order = row.get("display_order", 0) or 0,
            )
            new_db.add(obj)
            new_db.flush()  # get the new id without committing yet
            id_map[row["id"]] = obj.id
            log(f"  ADD   subject '{slug}' (old id={row['id']} → new id={obj.id})")

    new_db.commit()
    return id_map


# ═══════════════════════════════════════════════════════════════════════════
# MIGRATE SUBTOPICS
# ═══════════════════════════════════════════════════════════════════════════
def migrate_subtopics(old, new_db, subject_id_map: dict) -> dict[int, int]:
    log("\n── Subtopics ────────────────────────────")
    id_map = {}

    with old.cursor() as cur:
        # ADAPT: change subjects_id to whatever your FK column is named
        cur.execute("""
            SELECT
                id,
                subjects_id AS old_subject_id,  -- ADAPT
                name,
                description,
                sort_order  AS display_order,   -- ADAPT
                slug                            -- ADAPT: remove if absent
            FROM subtopics
            ORDER BY subjects_id, sort_order
        """)
        rows = cur.fetchall()

    for row in rows:
        new_subject_id = subject_id_map.get(row["old_subject_id"])
        if not new_subject_id:
            log(f"  WARN  subtopic id={row['id']}: parent subject not found, skipping")
            continue

        slug = get_slug(row)
        existing = new_db.execute(
            select(Subtopic).where(
                Subtopic.slug       == slug,
                Subtopic.subject_id == new_subject_id,
            )
        ).scalar_one_or_none()

        if existing:
            log(f"  SKIP  subtopic '{slug}'")
            id_map[row["id"]] = existing.id
        else:
            obj = Subtopic(
                subject_id    = new_subject_id,
                name          = row["name"],
                slug          = slug,
                description   = row.get("description", "") or "",
                display_order = row.get("display_order", 0) or 0,
            )
            new_db.add(obj)
            new_db.flush()
            id_map[row["id"]] = obj.id
            log(f"  ADD   subtopic '{slug}'")

    new_db.commit()
    return id_map


# ═══════════════════════════════════════════════════════════════════════════
# MIGRATE PAGES
# ═══════════════════════════════════════════════════════════════════════════
def migrate_pages(old, new_db, subtopic_id_map: dict) -> dict[int, int]:
    log("\n── Pages ────────────────────────────────")
    id_map = {}

    with old.cursor() as cur:
        # ADAPT: change subtopics_id to your FK column name
        cur.execute("""
            SELECT
                id,
                subtopics_id AS old_subtopic_id,  -- ADAPT
                name,
                description,
                sort_order   AS display_order,    -- ADAPT
                slug                              -- ADAPT: remove if absent
            FROM pages
            ORDER BY subtopics_id, sort_order
        """)
        rows = cur.fetchall()

    for row in rows:
        new_subtopic_id = subtopic_id_map.get(row["old_subtopic_id"])
        if not new_subtopic_id:
            log(f"  WARN  page id={row['id']}: parent subtopic not found, skipping")
            continue

        slug = get_slug(row)
        existing = new_db.execute(
            select(Page).where(
                Page.slug        == slug,
                Page.subtopic_id == new_subtopic_id,
            )
        ).scalar_one_or_none()

        if existing:
            log(f"  SKIP  page '{slug}'")
            id_map[row["id"]] = existing.id
        else:
            obj = Page(
                subtopic_id   = new_subtopic_id,
                name          = row["name"],
                slug          = slug,
                description   = row.get("description", "") or "",
                display_order = row.get("display_order", 0) or 0,
            )
            new_db.add(obj)
            new_db.flush()
            id_map[row["id"]] = obj.id
            log(f"  ADD   page '{slug}'")

    new_db.commit()
    return id_map


# ═══════════════════════════════════════════════════════════════════════════
# MIGRATE PAGE CONTENT SECTIONS
# ═══════════════════════════════════════════════════════════════════════════
def migrate_content(old, new_db, page_id_map: dict):
    log("\n── Content sections ─────────────────────")
    added = skipped = 0

    with old.cursor() as cur:
        # ADAPT: change pages_id and html_content to your column names
        cur.execute("""
            SELECT
                id,
                pages_id    AS old_page_id,  -- ADAPT
                html_content AS content,     -- ADAPT (may just be 'content')
                section_order AS display_order -- ADAPT
            FROM page_content
            ORDER BY pages_id, section_order
        """)
        rows = cur.fetchall()

    for row in rows:
        new_page_id = page_id_map.get(row["old_page_id"])
        if not new_page_id:
            skipped += 1
            continue

        # Idempotent check: count existing sections for this page
        # Content sections have no natural unique key, so we skip the whole
        # page's sections if any already exist (prevents duplicates on re-run)
        existing_count = new_db.execute(
            select(PageContent).where(PageContent.page_id == new_page_id)
        ).scalars().all()
        if existing_count:
            skipped += 1
            continue

        obj = PageContent(
            page_id       = new_page_id,
            content       = row["content"] or "",
            display_order = row.get("display_order", 0) or 0,
        )
        new_db.add(obj)
        added += 1

    new_db.commit()
    log(f"  Content: {added} added, {skipped} skipped")


# ═══════════════════════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════════════════════
def main():
    log("=== Content Migration: old CMS → FastAPI schema ===")

    # Ensure new DB tables exist (safe to call even if already created)
    Base.metadata.create_all(bind=engine)

    old    = old_db_conn()
    new_db = SessionLocal()

    try:
        subject_map  = migrate_subjects(old, new_db)
        subtopic_map = migrate_subtopics(old, new_db, subject_map)
        page_map     = migrate_pages(old, new_db, subtopic_map)
        migrate_content(old, new_db, page_map)
        log("\n✓ Migration complete.")
    except Exception as e:
        new_db.rollback()
        log(f"\n✗ Migration FAILED: {e}")
        raise
    finally:
        new_db.close()
        old.close()


if __name__ == "__main__":
    main()

db.flush() vs db.commit(): flush() writes the pending INSERT to the database within the current transaction, which makes the new obj.id available immediately (auto-incremented by the DB engine). commit() finalises the transaction. We flush after each row to get the new ID for building the id_map, then commit after each level (subjects, subtopics, pages) to save progress. If the script crashes mid-run, completed levels are preserved.

5 Running the migration

terminal — full run sequence

# 1. Open SSH tunnel to old DB (keep this terminal open, or use -f)
ssh -L 3307:localhost:3306 your_user@osztromok.com -N

# 2. In a new terminal — activate venv and run the migration
(venv) $ python -m scripts.migrate

# Expected output (abbreviated):
# === Content Migration: old CMS → FastAPI schema ===
# ── Subjects ─────────────────────────────
#   ADD   subject 'japan' (old id=1 → new id=1)
#   ADD   subject 'french' (old id=2 → new id=2)
#   ADD   subject 'hungarian' (old id=3 → new id=3)
# ── Subtopics ────────────────────────────
#   ADD   subtopic 'kanji'
#   ADD   subtopic 'grammar'
# ── Pages ────────────────────────────────
#   ADD   page 'kanji-tiles'
#   ADD   page 'kanji-links'
# ── Content sections ─────────────────────
#   Content: 47 added, 0 skipped
# ✓ Migration complete.

# 3. Re-run to confirm idempotency — all rows should say SKIP, nothing ADD
(venv) $ python -m scripts.migrate

The -m scripts.migrate form (module syntax) rather than python scripts/migrate.py ensures that the project root is correctly on sys.path, which is needed for from app.database import ... to work. Always run from the project root directory (the folder containing app/).

6 scripts/verify_routes.py — hit every URL

After migration, this script crawls every subject → subtopic → page URL in the new database, fires an HTTP request against the locally running FastAPI app, and reports anything that doesn't return HTTP 200.

terminal — install httpx if not already present

(venv) $ pip install httpx

scripts/verify_routes.py python

"""
Route verification script — run after migration.
Hits every subject / subtopic / page URL and reports non-200 responses.

Usage:
    uvicorn app.main:app &          # start the server in the background
    python -m scripts.verify_routes
"""

import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))

import httpx
from dotenv import load_dotenv
from sqlalchemy import select
from app.database import SessionLocal
from app.models   import Subject, Subtopic, Page

load_dotenv()

BASE_URL = "http://127.0.0.1:8000"


def check(client: httpx.Client, url: str, errors: list):
    try:
        r = client.get(url, follow_redirects=True)
        status = r.status_code
        icon   = "✓" if status == 200 else "✗"
        print(f"  {icon} {status}  {url}")
        if status != 200:
            errors.append((status, url))
    except Exception as e:
        print(f"  ✗ ERR  {url}  ({e})")
        errors.append(("ERR", url))


def main():
    db     = SessionLocal()
    errors = []

    with httpx.Client(base_url=BASE_URL, timeout=10) as client:
        print(f"Verifying routes against {BASE_URL}\n")

        subjects = db.execute(select(Subject)).scalars().all()

        for subject in subjects:
            print(f"\n[{subject.slug}]")
            check(client, f"/{subject.slug}", errors)

            subtopics = db.execute(
                select(Subtopic).where(Subtopic.subject_id == subject.id)
            ).scalars().all()

            for subtopic in subtopics:
                check(client, f"/{subject.slug}/{subtopic.slug}", errors)

                pages = db.execute(
                    select(Page).where(Page.subtopic_id == subtopic.id)
                ).scalars().all()

                for page in pages:
                    check(client, f"/{subject.slug}/{subtopic.slug}/{page.slug}", errors)

    db.close()
    print(f"\n{'─'*40}")
    if errors:
        print(f"✗  {len(errors)} problem(s) found:")
        for code, url in errors:
            print(f"   {code}  {url}")
        sys.exit(1)
    else:
        print("✓  All routes returned 200.")

terminal — run verification (server must be running)

# Start FastAPI in one terminal
(venv) $ uvicorn app.main:app --reload

# In a second terminal, run the check
(venv) $ python -m scripts.verify_routes

# Sample output when everything is working:
# Verifying routes against http://127.0.0.1:8000
#
# [japan]
#   ✓ 200  /japan
#   ✓ 200  /japan/kanji
#   ✓ 200  /japan/kanji/kanji-tiles
#   ✓ 200  /japan/kanji/kanji-links
# [french]
#   ✓ 200  /french
#   ✓ 200  /french/grammar
#   ✓ 200  /french/grammar/adjectives
# ────────────────────────────────────────
# ✓  All routes returned 200.

If you see 404s, the most common causes are:
① The slug in the new DB doesn't match what the old site used — check the slugify() output against the old URL
② A subtopic or page was orphaned (its parent was missing from the id_map) — look for WARN lines in the migration output
③ The CMS route has a bug in its slug lookup — test the URL directly in the browser with the echo=True engine to see the SQL query it ran

7 Common edge cases and how to handle them

Duplicate slugs within the same parent

If two pages under the same subtopic have names that produce the same slug (e.g. "Verb Forms" and "Verb-Forms"), the second insert will silently overwrite the first in the id_map and one page will be lost. Detect and fix before migrating:

MySQL — find duplicate name collisions before migrating

SELECT subtopics_id, LOWER(REPLACE(name,' ','-')) AS slug, COUNT(*) AS n
FROM pages
GROUP BY subtopics_id, slug
HAVING n > 1;

Fix duplicates in the old DB (rename one page) before running the migration.

NULL or empty content sections

The migration script already handles this with row["content"] or "". Empty sections are migrated as empty strings and will render as blank blocks. After migration you can clean them up via the admin interface.

HTML character encoding

The old PHP site may have stored content with Windows-1252 encoding or with HTML entities like   and —. These render correctly in browsers — {{ section.content | safe }} passes the HTML through unchanged. You only need to worry about encoding if the MySQL connection returns garbled characters.

Fix: force UTF-8 in the old DB connection python

# Already in old_db_conn() — make sure charset is utf8mb4 not latin1
pymysql.connect(
    ...,
    charset="utf8mb4",
    use_unicode=True,
)

# If the old DB has latin1 columns, force conversion on the cursor:
cur.execute("SET NAMES utf8mb4")

8 Alembic — schema migrations after go-live

Once the site is live with real content, you can't just call Base.metadata.create_all() again to add a new column — it only creates missing tables, it doesn't alter existing ones. Alembic is SQLAlchemy's official tool for versioned schema changes.

terminal — Alembic setup (one-time)

(venv) $ pip install alembic
(venv) $ alembic init migrations        # creates migrations/ folder and alembic.ini

alembic.ini — point at your DB ini

# Replace the default sqlalchemy.url with a reference to your .env
sqlalchemy.url = mysql+pymysql://%(DB_USER)s:%(DB_PASS)s@%(DB_HOST)s/%(DB_NAME)s

migrations/env.py — import your models python

from app.database import Base
from app import models  # noqa — registers all models with Base

# Set target_metadata so Alembic can detect schema changes
target_metadata = Base.metadata

terminal — typical Alembic workflow

# Generate a migration after changing a model (e.g. adding a column)
(venv) $ alembic revision --autogenerate -m "add icon column to subjects"

# Review the generated file in migrations/versions/ — always check it!

# Apply the migration to the database
(venv) $ alembic upgrade head

# Roll back one migration if something went wrong
(venv) $ alembic downgrade -1

# See the current migration state
(venv) $ alembic current

Always review autogenerated migrations. Alembic's --autogenerate compares the ORM models to the live database and generates the SQL diff — but it can miss things (renamed columns, index changes) or generate destructive operations unexpectedly. Read every generated file before running upgrade head. In Spring Boot terms: this is like reviewing a Liquibase or Flyway changeset before applying it.

9 New files this chapter

Project structure — additions

osztromok/
├── .env                     ← + OLD_DB_HOST/PORT/USER/PASS/NAME
├── requirements.txt         ← + httpx, alembic
├── alembic.ini              ← new (created by alembic init)
├── migrations/              ← new (created by alembic init)
│   ├── env.py               ← updated to import Base + models
│   └── versions/            ← migration scripts go here
└── scripts/
    ├── __init__.py          ← new (empty file, makes it a package)
    ├── migrate.py           ← new: old CMS → new schema
    └── verify_routes.py     ← new: hit every URL, report non-200s

✓ Chapter 9 Complete — Milestone reached

SSH tunnel — port-forwarding lets the migration script reach the remote MySQL server as if it were local
scripts/migrate.py — idempotent four-stage migration (subjects → subtopics → pages → content), outputs ADD/SKIP for every record
DictCursor — rows returned as dicts so column names are explicit and readable in the migration code
db.flush() — makes the new auto-increment ID available immediately without committing, used to build the old→new id mapping between stages
ADAPT: comments — every column name that must match the real old schema is flagged
scripts/verify_routes.py — crawls the new DB, builds every valid URL, fires HTTP requests, reports non-200s; exits with code 1 if any failures
Alembic introduced — one-time setup, autogenerate workflow, always-review warning
Site is now content-complete and ready for Chapter 10's deployment to the live server

Quick Reference — Chapter 9

ssh -L 3307:localhost:3306 user@host -N Tunnel remote MySQL to local port 3307

pymysql.connect(cursorclass=DictCursor) Returns rows as dicts — easier column access

db.flush() # get new id before commit Writes INSERT within transaction — id is available immediately

python -m scripts.migrate Run from project root — ensures app.* imports work

row.get("slug") or slugify(row["name"]) Use existing slug or auto-generate one from name

httpx.Client(follow_redirects=True) Follows 302 redirects — kanji routes return 302 → 200

alembic revision --autogenerate -m "desc" Generate migration from model changes — always review output

alembic upgrade head Apply all pending migrations to the database