Drop the restore's foreign keys in the database, not in the ORM metadata

Restoring a SQLite backup into PostgreSQL died part-way with

  insert or update on table "library_files" violates foreign key
  constraint "library_files_folder_id_fkey"
  DETAIL: Key (folder_id)=(1) is not present in table "library_folders".

The import recreates the schema and is supposed to create every table
without foreign keys, so the order rows arrive in cannot matter; the
constraints are added back once the data has landed. Phase 1 did that by
discarding each ForeignKeyConstraint from table.constraints before
create_all -- which only suppresses the inline REFERENCES clause.
Table.foreign_key_constraints is derived from the columns' ForeignKey
objects and was never touched, and when create_all meets a dependency
cycle it cannot sort, it falls back to emitting those tables' keys as
separate ALTER TABLE ... ADD FOREIGN KEY statements read from exactly
that property.

library_files, library_folders and print_archives form such a cycle, so
twelve constraints survived across the three of them -- measured against
a real PostgreSQL by running the old phase verbatim. The same cycle also
costs those tables their place in sorted_tables, so they were imported
alphabetically, putting library_files ahead of the library_folders rows
its folder_id references.

Phase 1 now creates the tables normally and drops every foreign key from
pg_constraint afterwards, in the same transaction, scoped to contype 'f'
in the public schema. That is indifferent to how create_all chose to
emit them, so a future cycle between other tables cannot bring this
back. Phase 3 is unchanged.

This also removes a second fault: the keys were stripped from the
process-wide Base.metadata and only restored after the drop/create
transaction, so a failure in between left the running app without them
until restart. The metadata is no longer modified at all.

Verified end to end against a real PostgreSQL -- a backup whose child
rows import before their parents restores cleanly, with all 90
constraints back afterwards. Four regression tests added.

backend/app/api/routes/settings.py

@@ -815,15 +815,8 @@ async def _import_sqlite_to_postgres(sqlite_path: Path, postgres_url: str):
         sorted_tables = [t.name for t in metadata.sorted_tables if t.name in tables_to_import]
 
         # Phase 1: Drop all tables and recreate WITHOUT foreign keys.
-        # This avoids all FK ordering/orphan issues during import.
-        saved_fks = {}
-        for table in metadata.sorted_tables:
-            fks = list(table.foreign_key_constraints)
-            if fks:
-                saved_fks[table.name] = fks
-                for fk in fks:
-                    table.constraints.discard(fk)
-
+        # This avoids all FK ordering/orphan issues during import; the
+        # constraints go back on at the end, once every row has landed.
         async with pg_engine.begin() as conn:
             # Cap how long DROP TABLE will wait for AccessExclusiveLock so
             # any residual concurrent writer (per-printer MQTT clients
@@ -856,11 +849,38 @@ async def _import_sqlite_to_postgres(sqlite_path: Path, postgres_url: str):
             )
             await conn.run_sync(metadata.create_all)
 
-        # Restore FK definitions in metadata (needed for re-adding later)
-        for table_name, fks in saved_fks.items():
-            table_obj = metadata.tables[table_name]
-            for fk in fks:
-                table_obj.constraints.add(fk)
+            # Now strip the foreign keys, at the database level.
+            #
+            # This used to be done by discarding each ForeignKeyConstraint
+            # from `table.constraints` before `create_all`. That only
+            # suppresses the inline REFERENCES clause inside CREATE TABLE:
+            # `Table.foreign_key_constraints` is derived from the *columns'*
+            # ForeignKey objects, which the discard never touched. When
+            # `create_all` meets a dependency cycle it can't sort -- and
+            # library_files / library_folders / print_archives are exactly
+            # such a cycle -- it falls back to emitting those tables' keys
+            # as separate ALTER TABLE ... ADD FOREIGN KEY statements read
+            # straight from that property. Twelve constraints survived,
+            # including library_files.folder_id, and because the same cycle
+            # also drops the ordering edge from `sorted_tables` the child
+            # table was imported before its parent and the restore died on
+            # a ForeignKeyViolationError.
+            #
+            # Dropping them from pg_constraint instead is indifferent to how
+            # create_all chose to emit them, so a future model cycle cannot
+            # reintroduce this. It also keeps the app's global Base.metadata
+            # untouched: the old code only put the constraints back *after*
+            # the transaction, so a failure in here left the running process
+            # with an FK-less metadata until restart.
+            await conn.execute(
+                text(
+                    "DO $$ DECLARE r RECORD; BEGIN "
+                    "FOR r IN (SELECT conrelid::regclass AS tbl, conname FROM pg_constraint "
+                    "WHERE contype = 'f' AND connamespace = 'public'::regnamespace) LOOP "
+                    "EXECUTE 'ALTER TABLE ' || r.tbl || ' DROP CONSTRAINT ' || quote_ident(r.conname); "
+                    "END LOOP; END $$;"
+                )
+            )
 
         # Phase 2: Import data (no FKs to worry about)
         async with pg_engine.begin() as conn:
@@ -958,7 +978,7 @@ async def _import_sqlite_to_postgres(sqlite_path: Path, postgres_url: str):
         src.close()
         logger.info("Cross-database import complete: %d tables imported", len(tables_to_import))
 
-        # Recreate FK constraints from ORM metadata (not from saved definitions).
+        # Recreate FK constraints from ORM metadata, which Phase 1 left intact.
         # Use individual transactions so orphaned SQLite data doesn't block valid FKs.
         from sqlalchemy.schema import AddConstraint
 
@@ -968,11 +988,28 @@ async def _import_sqlite_to_postgres(sqlite_path: Path, postgres_url: str):
                 try:
                     async with pg_engine.begin() as fk_conn:
                         await fk_conn.execute(AddConstraint(fk))
-                except Exception:
-                    failed_fks.append(f"{table.name}.{fk.name}")
+                except Exception as e:
+                    # Name the constraint by what it links, not by `fk.name`:
+                    # these are unnamed in the ORM, so that field is None and
+                    # the warning used to read "print_archives.None" for every
+                    # one of the five keys on that table -- unusable for
+                    # working out which rows to go and look at.
+                    cols = ", ".join(c.name for c in fk.columns)
+                    target = fk.elements[0].target_fullname if fk.elements else "unknown"
+                    failed_fks.append(f"{table.name}({cols}) -> {target}")
+                    # Postgres puts the offending key in a DETAIL line; it
+                    # names the exact orphan value, which is the one thing
+                    # that turns this into an actionable report.
+                    detail = next(
+                        (ln.strip() for ln in str(e).splitlines() if ln.startswith("DETAIL:")),
+                        str(e).splitlines()[0] if str(e) else e.__class__.__name__,
+                    )
+                    logger.info("FK %s(%s) -> %s not restored: %s", table.name, cols, target, detail)
         if failed_fks:
             logger.warning(
-                "Could not restore %d FK constraints (orphaned data in SQLite): %s",
+                "Could not restore %d FK constraints (orphaned data in the backup): %s. "
+                "The data is restored and usable; those columns are simply no longer "
+                "enforced. See the INFO lines above for the offending key in each case.",
                 len(failed_fks),
                 ", ".join(failed_fks),
             )

backend/tests/unit/test_postgres_restore_drop_cascade.py

@@ -15,10 +15,14 @@ so orphan tables can no longer block the restore.
 
 These tests guard against a regression to `metadata.drop_all` (which
 would re-introduce the bug for any user with orphan tables).
