Migrate existing homebox sqlite DB to postgres DB

[dhop90]

Steps to migrate homebox from sqlite 3.x database to postgresql

1. Force Drop database homebox in postgres, if DB already exists

2. Create database in postgres - used pgadmin4 create DB

3. Start app, let it initialize db - console message
   2025/11/26 19:10:29 OK 20241027025146_init.sql (2.26s)
   2025/11/26 19:10:29 OK 20250112202302_sync_children.go (46.09ms)
   2025/11/26 19:10:29 OK 20250419184104_merge_docs_attachments.sql (45.85ms)
   2025/11/26 19:10:29 OK 20250619215101_add_thumbnails.sql (17.91ms)
   2025/11/26 19:10:29 OK 20250625120010_add_mime_type.sql (9.93ms)
   2025/11/26 19:10:29 OK 20250629112901_fix_thumbnail_foriegn.sql (20.65ms)
   2025/11/26 19:10:29 goose: successfully migrated database to version: 20250629112901
   7:10PM INF ../go/src/app/app/api/handlers/v1/v1_ctrl_auth.go:98 > registering auth provider name=local
   7:10PM INF ../go/src/app/app/api/main.go:232 > Server is running on :7745

4. stop app

5. run python3 migrate_homebox_full.py from linux system, set PASSWORD and HOST to appropriate values

• console messages
  Migrating table: auth_roles
  Migrating table: sqlite_sequence
  Migrating table: auth_tokens
  Migrating table: groups
  Migrating table: group_invitation_tokens
  Migrating table: item_fields
  Migrating table: labels
  Migrating table: locations
  Migrating table: maintenance_entries
  Migrating table: notifiers
  Migrating table: users
  Migrating table: label_items
  Migrating table: items
  Migrating table: attachments
  Creating goose_db_version table
  Migration complete!

6. start app

migrate_homebox_full.py ----

#!/usr/bin/env python3
import sqlite3
import psycopg2
from psycopg2.extras import execute_values

# -----------------------------
# Configuration
# -----------------------------
SQLITE_FILE = "/mnt/qnap/homebox/homebox.db"
PG_CONN_STR = "dbname=homebox user=homebox password=PASSWORD host=HOST port=5432"

# Tables to skip or recreate manually
SKIP_TABLES = ["goose_db_version"]

# -----------------------------
# Connect to databases
# -----------------------------
sqlite_conn = sqlite3.connect(SQLITE_FILE)
sqlite_conn.row_factory = sqlite3.Row
sqlite_cur = sqlite_conn.cursor()

pg_conn = psycopg2.connect(PG_CONN_STR)
pg_cur = pg_conn.cursor()

# -----------------------------
# Helper: convert SQLite types to Postgres
# -----------------------------
def sqlite_type_to_pg(sqlite_type):
    t = sqlite_type.lower()
    if "int" in t:
        return "BIGINT"
    elif "bool" in t:
        return "BOOLEAN"
    elif "datetime" in t or "timestamp" in t:
        return "TIMESTAMPTZ"
    elif "real" in t or "double" in t or "float" in t:
        return "DOUBLE PRECISION"
    else:
        return "TEXT"

# -----------------------------
# Fetch tables
# -----------------------------
sqlite_cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [row["name"] for row in sqlite_cur.fetchall() if row["name"] not in SKIP_TABLES]

# -----------------------------
# Migrate schema and data
# -----------------------------
for table in tables:
    print(f"Migrating table: {table}")
    # Get column info
    sqlite_cur.execute(f"PRAGMA table_info({table})")
    cols = sqlite_cur.fetchall()
    col_defs = []
    col_names = []
    for col in cols:
        name = col["name"]
        typ = sqlite_type_to_pg(col["type"])
        col_def = f'"{name}" {typ}'
        if col["pk"]:
            col_def += " PRIMARY KEY"
        col_defs.append(col_def)
        col_names.append(f'"{name}"')

    # Create table in Postgres
    create_sql = f'CREATE TABLE IF NOT EXISTS "{table}" ({", ".join(col_defs)});'
    pg_cur.execute(create_sql)
    pg_conn.commit()

    # Copy data
    sqlite_cur.execute(f"SELECT * FROM {table}")
    rows = sqlite_cur.fetchall()
    if rows:
        values = [[row[col["name"]] for col in cols] for row in rows]
        insert_sql = f'INSERT INTO "{table}" ({", ".join(col_names)}) VALUES %s'
        execute_values(pg_cur, insert_sql, values)
        pg_conn.commit()

