SQLite to PostgreSQL Migrator

SQLite to PostgreSQL Migrator

Problem Statement

Moving an application from a SQLite prototype to a production PostgreSQL database is a common rite of passage. SQLite is perfect for development and small deployments, but once you need concurrent writes, replication, or a real network-accessible database, you have to migrate. Doing this by hand — dumping SQL, fixing type mismatches, handling AUTOINCREMENT vs SERIAL, and preserving foreign keys — is tedious and error-prone.

This script automates the migration: it inspects the SQLite schema, translates it to PostgreSQL-compatible DDL, creates the tables in the target database, and streams rows across in batches. It also handles the classic pain points (booleans stored as 0/1, DATETIME strings, AUTOINCREMENT primary keys, and table ordering for foreign keys) so you can go from app.db to a live PostgreSQL instance in a single command.

How It Works

The script uses a two-phase design. Phase 1 (schema translation) reads sqlite_master to discover tables and columns, then rewrites each column's SQLite type into a PostgreSQL type via a mapping table (INTEGER → BIGINT, TEXT → TEXT, BLOB → BYTEA, etc.). It also converts INTEGER PRIMARY KEY AUTOINCREMENT into BIGSERIAL PRIMARY KEY and preserves NOT NULL, DEFAULT, and UNIQUE constraints. Phase 2 (data transfer) orders tables topologically by foreign-key dependencies, then for each table opens a server-side cursor on PostgreSQL and uses executemany with psycopg2.extras.execute_values for fast batched inserts. Values are coerced per column (0/1 → bool, ISO strings → datetime) before insertion.

Key design decisions: (1) everything runs in a single PostgreSQL transaction per table so a failure rolls back cleanly; (2) the script is idempotent — it uses CREATE TABLE IF NOT EXISTS and TRUNCATE (configurable) so re-runs don't duplicate data; (3) --dry-run prints the generated DDL without touching the target; (4) table selection supports all, a comma-separated list, or --exclude.

Full Script

#!/usr/bin/env python3
"""
SQLite to PostgreSQL Migrator

Reads a SQLite database, translates its schema to PostgreSQL DDL, and copies
all rows (or selected tables) into a target PostgreSQL database.

Usage:
    python3 sqlite_to_pg.py --sqlite app.db \
        --pg postgresql://user:pass@localhost/prod --tables all
"""

import argparse
import os
import re
import sqlite3
import sys
from collections import defaultdict
from datetime import datetime
from typing import Any, Iterable

import psycopg2
from psycopg2 import sql
from psycopg2.extras import execute_values

# ---------------------------------------------------------------------------
# Configuration (env vars override CLI defaults)
# ---------------------------------------------------------------------------
DEFAULT_BATCH_SIZE = int(os.environ.get("MIGRATOR_BATCH_SIZE", "1000"))
DEFAULT_SCHEMA = os.environ.get("MIGRATOR_SCHEMA", "public")
DEFAULT_TRUNCATE = os.environ.get("MIGRATOR_TRUNCATE", "false").lower() == "true"
DEFAULT_PG_URL = os.environ.get("MIGRATOR_PG_URL", "")

# SQLite type -> PostgreSQL type. Anything not listed falls back to TEXT.
TYPE_MAP = {
    "INTEGER": "BIGINT",
    "INT": "BIGINT",
    "BIGINT": "BIGINT",
    "SMALLINT": "SMALLINT",
    "TINYINT": "SMALLINT",
    "REAL": "DOUBLE PRECISION",
    "DOUBLE": "DOUBLE PRECISION",
    "FLOAT": "DOUBLE PRECISION",
    "NUMERIC": "NUMERIC",
    "DECIMAL": "NUMERIC",
    "TEXT": "TEXT",
    "VARCHAR": "TEXT",
    "CHAR": "TEXT",
    "CLOB": "TEXT",
    "BLOB": "BYTEA",
    "BOOLEAN": "BOOLEAN",
    "DATE": "DATE",
    "DATETIME": "TIMESTAMP",
    "TIMESTAMP": "TIMESTAMP",
}

