name	sqlite-to-postgres
description	Migrate a Python app from SQLite to PostgreSQL with zero-downtime production cutover. Covers planning, parallel backend, cutover runbook, cleanup, and the common incidents that bite real migrations. Use when user says "migrate sqlite to postgres", "postgres migration", "db cutover", or hits the sqlite3 concurrency bug (singleton + threadpool race).
allowed-tools	Bash, Read, Write, Edit, Glob, Grep, Agent

SQLite → PostgreSQL Migration Playbook

Battle-tested 4-PR structure. Zero downtime. 7-day rollback window. Works for any Python sync-sqlite3 codebase regardless of framework (FastAPI / Flask / Django / bare asyncio).

Triggers

"migrate sqlite to postgres" / "sqlite → pg"
"db migration planning"
"sqlite concurrency bug" — sqlite3.InterfaceError: bad parameter or other API misuse under concurrent cursor use (singleton Connection shared across threads via asyncio.to_thread or threadpool)
"postgres cutover runbook"
Any question about psycopg3 adapters, JSONB dict issues, or ON CONFLICT DO NOTHING RETURNING race

Core principles

4 independent shippable PRs, never big-bang. Each PR deployable solo. Bake ≥48h between. Skipping bake is user's risk.
Two independent review passes: Backend Architect (operational correctness) + Database Optimizer (schema + concurrency). Then Code Reviewer before every push.
Never delete regression tests during cleanup. Migrate fixture + placeholder, not delete file. Lost business logic coverage = production risk.
Production data lives on the server. Scripts accept --source arg pointing at container path. Never trust local dev DB as migration target.
SQLite rollback window = 7 days. Keep SQLite volume + file until PR-4. Reverse migration script preserves writes made during the window.

Phase 0 — Discovery (before writing any plan)

Launch 3 Explore agents in parallel:

Agent 1: Schema + SQL patterns

Grep across codebase for:

INSERT OR IGNORE, INSERT OR REPLACE, ON CONFLICT (count per file; note dedup vs upsert semantics)
AUTOINCREMENT (all become GENERATED ALWAYS AS IDENTITY)
REAL, TEXT, INTEGER type affinities (SQLite lenient vs PG strict)
PRAGMA, sqlite_master, ROWID
strftime, datetime('now'), json_extract
LIMIT ?,? (SQLite-specific offset syntax)
? placeholders count (→ %s)
0/1 INTEGER boolean columns
JSON TEXT columns (candidates for JSONB)

Output: full type + SQL translation inventory.

Agent 2: Deploy infra + CI

CI config (GitHub Actions, GitLab CI, etc.)
Dockerfile base image + apt deps
docker-compose topology, volumes, networks
Target environment memory/disk budget
Existing secrets management

Output: change list for cutover deploy.

Agent 3: Async + concurrency surface

Every asyncio.to_thread, run_in_executor, threading.Thread that touches DB
Sync DB calls on event loop (route handlers, lifespan hooks)
Explicit transaction boundaries (BEGIN/COMMIT)
Per-module DB conn usage (singleton? pool? per-request?)

Output: concurrency map → determines pool sizing + async driver choice.

Phase 1 — Planning

Write plan file with sections:

Context (why migrating, user quotes if incident-driven)
Non-goals (explicit what this plan does NOT do — no async rewrite, no multi-tenant, no schema split, etc.)
Architecture decisions (driver, types, pool config, test isolation)
PR-1 through PR-4 specs with deploy validation + rollback per PR
Verification section — user-runnable curl/psql commands
Memory follow-ups (what to record post-ship)

Mandatory review chain

Plan agent drafts initial plan
Backend Architect reviews → operational correctness, scope realism, concurrency surface
Database Optimizer reviews → schema pitfalls, type choices, race conditions, index strategy
Apply BLOCKER+HIGH findings from both passes
User approves via ExitPlanMode

Database Optimizer surfaces BLOCKERs that Backend Architect misses. Real findings from actual migration:

NUMERIC + float adapter is cosmetic (defeats precision guarantee; adapter downcasts SUM() result)
ON CONFLICT DO NOTHING RETURNING id has race under concurrent writers (needs explicit transaction + bounded retry)
Pool max_size=5 deadlocks under nested checkouts (needs ≥10 for multi-loop apps)
Partial unique indexes need CHECK (col IS NULL OR col <> '') to close empty-string-vs-NULL trap
idle_in_transaction_session_timeout unset → hung conns hold locks forever