# -----------------------------
# Recreate goose_db_version manually
# -----------------------------
print("Creating goose_db_version table")
pg_cur.execute("""
DROP TABLE IF EXISTS goose_db_version;
CREATE TABLE goose_db_version (
    id BIGSERIAL PRIMARY KEY,
    version_id BIGINT NOT NULL,
    is_applied BIGINT NOT NULL,
    tstamp TIMESTAMPTZ DEFAULT now()
);
""")
pg_conn.commit()

# -----------------------------
# Cleanup
# -----------------------------
pg_cur.close()
pg_conn.close()
sqlite_cur.close()
sqlite_conn.close()

print("Migration complete!")

[lhns]

Thank you for sharing your migration script. I just had to change a few things for the newest version (0.23.1).
I installed this in the homebox docker container:

apk add python3 py-pip
pip install --break-system-packages psycopg2-binary

Here is the tweaked migration script

#!/usr/bin/env python3
import sqlite3
import json
import psycopg2
from psycopg2.extras import Json
from psycopg2.extras import execute_values

# -----------------------------
# Configuration
# -----------------------------
SQLITE_FILE = "/data/homebox.db"
PG_CONN_STR = "dbname=homebox user=homebox password=pw host=db port=5432"

# Tables to skip or recreate manually
SKIP_TABLES = ["goose_db_version"]
LAST_TABLES = ["user_groups", "auth_tokens", "auth_roles"]

# -----------------------------
# Connect to databases
# -----------------------------
sqlite_conn = sqlite3.connect(SQLITE_FILE)
sqlite_conn.row_factory = sqlite3.Row
sqlite_cur = sqlite_conn.cursor()

pg_conn = psycopg2.connect(PG_CONN_STR)
pg_cur = pg_conn.cursor()

# -----------------------------
# Helper: convert SQLite types to Postgres
# -----------------------------
def sqlite_type_to_pg(sqlite_type):
    t = sqlite_type.lower()
    if "int" in t:
        return "BIGINT"
    elif "bool" in t:
        return "BOOLEAN"
    elif "datetime" in t or "timestamp" in t:
        return "TIMESTAMPTZ"
    elif "real" in t or "double" in t or "float" in t:
        return "DOUBLE PRECISION"
    else:
        return "TEXT"

# -----------------------------
# Fetch tables
# -----------------------------
sqlite_cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [row["name"] for row in sqlite_cur.fetchall() if row["name"] not in SKIP_TABLES]
front = [t for t in tables if t not in LAST_TABLES]
back = [t for t in LAST_TABLES if t in tables]
tables = front + back

# -----------------------------
# Migrate schema and data
# -----------------------------
for table in tables:
    print(f"Migrating table: {table}")
    # Get column info
    sqlite_cur.execute(f"PRAGMA table_info({table})")
    cols = sqlite_cur.fetchall()
    col_defs = []
    col_names = []
    for col in cols:
        name = col["name"]
        typ = sqlite_type_to_pg(col["type"])
        col_def = f'"{name}" {typ}'
        if col["pk"]:
            col_def += " PRIMARY KEY"
        col_defs.append(col_def)
        col_names.append(f'"{name}"')

    # Create table in Postgres
    if table in ["sqlite_sequence"]:
      create_sql = f'CREATE TABLE IF NOT EXISTS "{table}" ({", ".join(col_defs)});'
      pg_cur.execute(create_sql)
      pg_conn.commit()

    # Copy data
    sqlite_cur.execute(f"SELECT * FROM {table}")
    rows = sqlite_cur.fetchall()
    def get_col_val(row, col_name):
        value = row[col_name]
        if col_name in ["insured", "archived", "lifetime_warranty", "sync_child_items_locations", "boolean_value", "primary", "is_superuser", "superuser", "default_insured", "default_archived", "default_lifetime_warranty", "include_warranty_fields", "include_purchase_fields", "include_sold_fields"]:
            if value == 0:
                return False
            if value == 1:
                return True
        elif col_name in ["default_tag_ids"]:
            decoded = value.decode("utf-8")
            return Json(json.loads(decoded))
        return value
    
    if rows:
        values = [[get_col_val(row, col["name"]) for col in cols] for row in rows]
        insert_sql = f'INSERT INTO "{table}" ({", ".join(col_names)}) VALUES %s'
        execute_values(pg_cur, insert_sql, values)
        pg_conn.commit()

# -----------------------------
# Cleanup
# -----------------------------
pg_cur.close()
pg_conn.close()
sqlite_cur.close()
sqlite_conn.close()

print("Migration complete!")