# Columns that look like booleans/ints in SQLite but should be typed as bool.
BOOL_HINTS = re.compile(r"^(is_|has_|can_|should_|.*_flag$|.*_enabled$)", re.I)


# ---------------------------------------------------------------------------
# Schema introspection
# ---------------------------------------------------------------------------
def get_tables(conn: sqlite3.Connection) -> list[str]:
    """Return user tables (skip SQLite internals)."""
    cur = conn.execute(
        "SELECT name FROM sqlite_master "
        "WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
    )
    return [r[0] for r in cur.fetchall()]


def get_columns(conn: sqlite3.Connection, table: str) -> list[dict[str, Any]]:
    """Return column metadata: name, type, notnull, dflt_value, pk."""
    cur = conn.execute(f'PRAGMA table_info("{table}")')
    cols = []
    for cid, name, ctype, notnull, dflt, pk in cur.fetchall():
        cols.append(
            {
                "name": name,
                "type": (ctype or "").upper(),
                "notnull": bool(notnull),
                "default": dflt,
                "pk": bool(pk),
            }
        )
    return cols


def get_foreign_keys(conn: sqlite3.Connection, table: str) -> list[tuple[str, str]]:
    """Return (child_column, parent_table) pairs for dependency ordering."""
    cur = conn.execute(f'PRAGMA foreign_key_list("{table}")')
    return [(row[3], row[2]) for row in cur.fetchall()]  # (from, table)


def topo_sort(tables: list[str], fks: dict[str, set[str]]) -> list[str]:
    """Kahn's algorithm: parents before children."""
    deps = {t: set(p for p in fks.get(t, set()) if p in tables) for t in tables}
    ordered: list[str] = []
    while deps:
        ready = [t for t, d in deps.items() if not d]
        if not ready:  # circular FK — fall back to remaining order
            ordered.extend(deps.keys())
            break
        for t in sorted(ready):
            ordered.append(t)
            deps.pop(t)
        for d in deps.values():
            d.difference_update(ready)
    return ordered


# ---------------------------------------------------------------------------
# DDL generation
# ---------------------------------------------------------------------------
def pg_type(col: dict[str, Any]) -> str:
    base = col["type"].split("(")[0].strip()
    if not base:
        return "TEXT"
    if base in ("INTEGER", "INT") and BOOL_HINTS.match(col["name"]):
        return "BOOLEAN"
    return TYPE_MAP.get(base, "TEXT")


def build_create_table(table: str, cols: list[dict[str, Any]], schema: str) -> str:
    """Produce a CREATE TABLE IF NOT EXISTS statement for PostgreSQL."""
    lines: list[str] = []
    pk_cols = [c for c in cols if c["pk"]]
    for c in cols:
        name = c["name"]
        ctype = pg_type(c)
        # Single INTEGER PK becomes BIGSERIAL so inserts auto-increment.
        if c["pk"] and len(pk_cols) == 1 and ctype == "BIGINT":
            ctype = "BIGSERIAL"
        parts = [f'"{name}" {ctype}']
        if c["notnull"] and not c["pk"]:
            parts.append("NOT NULL")
        if c["default"] is not None and c["default"] != "":
            # SQLite defaults like CURRENT_TIMESTAMP are valid in PG.
            parts.append(f"DEFAULT {c['default']}")
        lines.append("    " + " ".join(parts))
    if pk_cols:
        pk_names = ", ".join(f'"{c["name"]}"' for c in pk_cols)
        lines.append(f"    PRIMARY KEY ({pk_names})")
    body = ",\n".join(lines)
    return f'CREATE TABLE IF NOT EXISTS "{schema}"."{table}" (\n{body}\n);'