Phase 2 — Architecture decisions (post-review)

Driver: `psycopg[binary]>=3.1,<4` + `psycopg_pool.ConnectionPool`

Rejected:

Driver	Why not
`asyncpg`	Rewrites every call site to `await`. 4× the PR surface.
`psycopg3 async`	Same. Plus threadpool callers still need sync driver.
`psycopg2`	Unmaintained for Python 3.12+ in practice.
`pg8000` (pure Python)	2-3× slower. No upside given libpq is needed for `COPY`.

[binary] extra bundles libpq → no libpq-dev in Dockerfile.

Pool configuration

from psycopg_pool import ConnectionPool
from psycopg.types.json import JsonbBinaryDumper

def _configure_conn(conn):
    """Runs once per connection creation.

    psycopg3 does NOT auto-adapt dict/list → JSONB by default.
    Register at pool configure so every caller gets it for free.
    """
    conn.adapters.register_dumper(dict, JsonbBinaryDumper)
    conn.adapters.register_dumper(list, JsonbBinaryDumper)

_POOL = ConnectionPool(
    conninfo=DATABASE_URL,
    min_size=2,
    max_size=10,              # 5 deadlocks under nested checkouts + multi-loop apps
    open=True,
    timeout=10.0,
    max_lifetime=3600,
    max_idle=600,
    check=ConnectionPool.check_connection,
    configure=_configure_conn,
)

PG server-side safety (docker-compose `command:`)

command:
  - "postgres"
  - "-c" "statement_timeout=10000"                     # kill runaway query at 10s
  - "-c" "idle_in_transaction_session_timeout=60000"   # reclaim hung conns at 60s
  - "-c" "synchronous_commit=on"                       # default; keep for safety
  - "-c" "shared_preload_libraries=pg_stat_statements" # free query observability
  - "-c" "max_connections=20"                          # headroom for pg_dump + psql
  - "-c" "shared_buffers=128MB"                        # tune to host RAM
  - "-c" "effective_cache_size=384MB"
  - "-c" "work_mem=4MB"
  - "-c" "maintenance_work_mem=32MB"
  - "-c" "log_min_duration_statement=500"              # log slow queries

Types: `DOUBLE PRECISION` everywhere, not NUMERIC

NUMERIC(18,6) + float adapter is cosmetic. Matches current SQLite REAL semantics exactly. Zero Python-side coercion churn. If exact-cent accounting needed later, targeted migration on specific columns with Decimal end-to-end.

Booleans: real BOOLEAN for 0/1 columns. Audit == 1 literal comparisons across codebase; rewrite as if col: or col is True.

Timestamps: TIMESTAMPTZ. MANDATORY pre-migration audit — grep every .isoformat() writer column + every reader for dangerous patterns:

SQL: substr(col, 1, 10), strftime('%Y-%m-%d', col), DATE(col), col LIKE '2024-%'
Python: row["col"].startswith(...), row["col"][:10], row["col"].split("T"), datetime.fromisoformat(row["col"])
Pydantic schemas typing timestamps as str

After migration psycopg3 returns datetime objects. All the above break silently. Either migrate callers to handle datetime | str during transition OR serialize in response layer with ConfigDict(json_encoders={datetime: lambda v: v.isoformat()}).

JSON: JSONB. Callers pass dict/list (not str via json.dumps). Readers get dict directly (no json.loads). Register dumper at pool configure (above).

Partial unique indexes: portable to PG 15+. Add CHECK (col IS NULL OR col <> '') constraints on dedup cols to close empty-string-vs-NULL trap.

Test isolation: testcontainers-postgres + TRUNCATE per test

Session-scoped container (5-10s startup). Per-test: TRUNCATE all tables CASCADE.

# tests/fixtures/postgres.py
import pytest
from testcontainers.postgres import PostgresContainer

ALL_TABLES = ["<list every table>"]

@pytest.fixture(scope="session")
def pg_container():
    container = PostgresContainer("postgres:16-alpine")
    container.with_env("POSTGRES_HOST_AUTH_METHOD", "trust")
    with container:
        yield container

