From a6746548af552940cc5d828909d82c9c3cc986c3 Mon Sep 17 00:00:00 2001 From: Diego Imbert Date: Wed, 1 Apr 2026 00:32:34 +0200 Subject: [PATCH] refactor: split import_pg_database into create_pg_database + import_pg_database Co-Authored-By: Claude Opus 4.5 --- .../windmill-api-workspaces/src/workspaces.rs | 165 ++++++++++-------- backend/windmill-api/openapi.yaml | 36 +++- .../ForkDatatableSection.svelte | 27 ++- 3 files changed, 148 insertions(+), 80 deletions(-) diff --git a/backend/windmill-api-workspaces/src/workspaces.rs b/backend/windmill-api-workspaces/src/workspaces.rs index 64f0ac15bb..932a02b010 100644 --- a/backend/windmill-api-workspaces/src/workspaces.rs +++ b/backend/windmill-api-workspaces/src/workspaces.rs @@ -157,6 +157,7 @@ pub fn workspaced_service() -> Router { post(reset_workspace_diffs), ) .route("/compare/{target_workspace_id}", get(compare_workspaces)) + .route("/create_pg_database", post(create_pg_database)) .route("/import_pg_database", post(import_pg_database)) .route("/export_pg_schema", post(export_pg_schema)) .route( @@ -1609,17 +1610,105 @@ async fn pg_import_dump(target_db: &PgDatabase, dump_file: &DumpFile) -> Result< Ok(()) } +#[derive(Deserialize)] +struct CreatePgDatabaseRequest { + /// The datatable source to determine connection info: 'datatable://name' or '$res:path' + source: String, + /// Name for the new database + target_dbname: String, +} + +/// Create a new PostgreSQL database. For instance datatables, creates on the Windmill PG instance. +/// For resource datatables, creates on the same server as the source. +async fn create_pg_database( + ApiAuthed { is_admin, username, .. }: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + Json(req): Json, +) -> Result { + require_admin(is_admin, &username)?; + windmill_common::validate_dbname(&req.target_dbname)?; + + // Determine if this is an instance or resource-backed datatable + let is_instance_datatable = if let Some(dt_name) = req.source.strip_prefix("datatable://") { + let config = sqlx::query_scalar!( + "SELECT datatable->'datatables'->$2 FROM workspace_settings WHERE workspace_id = $1", + &w_id, + dt_name + ) + .fetch_optional(&db) + .await? + .flatten(); + config + .and_then(|v| { + v.get("database") + .and_then(|d| d.get("resource_type")) + .and_then(|r| r.as_str()) + .map(|s| s == "instance") + }) + .unwrap_or(false) + } else { + false + }; + + if is_instance_datatable { + windmill_common::create_custom_instance_database(&db, &req.target_dbname, "datatable") + .await?; + } else { + let source_pg = resolve_pg_source(&db, &w_id, &req.source).await?; + let (client, connection) = source_pg.connect().await?; + let join_handle = tokio::spawn(async move { connection.await }); + + let row = client + .query_one( + "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1)", + &[&req.target_dbname], + ) + .await + .map_err(|e| { + Error::internal_err(format!("Failed to check database existence: {}", e)) + })?; + let db_exists: bool = row.get(0); + + if db_exists { + drop(client); + let _ = join_handle.await; + return Err(Error::BadRequest(format!( + "Database '{}' already exists on the resource server", + req.target_dbname + ))); + } + + client + .execute(&format!("CREATE DATABASE \"{}\"", &req.target_dbname), &[]) + .await + .map_err(|e| { + Error::internal_err(format!( + "Failed to create database '{}': {}", + req.target_dbname, e + )) + })?; + + drop(client); + join_handle + .await + .map_err(|e| Error::internal_err(format!("join error: {}", e)))? + .map_err(|e| Error::internal_err(format!("tokio_postgres error: {}", e)))?; + } + + Ok(format!("Created database '{}'", req.target_dbname)) +} + #[derive(Deserialize)] struct ImportPgDatabaseRequest { source: String, target: String, #[serde(default)] target_dbname_override: Option, - #[serde(default)] - create_target_db: bool, fork_behavior: DataTableForkBehavior, } +/// Import (pg_dump/pg_import) from source to target. Does NOT create the target database. async fn import_pg_database( ApiAuthed { is_admin, username, .. }: ApiAuthed, Extension(db): Extension, @@ -1647,78 +1736,6 @@ async fn import_pg_database( target_pg.dbname = override_dbname.clone(); } - windmill_common::validate_dbname(&target_pg.dbname)?; - - if req.create_target_db { - // Determine if this is an instance or resource-backed datatable - let is_instance_datatable = if let Some(dt_name) = req.target.strip_prefix("datatable://") { - let config = sqlx::query_scalar!( - "SELECT datatable->'datatables'->$2 FROM workspace_settings WHERE workspace_id = $1", - &w_id, - dt_name - ) - .fetch_optional(&db) - .await? - .flatten(); - config - .and_then(|v| { - v.get("database") - .and_then(|d| d.get("resource_type")) - .and_then(|r| r.as_str()) - .map(|s| s == "instance") - }) - .unwrap_or(false) - } else { - false - }; - - if is_instance_datatable { - // Instance datatable: create on Windmill PG instance - windmill_common::create_custom_instance_database(&db, &target_pg.dbname, "datatable") - .await?; - } else { - // Resource datatable or $res: CREATE DATABASE on the source server - let (client, connection) = source_pg.connect().await?; - let join_handle = tokio::spawn(async move { connection.await }); - - let row = client - .query_one( - "SELECT EXISTS (SELECT 1 FROM pg_catalog.pg_database WHERE datname = $1)", - &[&target_pg.dbname], - ) - .await - .map_err(|e| { - Error::internal_err(format!("Failed to check database existence: {}", e)) - })?; - let db_exists: bool = row.get(0); - - if db_exists { - drop(client); - let _ = join_handle.await; - return Err(Error::BadRequest(format!( - "Database '{}' already exists on the resource server", - target_pg.dbname - ))); - } - - client - .execute(&format!("CREATE DATABASE \"{}\"", &target_pg.dbname), &[]) - .await - .map_err(|e| { - Error::internal_err(format!( - "Failed to create database '{}': {}", - target_pg.dbname, e - )) - })?; - - drop(client); - join_handle - .await - .map_err(|e| Error::internal_err(format!("join error: {}", e)))? - .map_err(|e| Error::internal_err(format!("tokio_postgres error: {}", e)))?; - } - } - let dump_file = pg_dump_database(&source_pg, schema_only).await?; pg_import_dump(&target_pg, &dump_file).await?; diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index cdb0fd15f6..f91420ecb7 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -3579,9 +3579,40 @@ paths: application/json: schema: {} + /w/{workspace}/workspaces/create_pg_database: + post: + summary: create a new PostgreSQL database for a datatable + operationId: createPgDatabase + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: Create pg database request + required: true + content: + application/json: + schema: + type: object + required: [source, target_dbname] + properties: + source: + type: string + description: "Datatable source to determine connection info: 'datatable://name' or '$res:path'" + target_dbname: + type: string + description: "Name for the new database" + responses: + "200": + description: status + content: + text/plain: + schema: + type: string + /w/{workspace}/workspaces/import_pg_database: post: - summary: import a PostgreSQL database from source to target + summary: import a PostgreSQL database from source to target via pg_dump operationId: importPgDatabase tags: - workspace @@ -3605,9 +3636,6 @@ paths: target_dbname_override: type: string description: "Override the target database name" - create_target_db: - type: boolean - description: "When true, CREATE DATABASE is run before importing" fork_behavior: type: string enum: diff --git a/frontend/src/lib/components/workspaceSettings/ForkDatatableSection.svelte b/frontend/src/lib/components/workspaceSettings/ForkDatatableSection.svelte index d90297d04e..712091e2f4 100644 --- a/frontend/src/lib/components/workspaceSettings/ForkDatatableSection.svelte +++ b/frontend/src/lib/components/workspaceSettings/ForkDatatableSection.svelte @@ -68,7 +68,11 @@ const steps: ForkStep[] = [ { - label: `CREATE DATABASE "${newDbName}" + pg_dump → pg_import (${behavior === 'schema_only' ? 'schema only' : 'schema + data'})`, + label: `CREATE DATABASE "${newDbName}"`, + status: 'pending' + }, + { + label: `pg_dump → pg_import (${behavior === 'schema_only' ? 'schema only' : 'schema + data'})`, status: 'pending' }, { @@ -108,6 +112,26 @@ cloneRunning = true let stepIdx = 0 + // Step 1: Create the database + job.steps[stepIdx].status = 'running' + try { + await WorkspaceService.createPgDatabase({ + workspace: job._sourceWorkspace, + requestBody: { + source: `datatable://${job.name}`, + target_dbname: job._newDbName + } + }) + job.steps[stepIdx].status = 'done' + } catch (e: any) { + job.steps[stepIdx].status = 'error' + job.steps[stepIdx].error = e?.body ?? e?.message ?? String(e) + cloneRunning = false + return + } + stepIdx++ + + // Step 2: Import data job.steps[stepIdx].status = 'running' try { await WorkspaceService.importPgDatabase({ @@ -116,7 +140,6 @@ source: `datatable://${job.name}`, target: `datatable://${job.name}`, target_dbname_override: job._newDbName, - create_target_db: true, fork_behavior: job.behavior } })