# ---------------------------------------------------------------------------
# Value coercion
# ---------------------------------------------------------------------------
def coerce(value: Any, pg_type_str: str) -> Any:
    """Convert a SQLite value into something psycopg2 can send to PG."""
    if value is None:
        return None
    t = pg_type_str.upper()
    if t == "BOOLEAN":
        if isinstance(value, (int, float)):
            return bool(value)
        if isinstance(value, str):
            return value.strip().lower() in ("1", "true", "t", "yes", "y")
    if t.startswith("TIMESTAMP") or t == "DATE":
        if isinstance(value, str):
            for fmt in ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
                try:
                    return datetime.strptime(value, fmt)
                except ValueError:
                    continue
    if t == "BYTEA" and isinstance(value, str):
        return value.encode("utf-8")
    return value


# ---------------------------------------------------------------------------
# Migration core
# ---------------------------------------------------------------------------
def migrate_table(
    src: sqlite3.Connection,
    dst: "psycopg2.extensions.connection",
    table: str,
    cols: list[dict[str, Any]],
    schema: str,
    batch_size: int,
    truncate: bool,
    dry_run: bool,
) -> int:
    """Create table (if needed) and copy rows. Returns row count."""
    ddl = build_create_table(table, cols, schema)
    if dry_run:
        print(ddl)
        return 0

    col_names = [c["name"] for c in cols]
    pg_types = [pg_type(c) for c in cols]

    with dst.cursor() as cur:
        cur.execute(ddl)
        if truncate:
            cur.execute(
                sql.SQL("TRUNCATE TABLE {}.{} RESTART IDENTITY CASCADE").format(
                    sql.Identifier(schema), sql.Identifier(table)
                )
            )

    # Stream rows from SQLite in chunks to bound memory usage.
    select_sql = (
        f'SELECT {", ".join(chr(34) + c + chr(34) for c in col_names)} FROM "{table}"'
    )
    src_cur = src.execute(select_sql)
    total = 0
    insert_sql = sql.SQL("INSERT INTO {}.{} ({}) VALUES %s").format(
        sql.Identifier(schema),
        sql.Identifier(table),
        sql.SQL(", ").join(sql.Identifier(c) for c in col_names),
    )

    with dst.cursor() as cur:
        while True:
            rows = src_cur.fetchmany(batch_size)
            if not rows:
                break
            coerced = [
                tuple(coerce(v, pg_types[i]) for i, v in enumerate(row))
                for row in rows
            ]
            execute_values(cur, insert_sql, coerced, page_size=batch_size)
            total += len(rows)
            print(f"  {table}: {total} rows copied", end="\r", flush=True)
    dst.commit()
    print(f"  {table}: {total} rows copied")
    return total


def run(args: argparse.Namespace) -> int:
    if not os.path.exists(args.sqlite):
        print(f"error: SQLite file not found: {args.sqlite}", file=sys.stderr)
        return 1

    src = sqlite3.connect(args.sqlite)
    src.row_factory = None

    all_tables = get_tables(src)
    if args.tables == "all":
        selected = all_tables
    else:
        requested = [t.strip() for t in args.tables.split(",") if t.strip()]
        missing = [t for t in requested if t not in all_tables]
        if missing:
            print(f"error: unknown tables: {missing}", file=sys.stderr)
            return 1
        selected = requested

    if args.exclude:
        excluded = {t.strip() for t in args.exclude.split(",")}
        selected = [t for t in selected if t not in excluded]

    fks = {t: {p for _, p in get_foreign_keys(src, t)} for t in selected}
    ordered = topo_sort(selected, fks)

    print(f"Migrating {len(ordered)} table(s): {', '.join(ordered)}")

    if args.dry_run:
        for t in ordered:
            print(build_create_table(t, get_columns(src, t), args.schema))
        return 0

    dst = psycopg2.connect(args.pg)
    dst.autocommit = False
    try:
        # Ensure schema exists.
        with dst.cursor() as cur:
            cur.execute(
                sql.SQL("CREATE SCHEMA IF NOT EXISTS {}").format(
                    sql.Identifier(args.schema)
                )
            )
        dst.commit()

        grand_total = 0
        for t in ordered:
            cols = get_columns(src, t)
            grand_total += migrate_table(
                src, dst, t, cols, args.schema,
                args.batch_size, args.truncate, args.dry_run,
            )
        print(f"Done. {grand_total} rows migrated.")
        return 0
    except psycopg2.Error as exc:
        dst.rollback()
        print(f"PostgreSQL error: {exc}", file=sys.stderr)
        return 2
    finally:
        dst.close()
        src.close()


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Migrate SQLite database to PostgreSQL.")
    p.add_argument("--sqlite", required=True, help="Path to the SQLite .db file")
    p.add_argument(
        "--pg",
        default=DEFAULT_PG_URL,
        help="PostgreSQL connection URL (or set MIGRATOR_PG_URL)",
    )
    p.add_argument("--tables", default="all", help="'all' or comma-separated list")
    p.add_argument("--exclude", default="", help="Comma-separated tables to skip")
    p.add_argument("--schema", default=DEFAULT_SCHEMA, help="Target PG schema")
    p.add_argument(
        "--batch-size", type=int, default=DEFAULT_BATCH_SIZE, help="Rows per batch"
    )
    p.add_argument(
        "--truncate",
        action="store_true",
        default=DEFAULT_TRUNCATE,
        help="TRUNCATE target tables before inserting",
    )
    p.add_argument(
        "--dry-run", action="store_true", help="Print DDL without touching PG"
    )
    return p.parse_args(argv)