@pytest.fixture
def pg_dsn(pg_container, monkeypatch):
    dsn = pg_container.get_connection_url().replace("postgresql+psycopg2://", "postgresql://")
    monkeypatch.setattr(config, "DATABASE_URL", dsn)
    from utils.db_backend._pool import close_pool
    close_pool()
    try:
        yield dsn
    finally:
        close_pool()

@pytest.fixture
def pg_fresh(pg_dsn):
    from utils.db_backend._pool import get_pool
    pool = get_pool()  # triggers schema bootstrap first time
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.execute("TRUNCATE " + ", ".join(ALL_TABLES) + " RESTART IDENTITY CASCADE")
        conn.commit()
    yield

Why not transactional rollback per test: application helpers often open short-lived with pool.connection() as conn: blocks that run in separate transactions — invisible to an outer test-wide transaction. TRUNCATE CASCADE is ~20ms on empty tables; fast enough.

Phase 3 — PR-1: Unblock SQLite concurrency (if applicable)

Skip this PR if migration isn't incident-driven. If sqlite3.InterfaceError is blocking CI/deploy, land this first as the unblock.

Fix: thread-local connections

# utils/database.py

import sqlite3
import threading
from pathlib import Path

_thread_local = threading.local()
_test_override_uri: str | None = None