+
+The second half of the file covers the follow-on fix: the recreated
+tables must carry no foreign keys at all while rows are being imported.
 """
 
 from __future__ import annotations
 
+import logging
 import sqlite3
 import tempfile
 from pathlib import Path
@@ -35,6 +39,9 @@ def _make_sqlite_source() -> Path:
     conn = sqlite3.connect(str(path))
     # `users` is in the ORM metadata so `tables_to_import` is non-empty.
     conn.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, username TEXT)")
+    # At least one row, so the import actually emits an INSERT -- the
+    # loop skips empty tables outright.
+    conn.execute("INSERT INTO users (id, username) VALUES (1, 'alice')")
     conn.commit()
     conn.close()
     return path
@@ -161,3 +168,181 @@ async def test_restore_cascade_drop_targets_only_public_schema():
         assert "schemaname = '*'" not in cascade
     finally:
         sqlite_path.unlink(missing_ok=True)
+
+
+def _mock_pg_engine(
+    executed_sql: list[str],
+    create_all_error: Exception | None = None,
+    fk_error: Exception | None = None,
+):
+    """Build a fake async engine that records every statement, plus a
+    `run_sync:<fn>` marker, into `executed_sql` in execution order.
+
+    `fk_error` makes every ADD CONSTRAINT fail, standing in for a backup
+    carrying orphaned rows."""
+    from sqlalchemy.schema import AddConstraint
+
+    mock_conn = MagicMock()
+
+    def _execute(stmt, *a, **k):
+        if fk_error is not None and isinstance(stmt, AddConstraint):
+            raise fk_error
+        executed_sql.append(getattr(stmt, "text", str(stmt)))
+
+    mock_conn.execute = AsyncMock(side_effect=_execute)
+
+    async def _run_sync(fn, *args, **kw):
+        executed_sql.append("run_sync:" + getattr(fn, "__name__", repr(fn)))
+        if create_all_error is not None:
+            raise create_all_error
+        return None
+
+    mock_conn.run_sync = AsyncMock(side_effect=_run_sync)
+
+    begin_cm = MagicMock()
+    begin_cm.__aenter__ = AsyncMock(return_value=mock_conn)
+    begin_cm.__aexit__ = AsyncMock(return_value=False)
+
+    mock_engine = MagicMock()
+    mock_engine.begin = MagicMock(return_value=begin_cm)
+    mock_engine.dispose = AsyncMock()
+    return mock_engine
+
+
+def _fk_names(table) -> set[str]:
+    return {id(fk) for fk in table.constraints if hasattr(fk, "elements")}
+
+
+@pytest.mark.asyncio
+async def test_restore_drops_every_foreign_key_before_importing_rows():
+    """The recreated schema must carry no FK constraints while rows land.
+
+    Regression (#restore FK violation): the fix used to discard each
+    ForeignKeyConstraint from `table.constraints` before `create_all`.
+    That only suppresses the inline REFERENCES clause -- when `create_all`
+    hits a dependency cycle it cannot sort (library_files /
+    library_folders / print_archives are exactly such a cycle) it emits
+    those tables' keys as separate ALTER TABLE ... ADD FOREIGN KEY
+    statements read from `Table.foreign_key_constraints`, which the
+    discard never touched. The child table then imported before its
+    parent and Postgres raised ForeignKeyViolationError on
+    `library_files_folder_id_fkey`."""
+    from backend.app.api.routes import settings as settings_module
+
+    sqlite_path = _make_sqlite_source()
+    try:
+        executed_sql: list[str] = []
+        with patch(
+            "backend.app.core.database._create_engine",
+            new=MagicMock(return_value=_mock_pg_engine(executed_sql)),
+        ):
+            await settings_module._import_sqlite_to_postgres(sqlite_path, "postgresql+asyncpg://test/test")
+
+        fk_drops = [i for i, s in enumerate(executed_sql) if "pg_constraint" in s and "DROP CONSTRAINT" in s]
+        assert fk_drops, (
+            "Expected an unconditional DROP CONSTRAINT sweep over pg_constraint "
+            "so no foreign key survives create_all's cycle-breaking ALTER "
+            "TABLE statements. Captured SQL: " + "; ".join(s[:100] for s in executed_sql)
+        )
+        drop_sql = executed_sql[fk_drops[0]]
+        # Foreign keys only ('f'), scoped to public -- not PK/unique/check,
+        # and not another application's schema on a shared Postgres.
+        assert "contype = 'f'" in drop_sql, drop_sql
+        assert "'public'::regnamespace" in drop_sql, drop_sql
+
+        # It has to land after the tables exist and before the first row.
+        create_idx = executed_sql.index("run_sync:create_all")
+        insert_idx = next((i for i, s in enumerate(executed_sql) if s.startswith("INSERT INTO")), -1)
+        assert insert_idx > 0, f"no row import happened, so the ordering is untested: {executed_sql}"
+        assert create_idx < fk_drops[0] < insert_idx, (
+            f"FK drop must sit between create_all and the first INSERT: {executed_sql}"
+        )
+    finally:
+        sqlite_path.unlink(missing_ok=True)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("create_all_fails", [False, True])
+async def test_restore_never_mutates_the_process_wide_orm_metadata(create_all_fails):
+    """`Base.metadata` is global to the running app. The old code removed
+    every FK from it and only put them back *after* the drop/create
+    transaction, so a failure in there left the live process unable to
+    emit or re-add foreign keys until restart."""
+    from backend.app.api.routes import settings as settings_module
+    from backend.app.core.database import Base
+
+    table = Base.metadata.tables["library_files"]
+    before = _fk_names(table)
+    assert before, "library_files should carry FK constraints to begin with"
+
+    sqlite_path = _make_sqlite_source()
+    try:
+        executed_sql: list[str] = []
+        boom = RuntimeError("create_all exploded") if create_all_fails else None
+        engine = _mock_pg_engine(executed_sql, create_all_error=boom)
+        with patch("backend.app.core.database._create_engine", new=MagicMock(return_value=engine)):
+            if create_all_fails:
+                with pytest.raises(RuntimeError, match="create_all exploded"):
+                    await settings_module._import_sqlite_to_postgres(sqlite_path, "postgresql+asyncpg://test/test")
+            else:
+                await settings_module._import_sqlite_to_postgres(sqlite_path, "postgresql+asyncpg://test/test")
+
+        assert _fk_names(table) == before, "The restore must not add or remove constraints on the shared ORM metadata"
+    finally:
+        sqlite_path.unlink(missing_ok=True)
+
+
+@pytest.mark.asyncio
+async def test_unrestorable_fk_is_reported_by_its_columns(caplog):
+    """A key that can't go back on must be named by what it links.
+
+    These constraints are unnamed in the ORM, so `fk.name` is None and the
+    warning used to read "print_archives.None" once per failure -- five of
+    that table's keys share it, so the report said nothing about which
+    columns to inspect."""
+    from backend.app.api.routes import settings as settings_module
+
+    orphan = RuntimeError(
+        'violates foreign key constraint "library_files_folder_id_fkey"\n'
+        'DETAIL:  Key (folder_id)=(9) is not present in table "library_folders".'
+    )
+    sqlite_path = _make_sqlite_source()
+    try:
+        executed_sql: list[str] = []
+        engine = _mock_pg_engine(executed_sql, fk_error=orphan)
+        with (
+            patch("backend.app.core.database._create_engine", new=MagicMock(return_value=engine)),
+            caplog.at_level(logging.INFO, logger="backend.app.api.routes.settings"),
+        ):
+            await settings_module._import_sqlite_to_postgres(sqlite_path, "postgresql+asyncpg://test/test")
+
+        warning = next((r.getMessage() for r in caplog.records if r.levelno == logging.WARNING), None)
+        assert warning is not None, "a failed FK restore must be reported"
+        assert ".None" not in warning, f"constraints must not be named by fk.name: {warning}"
+        assert "library_files(folder_id) -> library_folders.id" in warning, warning
+        # And the offending value is recorded so the rows can be found.
+        assert any("Key (folder_id)=(9)" in r.getMessage() for r in caplog.records), (
+            "the Postgres DETAIL line names the orphan; it must survive into the log"
+        )
+    finally:
+        sqlite_path.unlink(missing_ok=True)
+
+
+def test_library_tables_form_an_fk_cycle():
+    """Documents why the restore cannot import in dependency order.
+
+    library_files -> library_folders -> print_archives -> library_files.
+    SQLAlchemy's `sorted_tables` gives up on these three and falls back to
+    alphabetical, which puts the child (library_files) before its parent.
+    Dropping the constraints outright is the only ordering-independent
+    answer; if this cycle is ever broken, the restore still works, but the
+    comment in `_import_sqlite_to_postgres` should be revisited."""
+    from backend.app.core.database import Base
+
+    def refs(name: str) -> set[str]:
+        table = Base.metadata.tables[name]
+        return {fk.column.table.name for fk in table.foreign_keys}
+
+    assert "library_folders" in refs("library_files")
+    assert "print_archives" in refs("library_folders")
+    assert "library_files" in refs("print_archives")