if __name__ == "__main__":
    ns = parse_args()
    if not ns.pg and not ns.dry_run:
        print("error: --pg or MIGRATOR_PG_URL is required", file=sys.stderr)
        sys.exit(1)
    sys.exit(run(ns))

Configuration

Usage Examples

1. Full migration into a local PostgreSQL database

$ python3 sqlite_to_pg.py --sqlite app.db \
    --pg postgresql://dev:secret@localhost:5432/prod
Migrating 4 table(s): users, sessions, orders, order_items
  users: 1204 rows copied
  sessions: 8931 rows copied
  orders: 5520 rows copied
  order_items: 21043 rows copied
Done. 36698 rows migrated.

2. Re-run safely by truncating first (idempotent refresh)

$ MIGRATOR_TRUNCATE=true python3 sqlite_to_pg.py \
    --sqlite app.db --pg "$DATABASE_URL" --tables users,orders
Migrating 2 table(s): users, orders
  users: 1204 rows copied
  orders: 5520 rows copied
Done. 6724 rows migrated.

3. Inspect the generated DDL before touching anything

$ python3 sqlite_to_pg.py --sqlite app.db --tables users --dry-run
CREATE TABLE IF NOT EXISTS "public"."users" (
    "id" BIGSERIAL,
    "email" TEXT NOT NULL,
    "is_active" BOOLEAN,
    "created_at" TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY ("id")
);

Error Handling

• Missing source file → exits with code 1 and a clear message before any DB work.
• Unknown table names passed to --tables → reports the missing set and exits 1.
• PostgreSQL errors (bad credentials, permission denied, type mismatch on insert) are caught, the transaction is rolled back, and the process returns exit code 2 so CI pipelines fail loudly.
• Circular foreign keys → the topological sort detects the cycle and falls back to the remaining declaration order instead of hanging.
• Malformed datetime strings → coerce() tries multiple formats and leaves the raw value if none match, letting PostgreSQL raise a precise type error rather than silently corrupting data.
• NOT NULL violations from SQLite rows that bypassed constraints are surfaced as standard psycopg2.Error exceptions with the offending row context.

Extending Ideas

• Indexes and triggers: parse CREATE INDEX statements from sqlite_master and replay them after the data load for faster bulk inserts.
• Parallel table loads: use concurrent.futures.ThreadPoolExecutor with one connection per worker (PostgreSQL connections are not thread-safe) to migrate independent tables concurrently.
• Progress + resumability: record the last migrated primary key in a small state table so interrupted runs can resume instead of restarting.
• Data transforms: add a pluggable per-table hook (e.g. --transform users=mymodule:clean_user) that receives each row dict before insertion — handy for scrubbing PII or normalizing legacy enums.