def _get_db() -> sqlite3.Connection:
    conn = getattr(_thread_local, "conn", None)
    if conn is not None:
        return conn
    if _test_override_uri is not None:
        conn = sqlite3.connect(_test_override_uri, uri=True, check_same_thread=False)
    else:
        path = Path(DB_PATH)
        path.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(str(path), check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA foreign_keys=ON")
    conn.execute("PRAGMA busy_timeout=5000")  # CRITICAL for multi-writer
    _create_tables(conn)
    _thread_local.conn = conn
    return conn


def _reset_for_tests(uri: str | None = None) -> None:
    global _test_override_uri
    existing = getattr(_thread_local, "conn", None)
    if existing is not None:
        try: existing.close()
        except sqlite3.Error: pass
        del _thread_local.conn
    _test_override_uri = uri

Test fixture: tempfile DB (not shared-cache)

@pytest.fixture
def fresh_db(tmp_path):
    from utils import database as db_mod
    db_path = tmp_path / "test.db"
    db_mod._reset_for_tests(str(db_path))
    try:
        db_mod._get_db()  # warm conn + create tables
        yield
    finally:
        db_mod._reset_for_tests(None)

Do NOT use file::memory:?cache=shared&uri=true — SQLite shared-cache falls back to non-WAL journal; deadlocks under multi-writer load even with busy_timeout.

Regression tests

Distinct threads get distinct sqlite3.Connection objects
Concurrent access: 4 threads hammering same compute path → zero InterfaceError

Phase 4 — PR-2: Postgres backend parallel

Feature-flagged via DATABASE_URL:

Empty or missing → SQLite backend (current prod behavior)
postgresql://... or postgres://... → PG backend

Production stays SQLite until PR-3. All tests run on both backends.

Structure

<app>/utils/db_backend/
├── __init__.py          # Dispatcher (PEP 562 __getattr__)
├── sqlite_backend.py    # Existing SQLite impl moved here
├── pg_backend.py        # New psycopg3 impl (identical public API)
├── pg_schema.sql        # PG DDL
├── pg_types.py          # Type adapters + coercion helpers
└── _pool.py             # Pool lifecycle + advisory-lock schema bootstrap

Dispatcher (PEP 562 getattr + re-entry guard)

# db_backend/__init__.py
from contextlib import contextmanager
import config

_backend = None
_loading = False  # re-entry guard — submodule imports trigger __getattr__

def _get_backend():
    global _backend
    if _backend is not None:
        return _backend
    url = (getattr(config, "DATABASE_URL", None) or "").strip()
    if url.startswith(("postgresql://", "postgres://")):
        from . import pg_backend
        _backend = pg_backend
    else:
        from . import sqlite_backend
        _backend = sqlite_backend
    return _backend

def __getattr__(name):
    global _loading
    if _loading or name in ("sqlite_backend", "pg_backend"):
        raise AttributeError(name)
    _loading = True
    try:
        backend = _get_backend()
        if hasattr(backend, name):
            attr = getattr(backend, name)
            globals()[name] = attr
            return attr
        raise AttributeError(name)
    finally:
        _loading = False

@contextmanager
def raw_conn():
    """Context manager yielding (conn, cur) — works on both backends."""
    backend = _get_backend()
    with backend.raw_conn() as (conn, cur):
        yield conn, cur

Schema translation cheatsheet

SQLite	PG	Notes
`INTEGER PRIMARY KEY AUTOINCREMENT`	`BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY`
`TEXT` (iso timestamp writer)	`TIMESTAMPTZ`	Audit readers
`TEXT` (date-only `YYYY-MM-DD`)	`TEXT` (keep)
`INTEGER` (unix ms)	`BIGINT` (keep)
`REAL`	`DOUBLE PRECISION`
`INTEGER` 0/1 bool	`BOOLEAN`
`TEXT` JSON	`JSONB`	Caller contract: dict/list
`INSERT OR IGNORE`	`INSERT ... ON CONFLICT DO NOTHING`
`INSERT OR REPLACE`	`INSERT ... ON CONFLICT(pk) DO UPDATE SET ...`
`PRAGMA *`	Drop
`datetime('now')` DEFAULT	`now()`
`?` placeholder	`%s`	Shim for gradual migration (below)
`cursor.lastrowid`	`INSERT ... RETURNING id`
`ALTER TABLE ADD COLUMN` (try/except OperationalError)	`ADD COLUMN IF NOT EXISTS` (9.6+)
— (new)	`CHECK (col IS NULL OR col <> '')` on dedup cols	Closes NULL/empty trap
— (new)	`CREATE EXTENSION IF NOT EXISTS pg_stat_statements`	Free observability

Schema bootstrap with advisory lock

Cold-start concurrent pool checkouts can race on CREATE TABLE IF NOT EXISTS → deadlock on pg_class ACCESS EXCLUSIVE. Guard with session-wide advisory lock:

_SCHEMA_LOCK_KEY = 0xBAD1_BEEF

def _ensure_schema():
    schema_sql = (Path(__file__).parent / "pg_schema.sql").read_text()
    with _POOL.connection() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT pg_advisory_xact_lock(%s)", (_SCHEMA_LOCK_KEY,))
            cur.execute(schema_sql)
        conn.commit()  # releases xact lock

Dedup-INSERT race fix (ON CONFLICT DO NOTHING + RETURNING)

Problem: ON CONFLICT DO NOTHING RETURNING id returns 0 rows on conflict.
Two concurrent writers both see empty RETURNING before either commits →
both fall through to fallback SELECT → both see nothing → both return
(None, False). Caller treats as failure.

Fix: explicit transaction + bounded retry:

def save_row_with_dedup(data):
    pool = get_pool()
    with pool.connection() as conn:
        with conn.transaction():  # explicit txn covers INSERT + fallback
            with conn.cursor(row_factory=dict_row) as cur:
                cur.execute(INSERT_SQL, params)
                row = cur.fetchone()
                if row:
                    return int(row["id"]), True
                # Bounded retry for the race window
                for attempt in range(3):
                    existing_id = _find_existing_id(cur, data)
                    if existing_id is not None:
                        return existing_id, False
                    time.sleep(0.01 * (attempt + 1))  # 10ms, 20ms, 30ms
                raise RuntimeError("fallback lookup failed after retries")

Paramstyle shim for gradual caller migration

Callers typed ? for sqlite3. PG wants %s. Wrap sqlite cursor to accept both:

import re

class _ParamstyleCursor:
    """Wraps sqlite3.Cursor to accept %s placeholders.

    Rewrites %s → ? before execute. Named placeholders NOT supported —
    positional only (matches psycopg3 positional pyformat).
    """
    _PLACEHOLDER_RE = re.compile(r"%s")

    def __init__(self, inner):
        self._inner = inner

    def execute(self, sql, params=()):
        return self._inner.execute(self._PLACEHOLDER_RE.sub("?", sql), params)

    def executemany(self, sql, seq_params):
        return self._inner.executemany(self._PLACEHOLDER_RE.sub("?", sql), seq_params)

    def fetchone(self): return self._inner.fetchone()
    def fetchall(self): return self._inner.fetchall()
    def fetchmany(self, size=None): return self._inner.fetchmany(size) if size is not None else self._inner.fetchmany()
    def close(self): self._inner.close()
    def __iter__(self): return iter(self._inner)
    @property
    def description(self): return self._inner.description
    @property
    def rowcount(self): return self._inner.rowcount
    @property
    def lastrowid(self): return self._inner.lastrowid

Callers use uniform %s, works on both backends.

Parametrized tests

@pytest.fixture(params=["sqlite", "postgres"], ids=["sqlite", "postgres"])
def fresh_db(request, tmp_path, monkeypatch):
    if request.param == "sqlite":
        # tempfile path from PR-1
        ...
    else:
        request.getfixturevalue("pg_fresh")
    yield request.param

Markers @pytest.mark.sqlite / @pytest.mark.postgres skip off-target params.

Phase 5 — PR-3: Cutover

Order matters

PR-3 Phase A — deploy PG services (DATABASE_URL still empty). Bot stays SQLite.
Do NOT add depends_on: postgres: service_healthy yet — couples site availability to PG during dormant phase.

Snapshot live SQLite on server:

docker compose exec <app> python3 -c "import sqlite3; \
    sqlite3.connect('/app/data/app.db').backup(sqlite3.connect('/app/data/app-cutover.db'))"
docker cp <container>:/app/data/app-cutover.db ./offline-keepsake.db

Run migration via docker exec (NOT locally — prod data lives on server):

docker compose exec -e DATABASE_URL="postgresql://user:pass@postgres:5432/db" \
    <app> python3 -m scripts.migrate_sqlite_to_pg --source /app/data/app.db --commit

Run verify — must exit 0:

docker compose exec -e DATABASE_URL="..." \
    <app> python3 -m scripts.verify_migration --source /app/data/app.db

PR-3 Phase B — flip DATABASE_URL in deploy config, restore depends_on. Push. Bot restarts on PG.
Validate /health shows pool open. 30-min bake.

Migration script — hybrid INSERT + COPY

# scripts/migrate_sqlite_to_pg.py

# Dedup-aware tables — go through public save_* helpers that use
# ON CONFLICT DO NOTHING. Safe on re-run for these (unique constraints
# dedup). NO explicit dedup → duplicates on re-run; TRUNCATE first.
INSERT_TABLES = ["users", "trades", ...]

# Bulk tables — COPY for speed. TRUNCATE + COPY in same txn for atomic
# per-table replacement. Drop IDENTITY id column (PG re-sequences).
COPY_TABLES = ["historical_events", "audit_log", ...]

TIMESTAMPTZ_COLUMNS = {"created_at", "updated_at", "ts", ...}
BOOLEAN_COLUMNS = {"active", "verified", ...}
JSONB_COLUMNS = {"metadata", "payload", ...}


def _coerce_for_pg(col_name, value):
    """Type coercion for rough edges between backends."""
    if col_name in TIMESTAMPTZ_COLUMNS and value is not None:
        if isinstance(value, datetime):
            return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
        # CRITICAL: handle legacy unix timestamp storage
        if isinstance(value, (int, float)) and not isinstance(value, bool):
            return datetime.fromtimestamp(float(value), tz=timezone.utc)
        if isinstance(value, str) and value.strip():
            s = value.strip()
            # Try unix ts string first (legacy writer)
            if s.lstrip("-").replace(".", "", 1).isdigit():
                try: return datetime.fromtimestamp(float(s), tz=timezone.utc)
                except (ValueError, OSError, OverflowError): pass
            # Fall back to ISO
            try: dt = datetime.fromisoformat(s)
            except ValueError: dt = datetime.fromisoformat(s.rstrip("Z"))
            return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
        return value
    if col_name in BOOLEAN_COLUMNS and isinstance(value, int):
        return bool(value)
    if col_name in JSONB_COLUMNS and isinstance(value, str) and value:
        return json.loads(value)  # psycopg adapter re-serializes to JSONB
    return value


def copy_table(src_conn, dst_conn, table):
    """Bulk COPY via psycopg3 binary. TRUNCATE before populate."""
    with dst_conn.cursor() as cur:
        cur.execute(f"TRUNCATE {table} RESTART IDENTITY CASCADE")
    cur_src = src_conn.execute(f"SELECT * FROM {table}")
    col_names = [d[0] for d in cur_src.description if d[0] != "id"]  # drop IDENTITY
    with dst_conn.cursor() as cur_dst:
        with cur_dst.copy(
            sql.SQL("COPY {} ({}) FROM STDIN").format(
                sql.Identifier(table),
                sql.SQL(", ").join(sql.Identifier(c) for c in col_names),
            )
        ) as copy:
            for row in cur_src:
                tuple_row = tuple(
                    _coerce_for_pg(col_names[i], row[i + 1])  # +1 to skip id
                    for i in range(len(col_names))
                )
                copy.write_row(tuple_row)
    dst_conn.commit()

Verify script

# scripts/verify_migration.py

# Per table:
# - Row count match
# - MAX(id) sanity (contiguous if no deletes)
# - Aggregate spot-checks (SUM, MAX(timestamp))
# - Last 10 rows field-by-field equality on key columns

# Exit 0 only if all checks pass. Any FAIL → abort cutover.

Reverse migration (rollback)

Mirror migrate_sqlite_to_pg.py — reads PG via public helpers, writes SQLite via direct sqlite3. Same type coercion in reverse (datetime → ISO str, bool → 0/1, dict → json str). Preserves writes made during PG window if rollback needed.

Known incidents from real migrations

Incident 1: Unix timestamp string in TIMESTAMPTZ columns. Legacy writer path stored created_at='1776374631' (unix seconds as string). datetime.fromisoformat("1776374631") crashes. Coercion must try int parse before ISO (see _coerce_for_pg above).

Incident 2: Mid-migration crash leaves partial state. If INSERT_TABLES crashes mid-way, re-run duplicates rows (no natural key dedup). Fix: TRUNCATE INSERT_TABLES before every re-run. Add --truncate-first flag.

Incident 3: Dedup helpers don't dedup DRY_RUN-style rows. Production rows have tx_hash UNIQUE; test rows lack it. Partial index predicate WHERE tx_hash IS NOT NULL skips dedup for NULL rows → duplicates on re-run. Fix: secondary dedup predicate on (asset_id, taker_order_id, match_time) triplet with WHERE tx_hash IS NULL AND all three present.

Incident 4: psycopg3 dict → JSONB requires explicit dumper. Default raises ProgrammingError: cannot adapt type 'dict' using placeholder '%s'. Register JsonbBinaryDumper at pool configure (see above).

Incident 5: bot.depends_on: postgres: service_healthy in Phase A. If PG fails to start (bad password, OOM, cold-boot race), bot won't start → site down while bot isn't even using PG yet. Add depends_on ONLY in Phase B cutover.

Incident 6: Health endpoint crashes post-cleanup. config.DB_PATH removed in PR-4 but still referenced by /health → AttributeError every call. Grep aggressively before merging PR-4.

Phase 6 — PR-4: Cleanup

Preserve regression coverage — DO NOT DELETE TESTS WHOLESALE

Files tagged @pytest.mark.sqlite from PR-2 often contain valuable business-logic + security tests that happen to use _get_db().execute() for fixture setup. Do NOT delete. Migrate:

git show <PR-3-cutover>^:<test-file> > <test-file>
# Then:
# 1. Remove @pytest.mark.sqlite markers
# 2. Swap fixture: _fresh_db tempfile → pg_fresh
# 3. Replace "... ?" SQL with "... %s"
# 4. Drop only genuinely sqlite-internal assertions
#    (sqlite3.Cursor identity, PRAGMA, thread_local internals)

Loss threshold: ~<5% of tests should be genuinely sqlite-specific. Anything more → review triage.

Delete safely

sqlite_backend.py
Migration scripts (migrate_sqlite_to_pg.py, reverse, verify) — keep in git history only
aiosqlite from requirements
DB_PATH env var everywhere
sqlite-data docker volume mount + declaration

Collapse dispatcher

# db_backend/__init__.py becomes:
"""Backend facade — always PostgreSQL post-PR-4."""
from .pg_backend import *  # noqa: F401,F403

Fail-closed guard

# _pool.py
def get_pool() -> ConnectionPool:
    global _POOL
    if _POOL is not None:
        return _POOL
    if not config.DATABASE_URL:
        raise RuntimeError(
            "DATABASE_URL required — SQLite backend removed in PR-4. "
            "Set DATABASE_URL=postgresql://user:pass@host:5432/db."
        )
    _POOL = ConnectionPool(...)
    _ensure_schema()
    return _POOL

Audit before merge

# Zero hits expected:
grep -rn "import sqlite3" <app>/ | grep -v tests/fixtures
grep -rn "config.DB_PATH" <app>/
grep -rn "_ParamstyleCursor" <app>/
grep -rn "sqlite_backend" <app>/
grep -rn "aiosqlite" requirements.txt

Also update:

Documentation (runbook, architecture, env examples)
HealthOut schema if it referenced SQLite file size
.env.example — DB_PATH → DATABASE_URL

Code Reviewer checkpoints (mandatory per PR)

Before each push, verify:

Grep rules above return zero hits (where applicable to the PR)
Both backends export identical public function surface (pin via test)
psycopg[binary] wheel resolves on all target platforms (darwin/arm64, linux/amd64)
testcontainers fixture skips gracefully if Docker unavailable
No secrets in .env.example, CI config, or commit diff
Deploy config $$ escaping validated with docker-compose config
Co-Authored-By trailer on commits
Tests: no business-logic regression count; restored tests use pg_fresh
Rollback runbook accurately reflects current state

Rollback paths per phase

Phase	Rollback mechanism	Data loss window
PR-1	`git revert`	None (no schema change)
PR-2	`git revert`	None (prod still SQLite)
PR-3 first hour	Flip `DATABASE_URL` empty → redeploy	None (SQLite volume intact)
PR-3 hours 1-24	Run reverse migration → flip → redeploy	Preserved
PR-3 days 1-7	Restore `pg-backups/*.sql.gz` → reverse migration → flip	Up to last backup (cadence-dependent)
PR-4	`git revert` restores backend code but prod is on PG	Full reverse migration + replay

Keep SQLite file + volume through PR-4 (7-day window). Delete only after rollback window closes.

Effort estimates

For a ~1500-2000 LOC sqlite3 layer with 15-25 tables:

PR	Net LOC	Effort	Risk
PR-1	~+100	4-6h	Low (no schema change)
PR-2	~+2500	5-7 days	Zero prod behavior change
PR-3	~+1500	2-4 days	Medium (data migration + cutover)
PR-4	~−1500 net	1-2 days	Medium (hard cutover — rollback expensive)

Total: 2-3.5 weeks in 4 shippable slices with ≥48h bake between.

Quick-start checklist

Discovery (3 parallel Explore agents)
Plan agent → Backend Architect review → Database Optimizer review → user ExitPlanMode
PR-1: SQLite thread-local fix (skip if not incident-driven)
PR-2 Phase A: Reader datetime tolerance (prep for TIMESTAMPTZ)
PR-2 Phase B: Extract sqlite backend + dispatcher
PR-2 Phase C: pg_backend + pg_schema + pool
PR-2 Phase D: raw_conn shim + migrate raw callers + fold satellite tables
PR-2 Phase E: testcontainers + parametrized tests
PR-2 Phase F: paramstyle shim + markers on low-level test files
PR-2 Phase G: JSONB dumper registration (catches a very specific psycopg3 trap)
PR-3 Phase A: PG services (DATABASE_URL empty)
PR-3 Phase B: snapshot → migrate → verify → flip → pool telemetry
PR-4: remove sqlite backend + migrate tests + scrub refs

Every phase: Code Reviewer before push. No exceptions.

Related references

psycopg3 documentation: https://www.psycopg.org/psycopg3/docs/
psycopg_pool: https://www.psycopg.org/psycopg3/docs/api/pool.html
testcontainers-python: https://testcontainers-python.readthedocs.io/
PostgreSQL documentation: https://www.postgresql.org/docs/16/

License

Share freely. No warranty. Battle-tested once in production, but every migration surprises you differently.

reberthkss/SKILL.md