drop, migrate, then import

src/paperless_migration/scripts/wipe_db.py

@@ -0,0 +1,61 @@
+import django
+from django.apps import apps
+from django.db import connection
+from django.db.migrations.recorder import MigrationRecorder
+
+
+def _target_tables() -> list[str]:
+    tables = {
+        model._meta.db_table for model in apps.get_models(include_auto_created=True)
+    }
+    tables.add(MigrationRecorder.Migration._meta.db_table)
+    existing = set(connection.introspection.table_names())
+    return sorted(tables & existing)
+
+
+def _drop_sqlite_tables() -> None:
+    tables = _target_tables()
+    with connection.cursor() as cursor:
+        cursor.execute("PRAGMA foreign_keys=OFF;")
+        for table in tables:
+            cursor.execute(f'DROP TABLE IF EXISTS "{table}";')
+        cursor.execute("PRAGMA foreign_keys=ON;")
+
+
+def _drop_postgres_tables() -> None:
+    tables = _target_tables()
+    if not tables:
+        return
+    with connection.cursor() as cursor:
+        for table in tables:
+            cursor.execute(f'DROP TABLE IF EXISTS "{table}" CASCADE;')
+
+
+def _drop_mysql_tables() -> None:
+    tables = _target_tables()
+    with connection.cursor() as cursor:
+        cursor.execute("SET FOREIGN_KEY_CHECKS=0;")
+        for table in tables:
+            cursor.execute(f"DROP TABLE IF EXISTS `{table}`;")
+        cursor.execute("SET FOREIGN_KEY_CHECKS=1;")
+
+
+def main() -> None:
+    django.setup()
+    vendor = connection.vendor
+    print(f"Wiping database for {vendor}...")  # noqa: T201
+
+    if vendor == "sqlite":
+        _drop_sqlite_tables()
+    elif vendor == "postgresql":
+        _drop_postgres_tables()
+    elif vendor == "mysql":
+        _drop_mysql_tables()
+    else:
+        raise SystemExit(f"Unsupported database vendor: {vendor}")
+
+    print("Database wipe complete.")  # noqa: T201
+
+
+if __name__ == "__main__":
+    main()

src/paperless_migration/views.py

@@ -162,14 +162,6 @@ def import_stream(request):
     manage_path = Path(settings.BASE_DIR) / "manage.py"
     source_dir = export_path.parent
 
-    cmd = [
-        sys.executable,
-        str(manage_path),
-        "document_importer",
-        str(source_dir),
-        "--data-only",
-    ]
-
     env = os.environ.copy()
     env["DJANGO_SETTINGS_MODULE"] = "paperless.settings"
     env["PAPERLESS_MIGRATION_MODE"] = "0"
@@ -197,8 +189,10 @@ def import_stream(request):
             yield f"data: Failed to prepare import manifest: {exc}\n\n"
             return
 
+        def run_cmd(args, label):
+            yield f"data: {label}\n\n"
             process = subprocess.Popen(
-            cmd,
+                args,
                 stdout=subprocess.PIPE,
                 stderr=subprocess.STDOUT,
                 bufsize=1,
@@ -206,18 +200,59 @@ def import_stream(request):
                 env=env,
             )
             try:
-            yield "data: Starting import...\n\n"
                 if process.stdout:
                     for line in process.stdout:
                         yield f"data: {line.rstrip()}\n\n"
                 process.wait()
-            if process.returncode == 0:
+                return process.returncode
-                imported_marker.parent.mkdir(parents=True, exist_ok=True)
-                imported_marker.write_text("ok\n", encoding="utf-8")
-            yield f"data: Import finished with code {process.returncode}\n\n"
             finally:
                 if process and process.poll() is None:
                     process.kill()
+
+        wipe_cmd = [
+            sys.executable,
+            "-m",
+            "paperless_migration.scripts.wipe_db",
+        ]
+        migrate_cmd = [
+            sys.executable,
+            str(manage_path),
+            "migrate",
+            "--noinput",
+        ]
+        import_cmd = [
+            sys.executable,
+            str(manage_path),
+            "document_importer",
+            str(source_dir),
+            "--data-only",
+        ]
+        try:
+            wipe_code = yield from run_cmd(
+                wipe_cmd,
+                "Wiping database...",
+            )
+            if wipe_code != 0:
+                yield f"data: Wipe finished with code {wipe_code}\n\n"
+                return
+
+            migrate_code = yield from run_cmd(
+                migrate_cmd,
+                "Running migrations...",
+            )
+            if migrate_code != 0:
+                yield f"data: Migrate finished with code {migrate_code}\n\n"
+                return
+
+            import_code = yield from run_cmd(
+                import_cmd,
+                "Starting import...",
+            )
+            if import_code == 0:
+                imported_marker.parent.mkdir(parents=True, exist_ok=True)
+                imported_marker.write_text("ok\n", encoding="utf-8")
+            yield f"data: Import finished with code {import_code}\n\n"
+        finally:
             if backup_path and backup_path.exists():
                 try:
                     shutil.move(backup_path, export_path)