diff --git a/backend/.sqlx/query-0a686ca61444d7ad7484071727aa039a6ea6697e5a49a633b767c052aa3e0a18.json b/backend/.sqlx/query-0a686ca61444d7ad7484071727aa039a6ea6697e5a49a633b767c052aa3e0a18.json index 053857a0a0..aa608ecc57 100644 --- a/backend/.sqlx/query-0a686ca61444d7ad7484071727aa039a6ea6697e5a49a633b767c052aa3e0a18.json +++ b/backend/.sqlx/query-0a686ca61444d7ad7484071727aa039a6ea6697e5a49a633b767c052aa3e0a18.json @@ -48,6 +48,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json b/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json index 03bd3e00c1..b2421d64f2 100644 --- a/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json +++ b/backend/.sqlx/query-254776959f60455a00a2c29a1783ee6b27c836197df6454d3bb5c59441c34f44.json @@ -69,6 +69,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-266741894b9a6ed650cf999b6e5b7c797424f8291517d3abf096fe45548a0273.json b/backend/.sqlx/query-266741894b9a6ed650cf999b6e5b7c797424f8291517d3abf096fe45548a0273.json new file mode 100644 index 0000000000..41879871e1 --- /dev/null +++ b/backend/.sqlx/query-266741894b9a6ed650cf999b6e5b7c797424f8291517d3abf096fe45548a0273.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT git_sync FROM workspace_settings WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "git_sync", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "266741894b9a6ed650cf999b6e5b7c797424f8291517d3abf096fe45548a0273" +} diff --git a/backend/.sqlx/query-2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c.json b/backend/.sqlx/query-2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c.json index 030a85c000..c9110033a5 100644 --- a/backend/.sqlx/query-2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c.json +++ b/backend/.sqlx/query-2f42460fdd8aa125c8fd46b3cd02e47f57de0f073d3ce3bc7d21a7e404a83b5c.json @@ -42,6 +42,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410.json b/backend/.sqlx/query-30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410.json new file mode 100644 index 0000000000..0efe3f75d5 --- /dev/null +++ b/backend/.sqlx/query-30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dependency_map\n WHERE importer_path = $1 AND importer_kind = 'script'\n AND workspace_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "30e9250edd0c4c6025322f51b758174435d6df4f835cb0883ba40897cb99a410" +} diff --git a/backend/.sqlx/query-4a393e61e33c1204c01044254692629e6ee2c053c7b1fa10f84462a327c739b3.json b/backend/.sqlx/query-4a393e61e33c1204c01044254692629e6ee2c053c7b1fa10f84462a327c739b3.json index cac6b88592..a8c925c966 100644 --- a/backend/.sqlx/query-4a393e61e33c1204c01044254692629e6ee2c053c7b1fa10f84462a327c739b3.json +++ b/backend/.sqlx/query-4a393e61e33c1204c01044254692629e6ee2c053c7b1fa10f84462a327c739b3.json @@ -28,6 +28,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72.json b/backend/.sqlx/query-5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72.json new file mode 100644 index 0000000000..c58167c782 --- /dev/null +++ b/backend/.sqlx/query-5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path)\n VALUES ($1, $2, 'script', $3)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "5fbd5dfb8df9aecce023c5d482ee58efcbc23c8682a3f0ec94977eb56023ed72" +} diff --git a/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json b/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json index 4598ac08df..89583ff5ef 100644 --- a/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json +++ b/backend/.sqlx/query-620ddf29c5e867079df4c2aa6e80bccb19beeb9ddfa308ca97f254cd5ba8157e.json @@ -62,6 +62,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json b/backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json index 1fa370e682..75b8108281 100644 --- a/backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json +++ b/backend/.sqlx/query-661f472ff3860983322162420457f5033b9c9afc344d9c3e385ba20a3ad2197a.json @@ -5,7 +5,7 @@ "columns": [ { "ordinal": 0, - "name": "bool", + "name": "?column?", "type_info": "Bool" } ], diff --git a/backend/.sqlx/query-6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a.json b/backend/.sqlx/query-6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a.json index 72c175aff7..7d9d04bc37 100644 --- a/backend/.sqlx/query-6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a.json +++ b/backend/.sqlx/query-6b313cc9a57ae3c943bda4a3213f7f6231a44b6ef5a52754074d136007f4f72a.json @@ -37,6 +37,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-76d73f2fe5ac75e5033fb93d60f2ef5ab2b0b73a80dcb6a250c17c3b2f15c6b4.json b/backend/.sqlx/query-76d73f2fe5ac75e5033fb93d60f2ef5ab2b0b73a80dcb6a250c17c3b2f15c6b4.json new file mode 100644 index 0000000000..97c52e993e --- /dev/null +++ b/backend/.sqlx/query-76d73f2fe5ac75e5033fb93d60f2ef5ab2b0b73a80dcb6a250c17c3b2f15c6b4.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dependency_map WHERE workspace_id = $1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "76d73f2fe5ac75e5033fb93d60f2ef5ab2b0b73a80dcb6a250c17c3b2f15c6b4" +} diff --git a/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json b/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json index f296c4afc6..983c1586f0 100644 --- a/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json +++ b/backend/.sqlx/query-9d3556319411a27a875bf6cf0e5eda837cc63e4d8be912c0b5bfeea4a0c8db2e.json @@ -42,6 +42,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f.json b/backend/.sqlx/query-aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f.json new file mode 100644 index 0000000000..808f531923 --- /dev/null +++ b/backend/.sqlx/query-aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT importer_path FROM dependency_map\n WHERE imported_path = $1 AND importer_kind = 'script'\n AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "importer_path", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "aa44311f9bae8253c6c483d122597ce7e470c6ba1905e7e877dca35d9d68968f" +} diff --git a/backend/.sqlx/query-b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42.json b/backend/.sqlx/query-b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42.json index f95d9d95b2..86a1402ad0 100644 --- a/backend/.sqlx/query-b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42.json +++ b/backend/.sqlx/query-b69891c25dd029b1a54e97ace292433e1485324ff7dc802fe75d21c8c6db1d42.json @@ -42,6 +42,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/.sqlx/query-deb2e4c21d4f9d0b9439780ca6ebee13450609b578c469cb437c5420ab520258.json b/backend/.sqlx/query-deb2e4c21d4f9d0b9439780ca6ebee13450609b578c469cb437c5420ab520258.json new file mode 100644 index 0000000000..54e871eb13 --- /dev/null +++ b/backend/.sqlx/query-deb2e4c21d4f9d0b9439780ca6ebee13450609b578c469cb437c5420ab520258.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT content FROM script WHERE hash = $1 AND workspace_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "content", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Int8", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "deb2e4c21d4f9d0b9439780ca6ebee13450609b578c469cb437c5420ab520258" +} diff --git a/backend/.sqlx/query-ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba.json b/backend/.sqlx/query-ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba.json index 73af3f369c..9ea1ab28f3 100644 --- a/backend/.sqlx/query-ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba.json +++ b/backend/.sqlx/query-ef132ac8d79579b08d7359789b6f22991f51e1c945efc2924df6253d62b83bba.json @@ -42,6 +42,7 @@ "bash", "postgresql", "nativets", + "Nativets", "bun", "mysql", "bigquery", diff --git a/backend/Cargo.lock b/backend/Cargo.lock index bad5816a98..6d244fe229 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -9471,6 +9471,7 @@ dependencies = [ "windmill-api", "windmill-api-client", "windmill-common", + "windmill-git-sync", "windmill-queue", "windmill-worker", ] @@ -9541,6 +9542,7 @@ dependencies = [ "uuid 1.6.1", "windmill-audit", "windmill-common", + "windmill-git-sync", "windmill-parser", "windmill-parser-py-imports", "windmill-queue", @@ -9610,6 +9612,19 @@ dependencies = [ "uuid 1.6.1", ] +[[package]] +name = "windmill-git-sync" +version = "1.235.0" +dependencies = [ + "rsmq_async", + "serde", + "serde_json", + "sqlx 0.7.3", + "tracing", + "windmill-common", + "windmill-queue", +] + [[package]] name = "windmill-parser" version = "1.235.1" @@ -9810,6 +9825,7 @@ dependencies = [ "uuid 1.6.1", "windmill-audit", "windmill-common", + "windmill-git-sync", "windmill-parser", "windmill-parser-bash", "windmill-parser-go", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index d8a0a9dce7..fcbfb0b8db 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -11,6 +11,7 @@ members = [ "./windmill-worker", "./windmill-common", "./windmill-audit", + "./windmill-git-sync", "./parsers/windmill-parser", "./parsers/windmill-parser-ts", "./parsers/windmill-parser-wasm", @@ -34,7 +35,7 @@ opt-level = 0 incremental = true [features] -enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise"] +enterprise = ["windmill-worker/enterprise", "windmill-queue/enterprise", "windmill-api/enterprise", "windmill-git-sync/enterprise"] enterprise_saml = ["windmill-api/enterprise_saml"] benchmark = ["windmill-api/benchmark", "windmill-worker/benchmark", "windmill-queue/benchmark"] flamegraph = ["windmill-common/flamegraph", "windmill-worker/flamegraph"] @@ -46,6 +47,7 @@ anyhow.workspace = true tokio.workspace = true dotenv.workspace = true windmill-common = { workspace = true, features = ["tracing_init"] } +windmill-git-sync.workspace = true windmill-api.workspace = true windmill-worker.workspace = true futures.workspace = true @@ -83,6 +85,7 @@ windmill-queue = { path = "./windmill-queue" } windmill-worker = { path = "./windmill-worker" } windmill-common = { path = "./windmill-common" } windmill-audit = { path = "./windmill-audit" } +windmill-git-sync = { path = "./windmill-git-sync" } windmill-parser = { path = "./parsers/windmill-parser" } windmill-parser-ts = { path = "./parsers/windmill-parser-ts" } windmill-parser-py = { path = "./parsers/windmill-parser-py" } @@ -153,7 +156,7 @@ swc_ecma_visit = "0.93.7" base64 = "0.21.0" hmac = "0.12.1" sha2 = "0.10.6" -sqlx = { version = "^0", features = [ +sqlx = { version = "0.7.3", features = [ "macros", "migrate", "uuid", diff --git a/backend/migrations/20231227014107_add_dependency_map.down.sql b/backend/migrations/20231227014107_add_dependency_map.down.sql new file mode 100644 index 0000000000..ec6a4f8b45 --- /dev/null +++ b/backend/migrations/20231227014107_add_dependency_map.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +DROP TABLE dependency_map; \ No newline at end of file diff --git a/backend/migrations/20231227014107_add_dependency_map.up.sql b/backend/migrations/20231227014107_add_dependency_map.up.sql new file mode 100644 index 0000000000..7014d0bd63 --- /dev/null +++ b/backend/migrations/20231227014107_add_dependency_map.up.sql @@ -0,0 +1,14 @@ +-- Add up migration script here + +CREATE TYPE IMPORTER_KIND AS ENUM ('script', 'flow', 'app'); + +CREATE TABLE IF NOT EXISTS dependency_map ( + workspace_id VARCHAR(50) NOT NULL, + importer_path VARCHAR(510) NOT NULL, + importer_kind IMPORTER_KIND NOT NULL, + imported_path VARCHAR(510) NOT NULL, + PRIMARY KEY (workspace_id, importer_path, importer_kind, imported_path) +); + +CREATE UNIQUE INDEX IF NOT EXISTS dependency_map_imported_path_idx ON dependency_map (workspace_id, imported_path); + diff --git a/backend/parsers/windmill-parser-py-imports/src/lib.rs b/backend/parsers/windmill-parser-py-imports/src/lib.rs index f221595efe..12ad11d534 100644 --- a/backend/parsers/windmill-parser-py-imports/src/lib.rs +++ b/backend/parsers/windmill-parser-py-imports/src/lib.rs @@ -63,7 +63,7 @@ lazy_static! { fn process_import(module: Option, path: &str, level: usize) -> Vec { if level > 0 { - let mut imports = vec!["requests".to_string()]; + let mut imports = vec![]; let splitted_path = path.split("/"); let base = splitted_path .clone() @@ -78,10 +78,7 @@ fn process_import(module: Option, path: &str, level: usize) -> Vec, path: &str, level: usize) -> Vec error::Result> { + let nimports = parse_code_for_imports(code, path)?; + return Ok(nimports + .into_iter() + .filter_map(|x| { + if x.starts_with("relative:") { + Some(x.replace("relative:", "")) + } else { + None + } + }) + .collect()); +} + +fn parse_code_for_imports(code: &str, path: &str) -> error::Result> { + let code = code.split(DEF_MAIN).next().unwrap_or(""); + let ast = Suite::parse(code, "main.py").map_err(|e| { + error::Error::ExecutionErr(format!("Error parsing code: {}", e.to_string())) + })?; + let nimports: Vec = ast + .into_iter() + .filter_map(|x| match x { + Stmt::Import(StmtImport { names, .. }) => Some( + names + .into_iter() + .map(|x| { + let name = x.name.to_string(); + process_import(Some(name), path, 0) + }) + .flatten() + .collect::>(), + ), + Stmt::ImportFrom(StmtImportFrom { level: Some(i), module, .. }) if i.to_u32() > 0 => { + Some(process_import( + module.map(|x| x.to_string()), + path, + i.to_usize(), + )) + } + Stmt::ImportFrom(StmtImportFrom { level: _, module, .. }) => { + Some(process_import(module.map(|x| x.to_string()), path, 0)) + } + _ => None, + }) + .flatten() + .filter(|x| !STDIMPORTS.contains(&x.as_str())) + .unique() + .collect(); + return Ok(nimports); +} + #[async_recursion] pub async fn parse_python_imports( code: &str, @@ -127,41 +175,7 @@ pub async fn parse_python_imports( imports.extend(lines); } - let code = code.split(DEF_MAIN).next().unwrap_or(""); - let ast = Suite::parse(code, "main.py").map_err(|e| { - error::Error::ExecutionErr(format!("Error parsing code: {}", e.to_string())) - })?; - let nimports: Vec = ast - .into_iter() - .filter_map(|x| match x { - Stmt::Import(StmtImport { names, .. }) => Some( - names - .into_iter() - .map(|x| { - let name = x.name.to_string(); - process_import(Some(name), path, 0) - }) - .flatten() - .collect::>(), - ), - Stmt::ImportFrom(StmtImportFrom { level: Some(i), module, .. }) - if i.to_u32() > 0 => - { - Some(process_import( - module.map(|x| x.to_string()), - path, - i.to_usize(), - )) - } - Stmt::ImportFrom(StmtImportFrom { level: _, module, .. }) => { - Some(process_import(module.map(|x| x.to_string()), path, 0)) - } - _ => None, - }) - .flatten() - .filter(|x| !STDIMPORTS.contains(&x.as_str())) - .unique() - .collect(); + let nimports = parse_code_for_imports(code, path)?; for n in nimports.iter() { let nested = if n.starts_with("relative:") { let rpath = n.replace("relative:", ""); diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 37d205c337..a3a3033d95 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -9,7 +9,7 @@ name = "windmill_api" path = "src/lib.rs" [features] -enterprise = ["windmill-queue/enterprise", "async-stripe", "windmill-audit/enterprise"] +enterprise = ["windmill-queue/enterprise", "async-stripe", "windmill-audit/enterprise", "windmill-git-sync/enterprise"] enterprise_saml = ["samael"] benchmark = [] @@ -27,6 +27,7 @@ windmill-common = { workspace = true, features = [ windmill-audit.workspace = true windmill-parser.workspace = true windmill-parser-py-imports.workspace = true +windmill-git-sync.workspace = true tokio.workspace = true anyhow.workspace = true argon2.workspace = true diff --git a/backend/windmill-api/src/apps.rs b/backend/windmill-api/src/apps.rs index 7bacf8a222..710747f841 100644 --- a/backend/windmill-api/src/apps.rs +++ b/backend/windmill-api/src/apps.rs @@ -9,7 +9,6 @@ use std::collections::HashMap; */ use crate::{ db::{ApiAuthed, DB}, - deployment_metadata_helpers, users::{require_owner_of_path, OptAuthed}, variables::build_crypt, webhook_util::{WebhookMessage, WebhookShared}, @@ -588,12 +587,16 @@ async fn create_app( ) .await?; - let mut tx = PushIsolationLevel::Transaction(tx); + let tx = PushIsolationLevel::Transaction(tx); let (dependency_job_uuid, new_tx) = push( &db, tx, &w_id, - JobPayload::AppDependencies { path: app.path.clone(), version: v_id }, + JobPayload::AppDependencies { + path: app.path.clone(), + version: v_id, + deployment_message: app.deployment_message, + }, PushArgs::empty(), &authed.username, &authed.email, @@ -615,25 +618,7 @@ async fn create_app( .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); - tx = PushIsolationLevel::Transaction(new_tx); - tx = deployment_metadata_helpers::handle_deployment_metadata( - tx, - &authed, - &db, - &w_id, - deployment_metadata_helpers::DeployedObject::App { path: app.path.clone(), version: v_id }, - app.deployment_message, - ) - .await?; - - match tx { - PushIsolationLevel::Transaction(tx) => tx.commit().await?, - _ => { - return Err(Error::InternalErr( - "Expected a transaction here".to_string(), - )); - } - } + new_tx.commit().await?; webhook.send_message( w_id.clone(), @@ -849,14 +834,18 @@ async fn update_app( ) .await?; - let mut tx: PushIsolationLevel<'_, rsmq_async::MultiplexedRsmq> = + let tx: PushIsolationLevel<'_, rsmq_async::MultiplexedRsmq> = PushIsolationLevel::Transaction(tx); if let Some(v_id) = v_id { let (dependency_job_uuid, new_tx) = push( &db, tx, &w_id, - JobPayload::AppDependencies { path: npath.clone(), version: v_id }, + JobPayload::AppDependencies { + path: npath.clone(), + version: v_id, + deployment_message: ns.deployment_message, + }, PushArgs::empty(), &authed.username, &authed.email, @@ -877,26 +866,7 @@ async fn update_app( ) .await?; tracing::info!("Pushed app dependency job {}", dependency_job_uuid); - - tx = PushIsolationLevel::Transaction(new_tx); - tx = deployment_metadata_helpers::handle_deployment_metadata( - tx, - &authed, - &db, - &w_id, - deployment_metadata_helpers::DeployedObject::App { path: npath.clone(), version: v_id }, - ns.deployment_message, - ) - .await?; - } - - match tx { - PushIsolationLevel::Transaction(tx) => tx.commit().await?, - _ => { - return Err(Error::InternalErr( - "Expected a transaction here".to_string(), - )); - } + new_tx.commit().await?; } webhook.send_message( diff --git a/backend/windmill-api/src/flows.rs b/backend/windmill-api/src/flows.rs index 6ff5d8ed8b..b69493dc0e 100644 --- a/backend/windmill-api/src/flows.rs +++ b/backend/windmill-api/src/flows.rs @@ -7,7 +7,6 @@ */ use crate::db::ApiAuthed; -use crate::deployment_metadata_helpers; use crate::{ db::DB, schedule::clear_schedule, @@ -355,7 +354,7 @@ async fn create_flow( ) .await?; - let mut tx = PushIsolationLevel::Transaction(tx); + let tx = PushIsolationLevel::Transaction(tx); let (dependency_job_uuid, mut new_tx) = push( &db, tx, @@ -363,6 +362,7 @@ async fn create_flow( JobPayload::FlowDependencies { path: nf.path.clone(), dedicated_worker: nf.dedicated_worker, + deployment_message: nf.deployment_message, }, PushArgs::empty(), &authed.username, @@ -393,26 +393,7 @@ async fn create_flow( .execute(&mut new_tx) .await?; - tx = PushIsolationLevel::Transaction(new_tx); - tx = deployment_metadata_helpers::handle_deployment_metadata( - tx, - &authed, - &db, - &w_id, - deployment_metadata_helpers::DeployedObject::Flow { path: nf.path.clone() }, - nf.deployment_message, - ) - .await?; - - match tx { - PushIsolationLevel::Transaction(tx) => tx.commit().await?, - _ => { - return Err(Error::InternalErr( - "Expected a transaction here".to_string(), - )); - } - } - + new_tx.commit().await?; webhook.send_message( w_id.clone(), WebhookMessage::CreateFlow { workspace: w_id.clone(), path: nf.path.clone() }, @@ -585,7 +566,7 @@ async fn update_flow( }, ); - let mut tx = PushIsolationLevel::Transaction(tx); + let tx = PushIsolationLevel::Transaction(tx); let (dependency_job_uuid, mut new_tx) = push( &db, @@ -594,6 +575,7 @@ async fn update_flow( JobPayload::FlowDependencies { path: nf.path.clone(), dedicated_worker: nf.dedicated_worker, + deployment_message: nf.deployment_message, }, PushArgs::empty(), &authed.username, @@ -631,25 +613,7 @@ async fn update_flow( .await?; } - tx = PushIsolationLevel::Transaction(new_tx); - tx = deployment_metadata_helpers::handle_deployment_metadata( - tx, - &authed, - &db, - &w_id, - deployment_metadata_helpers::DeployedObject::Flow { path: nf.path.clone() }, - nf.deployment_message, - ) - .await?; - - match tx { - PushIsolationLevel::Transaction(tx) => tx.commit().await?, - _ => { - return Err(Error::InternalErr( - "Expected a transaction here".to_string(), - )); - } - } + new_tx.commit().await?; Ok(nf.path.to_string()) } diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index aa067f89a6..7d7f49cbcc 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -49,7 +49,6 @@ mod audit; mod capture; mod configs; mod db; -mod deployment_metadata_helpers; mod drafts; pub mod ee; pub mod embeddings; diff --git a/backend/windmill-api/src/scripts.rs b/backend/windmill-api/src/scripts.rs index 7831aef244..f7ce0cf421 100644 --- a/backend/windmill-api/src/scripts.rs +++ b/backend/windmill-api/src/scripts.rs @@ -8,7 +8,6 @@ use crate::{ db::{ApiAuthed, DB}, - deployment_metadata_helpers, schedule::clear_schedule, users::{maybe_refresh_folders, require_owner_of_path, AuthCache}, webhook_util::{WebhookMessage, WebhookShared}, @@ -47,6 +46,7 @@ use windmill_common::{ not_found_if_none, paginate, query_elems_from_hub, require_admin, Pagination, StripPath, }, }; +use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; use windmill_queue::{ self, schedule::push_scheduled_job, PushArgs, PushIsolationLevel, QueueTransaction, }; @@ -325,7 +325,7 @@ async fn create_script( let script_path = ns.path.clone(); let hash = ScriptHash(hash_script(&ns)); let authed = maybe_refresh_folders(&ns.path, &w_id, authed, &db).await; - let mut tx: QueueTransaction<'_, _> = (rsmq, user_db.begin(&authed).await?).into(); + let mut tx: QueueTransaction<'_, _> = (rsmq.clone(), user_db.begin(&authed).await?).into(); if sqlx::query_scalar!( "SELECT 1 FROM script WHERE hash = $1 AND workspace_id = $2", @@ -604,17 +604,10 @@ async fn create_script( ); } - let mut tx = PushIsolationLevel::Transaction(tx); + let tx = PushIsolationLevel::Transaction(tx); + let permissioned_as = username_to_permissioned_as(&authed.username); if needs_lock_gen { - let dependencies = match ns.language { - ScriptLang::Python3 => { - windmill_parser_py_imports::parse_python_imports(&ns.content, &w_id, &ns.path, &db) - .await? - .join("\n") - } - _ => ns.content, - }; let tag = if ns.dedicated_worker.is_some_and(|x| x) { Some(format!("{}:{}", &w_id, &ns.path,)) } else { @@ -626,15 +619,15 @@ async fn create_script( &w_id, JobPayload::Dependencies { hash, - dependencies, language: ns.language, path: ns.path, dedicated_worker: ns.dedicated_worker, + deployment_message: ns.deployment_message, }, PushArgs::empty(), &authed.username, &authed.email, - username_to_permissioned_as(&authed.username), + permissioned_as, None, None, None, @@ -650,26 +643,18 @@ async fn create_script( None, ) .await?; - tx = PushIsolationLevel::Transaction(new_tx); - } - - tx = deployment_metadata_helpers::handle_deployment_metadata( - tx, - &authed, - &db, - &w_id, - deployment_metadata_helpers::DeployedObject::Script { hash: hash, path: script_path }, - ns.deployment_message, - ) - .await?; - - match tx { - PushIsolationLevel::Transaction(tx) => tx.commit().await?, - _ => { - return Err(Error::InternalErr( - "Expected a transaction here".to_string(), - )); - } + new_tx.commit().await?; + } else { + handle_deployment_metadata( + &authed.email, + &authed.username, + &db, + &w_id, + DeployedObject::Script { hash: hash, path: script_path }, + ns.deployment_message, + rsmq, + ) + .await?; } Ok((StatusCode::CREATED, format!("{}", hash))) diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 9a5ccfa3d8..b1e0ba4a88 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -44,6 +44,7 @@ use windmill_common::s3_helpers::LargeFileStorage; use windmill_common::schedule::Schedule; use windmill_common::users::username_to_permissioned_as; use windmill_common::worker::CLOUD_HOSTED; +use windmill_common::workspaces::WorkspaceGitRepo; use windmill_common::{ error::{to_anyhow, Error, JsonResult, Result}, flows::Flow, @@ -1045,13 +1046,6 @@ struct EditGitSyncConfig { git_sync_settings: Option, } -#[derive(Serialize, Deserialize, Debug)] -pub struct WorkspaceGitRepo { - pub script_path: String, - pub git_repo_resource_path: String, - pub use_individual_branch: Option, -} - async fn edit_git_sync_config( authed: ApiAuthed, Extension(db): Extension, @@ -1505,6 +1499,9 @@ async fn delete_workspace( let mut tx = db.begin().await?; require_super_admin(&db, &email).await?; + sqlx::query!("DELETE FROM dependency_map WHERE workspace_id = $1", &w_id) + .execute(&mut *tx) + .await?; sqlx::query!("DELETE FROM queue WHERE workspace_id = $1", &w_id) .execute(&mut *tx) .await?; diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 2595b7af75..6f2fc5abfd 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -283,17 +283,19 @@ pub enum JobPayload { Dependencies { path: String, hash: ScriptHash, - dependencies: String, language: ScriptLang, dedicated_worker: Option, + deployment_message: Option, }, FlowDependencies { path: String, dedicated_worker: Option, + deployment_message: Option, }, AppDependencies { path: String, version: i64, + deployment_message: Option, }, Flow { path: String, diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 342e72a406..091cbc0254 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -36,6 +36,7 @@ pub mod users; pub mod utils; pub mod variables; pub mod worker; +pub mod workspaces; #[cfg(feature = "tracing_init")] pub mod tracing_init; diff --git a/backend/windmill-common/src/users.rs b/backend/windmill-common/src/users.rs index a349d928f6..7675c8799e 100644 --- a/backend/windmill-common/src/users.rs +++ b/backend/windmill-common/src/users.rs @@ -8,6 +8,7 @@ pub const SUPERADMIN_SECRET_EMAIL: &str = "superadmin_secret@windmill.dev"; pub const SUPERADMIN_NOTIFICATION_EMAIL: &str = "superadmin_notification@windmill.dev"; +pub const SUPERADMIN_SYNC_EMAIL: &str = "superadmin_sync@windmill.dev"; pub fn username_to_permissioned_as(user: &str) -> String { if user.contains('@') { diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs new file mode 100644 index 0000000000..0a5a99bb03 --- /dev/null +++ b/backend/windmill-common/src/workspaces.rs @@ -0,0 +1,8 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Serialize, Deserialize, Debug)] +pub struct WorkspaceGitRepo { + pub script_path: String, + pub git_repo_resource_path: String, + pub use_individual_branch: Option, +} diff --git a/backend/windmill-git-sync/Cargo.toml b/backend/windmill-git-sync/Cargo.toml new file mode 100644 index 0000000000..074d3fec56 --- /dev/null +++ b/backend/windmill-git-sync/Cargo.toml @@ -0,0 +1,22 @@ +[package] +name = "windmill-git-sync" +version.workspace = true +authors.workspace = true +edition.workspace = true + +[lib] +name = "windmill_git_sync" +path = "./src/lib.rs" + +[features] +enterprise = ["windmill-queue/enterprise"] +default = [] + +[dependencies] +serde.workspace = true +sqlx.workspace = true +serde_json.workspace = true +tracing.workspace = true +windmill-common = { workspace = true, features = ["axum"] } +windmill-queue.workspace = true +rsmq_async.workspace = true \ No newline at end of file diff --git a/backend/windmill-api/src/deployment_metadata_helpers.rs b/backend/windmill-git-sync/src/lib.rs similarity index 54% rename from backend/windmill-api/src/deployment_metadata_helpers.rs rename to backend/windmill-git-sync/src/lib.rs index f0b32b2290..e246add07a 100644 --- a/backend/windmill-api/src/deployment_metadata_helpers.rs +++ b/backend/windmill-git-sync/src/lib.rs @@ -8,17 +8,18 @@ use std::collections::HashMap; -use crate::db::ApiAuthed; -use crate::db::DB; -use crate::workspaces::{WorkspaceGitRepo, WorkspaceSettings}; +use sqlx::{Pool, Postgres}; +use windmill_common::users::SUPERADMIN_SYNC_EMAIL; +use windmill_common::workspaces::WorkspaceGitRepo; use serde_json::json; use windmill_common::error::{Error, Result}; use windmill_common::jobs::JobPayload; use windmill_common::scripts::ScriptHash; -use windmill_common::users::username_to_permissioned_as; use windmill_queue::PushIsolationLevel; +pub type DB = Pool; + #[derive(Clone)] pub enum DeployedObject { Script { hash: ScriptHash, path: String }, @@ -37,13 +38,14 @@ impl DeployedObject { } pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send + 'c>( - mut tx: PushIsolationLevel<'c, R>, - authed: &ApiAuthed, + email: &str, + created_by: &str, db: &DB, w_id: &str, obj: DeployedObject, deployment_message: Option, -) -> Result> { + rsmq: Option, +) -> Result<()> { let skip_git_sync = if obj.get_path().starts_with("u/") { tracing::debug!( "Ignoring {} from git sync as it's in a private user folder", @@ -54,10 +56,10 @@ pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send false }; - let workspace_git_repo_setting = sqlx::query_as::<_, WorkspaceSettings>( - "SELECT * FROM workspace_settings WHERE workspace_id = $1", + let workspace_git_repo_setting = sqlx::query_scalar!( + "SELECT git_sync FROM workspace_settings WHERE workspace_id = $1", + w_id ) - .bind(&w_id) .fetch_optional(db) .await?; @@ -69,11 +71,10 @@ pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send let workspace_git_repo = workspace_git_repo_setting .unwrap() - .git_sync .map(|conf| serde_json::from_value::(conf).ok()) .flatten(); - let (git_sync_job_uuid, mut new_tx) = if !skip_git_sync && workspace_git_repo.is_some() { + let git_sync_job_uuid = if !skip_git_sync && workspace_git_repo.is_some() { let workspace_git_repo = workspace_git_repo.unwrap(); let mut args: HashMap = HashMap::new(); @@ -84,39 +85,56 @@ pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send .strip_prefix("$res:")), ); - let default_commit_msg: String; - match obj.clone() { + let message = match obj.clone() { DeployedObject::Script { path, .. } => { args.insert("path".to_string(), json!(path.to_string())); - default_commit_msg = format!("Script '{}' deployed", path); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Script '{}' deployed", path) + } else { + deployment_message.clone().unwrap() + } } DeployedObject::Flow { path } => { args.insert("path".to_string(), json!(path.to_string())); - default_commit_msg = format!("Flow '{}' deployed", path); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("Flow '{}' deployed", path) + } else { + deployment_message.clone().unwrap() + } } DeployedObject::App { path, .. } => { args.insert("path".to_string(), json!(path.to_string())); - default_commit_msg = format!("App '{}' deployed", path); + if deployment_message.as_ref().is_none() + || deployment_message.as_ref().is_some_and(|x| x.is_empty()) + { + format!("App '{}' deployed", path) + } else { + deployment_message.clone().unwrap() + } } - } - args.insert( - "commit_msg".to_string(), - json!(deployment_message.clone().unwrap_or(default_commit_msg)), - ); + }; + + args.insert("commit_msg".to_string(), json!(message)); args.insert( "use_individual_branch".to_string(), json!(workspace_git_repo.use_individual_branch.unwrap_or(false)), ); + let tx: PushIsolationLevel<'_, R> = PushIsolationLevel::IsolatedRoot(db.clone(), rsmq); + let (job_uuid, new_tx) = windmill_queue::push( - &db, + db, tx, &w_id, JobPayload::DeploymentCallback { path: workspace_git_repo.script_path.clone() }, args, - &authed.username, - &authed.email, - username_to_permissioned_as(&authed.username), + created_by, + &email, + SUPERADMIN_SYNC_EMAIL.to_string(), None, None, None, @@ -132,16 +150,10 @@ pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send None, ) .await?; - (Some(job_uuid), new_tx) + new_tx.commit().await?; + Some(job_uuid) } else { - let new_tx = match tx { - PushIsolationLevel::Isolated(user_db, authed, rsmq) => { - (rsmq, user_db.begin(&authed).await?).into() - } - PushIsolationLevel::IsolatedRoot(db, rsmq) => (rsmq, db.begin().await?).into(), - PushIsolationLevel::Transaction(tx) => tx, - }; - (None, new_tx) + None }; // We're not persisting the default commit msg as it's pretty useless. We will persist the ones manually set by users @@ -153,28 +165,27 @@ pub async fn handle_deployment_metadata<'c, R: rsmq_async::RsmqConnection + Send if deployment_message.is_some() || job_uuids.len() > 0 { // if the git sync job hasn't been triggered, and there is not custom deployment message, there's not point adding an entry to the table match obj.clone() { - DeployedObject::Script { path, hash, .. } => { - sqlx::query!( - "INSERT INTO deployment_metadata (workspace_id, path, script_hash, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)", - w_id, path, hash.0, &job_uuids, deployment_message, - ) - }, - DeployedObject::Flow { path } => { - sqlx::query!( - "INSERT INTO deployment_metadata (workspace_id, path, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4) ON CONFLICT (workspace_id, path) WHERE script_hash IS NULL AND app_version IS NULL DO UPDATE SET callback_job_ids = $3, deployment_msg = $4", - w_id, path, &job_uuids, deployment_message, - ) - } - DeployedObject::App { path, version } => { - sqlx::query!( - "INSERT INTO deployment_metadata (workspace_id, path, app_version, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)", - w_id, path, version, &job_uuids, deployment_message, - ) - } - }.execute(&mut new_tx) - .await?; + DeployedObject::Script { path, hash, .. } => { + sqlx::query!( + "INSERT INTO deployment_metadata (workspace_id, path, script_hash, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)", + w_id, path, hash.0, &job_uuids, deployment_message, + ) + }, + DeployedObject::Flow { path } => { + sqlx::query!( + "INSERT INTO deployment_metadata (workspace_id, path, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4) ON CONFLICT (workspace_id, path) WHERE script_hash IS NULL AND app_version IS NULL DO UPDATE SET callback_job_ids = $3, deployment_msg = $4", + w_id, path, &job_uuids, deployment_message, + ) + } + DeployedObject::App { path, version } => { + sqlx::query!( + "INSERT INTO deployment_metadata (workspace_id, path, app_version, callback_job_ids, deployment_msg) VALUES ($1, $2, $3, $4, $5)", + w_id, path, version, &job_uuids, deployment_message, + ) + } + }.execute(db) + .await?; } - tx = PushIsolationLevel::Transaction(new_tx); - return Ok(tx); + return Ok(()); } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 44e6f04996..b1d306f1b5 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -2496,21 +2496,23 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection dedicated_worker, None, ), - JobPayload::Dependencies { hash, dependencies, language, path, dedicated_worker } => ( - Some(hash.0), - Some(path), - Some((dependencies, None)), - JobKind::Dependencies, - None, - None, - Some(language), - None, - None, - None, - dedicated_worker, - None, - ), - JobPayload::FlowDependencies { path, dedicated_worker } => { + JobPayload::Dependencies { hash, language, path, dedicated_worker, deployment_message } => { + ( + Some(hash.0), + Some(path), + Some((deployment_message.unwrap_or_else(String::new), None)), + JobKind::Dependencies, + None, + None, + Some(language), + None, + None, + None, + dedicated_worker, + None, + ) + } + JobPayload::FlowDependencies { path, dedicated_worker, deployment_message } => { let value_json = fetch_scalar_isolated!( sqlx::query_scalar!( "SELECT value FROM flow WHERE path = $1 AND workspace_id = $2", @@ -2528,7 +2530,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection ( None, Some(path), - None, + Some((deployment_message.unwrap_or_else(String::new), None)), JobKind::FlowDependencies, Some(value.clone()), Some(FlowStatus::new(&value)), // this is a new flow being pushed, flow_status is set to flow_value @@ -2540,10 +2542,10 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection None, ) } - JobPayload::AppDependencies { path, version } => ( + JobPayload::AppDependencies { path, version, deployment_message } => ( Some(version), Some(path), - None, + Some((deployment_message.unwrap_or_else(String::new), None)), JobKind::AppDependencies, None, None, diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 74966c0a7e..de8fd33767 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -10,7 +10,7 @@ path = "src/lib.rs" [features] default = [] -enterprise = ["windmill-queue/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:sha2", "dep:tiberius", "dep:tokio-util"] +enterprise = ["windmill-queue/enterprise", "windmill-git-sync/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:sha2", "dep:tiberius", "dep:tokio-util"] benchmark = ["windmill-queue/benchmark"] flamegraph = [] @@ -31,6 +31,7 @@ windmill-parser-py-imports.workspace = true windmill-parser-bash.workspace = true windmill-parser-sql.workspace = true windmill-parser-graphql.workspace = true +windmill-git-sync.workspace = true sqlx.workspace = true uuid.workspace = true tracing.workspace = true diff --git a/backend/windmill-worker/loader.py b/backend/windmill-worker/loader.py index b94cece909..b04ac3580a 100644 --- a/backend/windmill-worker/loader.py +++ b/backend/windmill-worker/loader.py @@ -28,22 +28,23 @@ class WindmillFinder(MetaPathFinder): return ModuleSpec(name, WindmillLoader(name)) elif l > 2: script_path = "/".join(splitted) - import requests + import urllib.parse + import urllib.request + headers = {"Authorization": f"Bearer {os.environ.get('WM_TOKEN')}"} url = f"{os.environ.get('BASE_INTERNAL_URL')}/api/w/{os.environ.get('WM_WORKSPACE')}/scripts/raw/p/{script_path}.py" - r = requests.get( - url, headers={"Authorization": f"Bearer {os.environ.get('WM_TOKEN')}"} - ) - - if r.status_code == 200: - folder = os.getcwd() + "/tmp/" + "/".join(splitted[:-1]) - fullpath = folder + "/" + splitted[-1] + ".py" - os.makedirs(folder, exist_ok=True) - with open(fullpath, "w+") as f: - f.write(r.text) - return ModuleSpec(name, SourceFileLoader(name, fullpath)) - else: + req = urllib.request.Request(url, None, headers) + try: + with urllib.request.urlopen(req) as response: + r = response.read().decode("utf-8") + folder = os.getcwd() + "/tmp/" + "/".join(splitted[:-1]) + fullpath = folder + "/" + splitted[-1] + ".py" + os.makedirs(folder, exist_ok=True) + with open(fullpath, "w+") as f: + f.write(r) + return ModuleSpec(name, SourceFileLoader(name, fullpath)) + except: # raise ImportError(f"Script {script_path} not found") return ModuleSpec(name, WindmillLoader(name)) diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index b421b2ae28..e8a3ba2756 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -25,6 +25,8 @@ use std::{ }, time::Duration, }; +use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; +use windmill_parser_py_imports::parse_relative_imports; use uuid::Uuid; #[cfg(feature = "enterprise")] @@ -33,9 +35,9 @@ use windmill_common::{ error::{self, to_anyhow, Error}, flows::{FlowModule, FlowModuleValue, FlowValue}, get_latest_deployed_hash_for_path, - jobs::{JobKind, QueuedJob}, + jobs::{JobKind, JobPayload, QueuedJob}, scripts::{get_full_hub_script_by_path, ScriptHash, ScriptLang}, - users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL}, + users::{SUPERADMIN_NOTIFICATION_EMAIL, SUPERADMIN_SECRET_EMAIL, SUPERADMIN_SYNC_EMAIL}, utils::{rd_string, StripPath}, worker::{ to_raw_value, to_raw_value_owned, update_ping, CLOUD_HOSTED, WORKER_CONFIG, WORKER_GROUP, @@ -149,7 +151,8 @@ pub async fn create_token_for_owner( .await? .unwrap_or(false) || email == SUPERADMIN_SECRET_EMAIL - || email == SUPERADMIN_NOTIFICATION_EMAIL; + || email == SUPERADMIN_NOTIFICATION_EMAIL + || owner == SUPERADMIN_SYNC_EMAIL; sqlx::query_scalar!( "INSERT INTO token @@ -2515,6 +2518,7 @@ async fn handle_queued_job( worker_dir, base_internal_url, &client.get_token().await, + rsmq.clone(), ) .await } @@ -2529,6 +2533,7 @@ async fn handle_queued_job( worker_dir, base_internal_url, &client.get_token().await, + rsmq.clone(), ) .await .map(|()| serde_json::from_str("{}").unwrap()), @@ -2543,6 +2548,7 @@ async fn handle_queued_job( worker_dir, base_internal_url, &client.get_token().await, + rsmq.clone(), ) .await .map(|()| serde_json::from_str("{}").unwrap()), @@ -3044,7 +3050,7 @@ mount {{ } #[tracing::instrument(level = "trace", skip_all)] -async fn handle_dependency_job( +async fn handle_dependency_job( job: &QueuedJob, logs: &mut String, mem_peak: &mut i32, @@ -3055,7 +3061,17 @@ async fn handle_dependency_job( worker_dir: &str, base_internal_url: &str, token: &str, + rsmq: Option, ) -> error::Result> { + let raw_code = sqlx::query_scalar!( + "SELECT content FROM script WHERE hash = $1 AND workspace_id = $2", + &job.script_hash.unwrap_or(ScriptHash(0)).0, + &job.workspace_id + ) + .fetch_optional(db) + .await? + .unwrap_or_else(|| "No script found at this hash".to_string()); + let script_path = job.script_path(); let content = capture_dependency_job( &job.id, job.language.as_ref().map(|v| Ok(v)).unwrap_or_else(|| { @@ -3063,10 +3079,7 @@ async fn handle_dependency_job( "Job Language required for dependency jobs".to_owned(), )) })?, - job.raw_code - .as_ref() - .map(|a| a.as_str()) - .unwrap_or_else(|| "no raw code"), + &raw_code, logs, mem_peak, canceled_by, @@ -3077,19 +3090,83 @@ async fn handle_dependency_job( worker_dir, base_internal_url, token, - job.script_path(), + script_path, ) .await; match content { Ok(content) => { + let hash = job.script_hash.unwrap_or(ScriptHash(0)); + let w_id = &job.workspace_id; sqlx::query!( "UPDATE script SET lock = $1 WHERE hash = $2 AND workspace_id = $3", &content, - &job.script_hash.unwrap_or(ScriptHash(0)).0, - &job.workspace_id + &hash.0, + w_id ) .execute(db) .await?; + + let deployment_message = job.raw_code.as_ref().map(|x| x.clone()); + + if let Err(e) = handle_deployment_metadata( + &job.email, + &job.created_by, + &db, + &w_id, + DeployedObject::Script { hash, path: script_path.to_string() }, + deployment_message.clone(), + rsmq.clone(), + ) + .await + { + tracing::error!(%e, "error handling deployment metadata"); + } + + if &job.language == &Some(ScriptLang::Python3) { + if let Ok(relative_imports) = parse_relative_imports(&raw_code, script_path) { + logs.push_str("\n--- RELATIVE IMPORTS ---\n\n"); + logs.push_str(&relative_imports.join("\n")); + if !relative_imports.is_empty() { + let mut tx = db.begin().await?; + sqlx::query!( + "DELETE FROM dependency_map + WHERE importer_path = $1 AND importer_kind = 'script' + AND workspace_id = $2", + script_path, + w_id + ) + .execute(&mut *tx) + .await?; + for import in relative_imports { + sqlx::query!( + "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path) + VALUES ($1, $2, 'script', $3)", + w_id, + script_path, + import + ) + .execute(&mut *tx) + .await?; + logs.push_str(&format!("{}\n", import)); + } + tx.commit().await?; + } + if let Err(e) = trigger_python_dependents_to_recompute_dependencies( + w_id, + script_path, + deployment_message, + &job.email, + &job.created_by, + &job.permissioned_as, + db, + rsmq, + ) + .await + { + tracing::error!(%e, "error triggering python dependents to recompute dependencies"); + } + } + } Ok(to_raw_value_owned( json!({ "success": "Successful lock file generation", "lock": content }), )) @@ -3108,7 +3185,80 @@ async fn handle_dependency_job( } } -async fn handle_flow_dependency_job( +async fn trigger_python_dependents_to_recompute_dependencies< + R: rsmq_async::RsmqConnection + Send + Sync + Clone, +>( + w_id: &str, + script_path: &str, + deployment_message: Option, + email: &str, + created_by: &str, + permissioned_as: &str, + db: &sqlx::Pool, + rsmq: Option, +) -> error::Result<()> { + let script_importers = sqlx::query_scalar!( + "SELECT importer_path FROM dependency_map + WHERE imported_path = $1 AND importer_kind = 'script' + AND workspace_id = $2", + script_path, + w_id + ) + .fetch_all(db) + .await?; + for s in script_importers.iter() { + let tx: PushIsolationLevel<'_, R> = + PushIsolationLevel::IsolatedRoot(db.clone(), rsmq.clone()); + let r = get_latest_deployed_hash_for_path(db, w_id, s.as_str()).await; + if let Ok(r) = r { + let (job_uuid, new_tx) = windmill_queue::push( + db, + tx, + &w_id, + JobPayload::Dependencies { + path: s.clone(), + hash: r.0, + language: r.5, + dedicated_worker: r.6, + deployment_message: deployment_message.clone(), + }, + PushArgs::empty(), + &created_by, + email, + permissioned_as.to_string(), + None, + None, + None, + None, + None, + false, + false, + None, + true, + None, + None, + None, + None, + ) + .await?; + tracing::info!( + "pushed dependency job due to common python path: {job_uuid} for path {path} with hash {hash}", + path = s, + hash = r.0 + ); + new_tx.commit().await?; + } else { + tracing::error!( + "error getting latest deployed hash for path {path}: {err}", + path = s, + err = r.unwrap_err() + ); + } + } + Ok(()) +} + +async fn handle_flow_dependency_job( job: &QueuedJob, logs: &mut String, mem_peak: &mut i32, @@ -3119,6 +3269,7 @@ async fn handle_flow_dependency_job( worker_dir: &str, base_internal_url: &str, token: &str, + rsmq: Option, ) -> error::Result<()> { let job_path = job.script_path.clone().ok_or_else(|| { error::Error::InternalErr( @@ -3170,6 +3321,21 @@ async fn handle_flow_dependency_job( ) .execute(db) .await?; + + if let Err(e) = handle_deployment_metadata( + &job.email, + &job.created_by, + &db, + &job.workspace_id, + DeployedObject::Flow { path: job_path }, + job.raw_code.as_ref().map(|x| x.clone()), + rsmq.clone(), + ) + .await + { + tracing::error!(%e, "error handling deployment metadata"); + } + Ok(()) } @@ -3295,22 +3461,11 @@ async fn lock_modules( new_flow_modules.push(e); continue; }; - // sync with windmill-api/scripts - let dependencies = match language { - ScriptLang::Python3 => windmill_parser_py_imports::parse_python_imports( - &content, - &job.workspace_id, - &path.clone().unwrap_or_else(|| job_path.to_string()), - &db, - ) - .await? - .join("\n"), - _ => content.clone(), - }; + let new_lock = capture_dependency_job( &job.id, &language, - &dependencies, + &content, logs, mem_peak, canceled_by, @@ -3321,7 +3476,7 @@ async fn lock_modules( worker_dir, base_internal_url, token, - job.script_path(), + &path.clone().unwrap_or_else(|| job_path.to_string()), ) .await; match new_lock { @@ -3396,24 +3551,11 @@ async fn lock_modules_app( .as_str() .unwrap_or_default() .to_string(); - let dependencies = match language { - ScriptLang::Python3 => { - windmill_parser_py_imports::parse_python_imports( - &content, - &job.workspace_id, - job_path, - &db, - ) - .await? - .join("\n") - } - _ => content.clone(), - }; logs.push_str("Found lockable inline script. Generating lock...\n"); let new_lock = capture_dependency_job( &job.id, &language, - &dependencies, + &content, logs, mem_peak, canceled_by, @@ -3498,7 +3640,7 @@ async fn lock_modules_app( } } -async fn handle_app_dependency_job( +async fn handle_app_dependency_job( job: &QueuedJob, logs: &mut String, mem_peak: &mut i32, @@ -3509,6 +3651,7 @@ async fn handle_app_dependency_job( worker_dir: &str, base_internal_url: &str, token: &str, + rsmq: Option, ) -> error::Result<()> { let job_path = job.script_path.clone().ok_or_else(|| { error::Error::InternalErr( @@ -3558,6 +3701,41 @@ async fn handle_app_dependency_job( sqlx::query!("UPDATE app_version SET value = $1 WHERE id = $2", value, id,) .execute(db) .await?; + + if let Err(e) = handle_deployment_metadata( + &job.email, + &job.created_by, + &db, + &job.workspace_id, + DeployedObject::App { path: job_path, version: id }, + job.raw_code.as_ref().map(|x| x.clone()), + rsmq.clone(), + ) + .await + { + tracing::error!(%e, "error handling deployment metadata"); + } + + // tx = PushIsolationLevel::Transaction(new_tx); + // tx = handle_deployment_metadata( + // tx, + // &authed, + // &db, + // &w_id, + // DeployedObject::App { path: app.path.clone(), version: v_id }, + // app.deployment_message, + // ) + // .await?; + + // match tx { + // PushIsolationLevel::Transaction(tx) => tx.commit().await?, + // _ => { + // return Err(Error::InternalErr( + // "Expected a transaction here".to_string(), + // )); + // } + // } + Ok(()) } else { Ok(()) @@ -3582,10 +3760,18 @@ async fn capture_dependency_job( ) -> error::Result { match job_language { ScriptLang::Python3 => { + let reqs = windmill_parser_py_imports::parse_python_imports( + job_raw_code, + &w_id, + script_path, + &db, + ) + .await? + .join("\n"); create_dependencies_dir(job_dir).await; let req: std::result::Result = pip_compile( job_id, - job_raw_code, + &reqs, logs, mem_peak, canceled_by, diff --git a/frontend/src/lib/components/ResultJobLoader.svelte b/frontend/src/lib/components/ResultJobLoader.svelte index a8103e8792..cfbace2013 100644 --- a/frontend/src/lib/components/ResultJobLoader.svelte +++ b/frontend/src/lib/components/ResultJobLoader.svelte @@ -177,7 +177,7 @@ if (currentId === id || allowConcurentRequests) { job = { ...maybe_job, id } await tick() - if ('error' in (job.result ?? {})) { + if ('error' in (job?.result ?? {})) { callbacks?.error() dispatch('doneError', { id, diff --git a/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte b/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte index ca5b62e986..626d312206 100644 --- a/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte +++ b/frontend/src/lib/components/apps/components/helpers/RunnableComponent.svelte @@ -373,7 +373,6 @@ ): Promise { try { if (cancellableRun && !dynamicArgsOverride) { - console.log('runComponent cancellable Run') await cancellableRun() } else { console.log('Run component') @@ -530,7 +529,6 @@ onMount(() => { cancellableRun = (inlineScript?: InlineScript, setRunnableJobEditorPanel?: boolean) => { - console.log('cancellableRun', inlineScript) let rejectCb: (err: Error) => void let p: Partial> = new Promise((resolve, reject) => { rejectCb = reject diff --git a/frontend/src/lib/components/runs/JobPreview.svelte b/frontend/src/lib/components/runs/JobPreview.svelte index b9195758eb..d6ef501060 100644 --- a/frontend/src/lib/components/runs/JobPreview.svelte +++ b/frontend/src/lib/components/runs/JobPreview.svelte @@ -83,9 +83,7 @@ Result Logs - {#if job?.job_kind == 'dependencies'} - Code - {:else if job?.job_kind == 'preview'} + {#if job?.job_kind == 'preview'} Code {/if} diff --git a/frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte b/frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte index 8343bb472a..bd34eabfbc 100644 --- a/frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/apps/get/[...path]/+page.svelte @@ -59,8 +59,11 @@ /> {#if can_write && !hideEditBtn}
- Edit
{/if} diff --git a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte index 3ddb65e633..bc19d5595c 100644 --- a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte @@ -512,9 +512,7 @@ Result Logs Metrics - {#if job?.job_kind == 'dependencies'} - Code - {:else if job?.job_kind == 'preview'} + {#if job?.job_kind == 'preview'} Code {/if} diff --git a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte index 6f4762a5e5..6c2c24319f 100644 --- a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte @@ -389,8 +389,11 @@
Users
 + +
Git Sync
+ -
Dev/Staging/Prod
+
Deployment UI
 {#if WORKSPACE_SHOW_SLACK_CMD} @@ -416,9 +419,6 @@
S3 Storage
- -
Git sync
-
Delete Workspace
@@ -705,7 +705,7 @@ /> {:else if tab == 'windmill_lfs'} - + {#if !$enterpriseLicense} Windmill S3 bucket browser will not work for buckets containing more than 20 files. @@ -755,6 +755,7 @@ Automatically saving scripts to a Git repository on each deploy is a Windmill EE feature. +
{/if}
- {#key s3ResourceInitialPath} - { - editWindmillGitSyncSettings(ev.detail, gitSyncSettings.use_individual_branch) - }} - /> - - {/key} + {#if gitSyncSettings} + {#key gitSyncSettings} + { + editWindmillGitSyncSettings(ev.detail, gitSyncSettings?.use_individual_branch) + }} + /> + + {/key} + {/if}
{#if gitSyncTestJob !== undefined} @@ -801,18 +804,20 @@
- { - editWindmillGitSyncSettings(gitSyncSettings.git_repo_resource_path, ev.detail) - }} - options={{ - right: 'Create one branch per deployed script/flow/app', - rightTooltip: - "If set, Windmill will create a unique branch per script/flow/app being pushed, prefixed with 'wm_deploy/'." - }} - /> + {#if gitSyncSettings} + { + editWindmillGitSyncSettings(gitSyncSettings.git_repo_resource_path, ev.detail) + }} + options={{ + right: 'Create one branch per deployed script/flow/app', + rightTooltip: + "If set, Windmill will create a unique branch per script/flow/app being pushed, prefixed with 'wm_deploy/'." + }} + /> + {/if}
@@ -841,12 +846,12 @@ 
> wmill workspace add WORKSPACE_NAME WORKSPACE_ID WINDMILL_URL
-> echo 'u/' > .wmillignore
-> wmill sync pull --raw --skip-variables --skip-secrets --skip-resources
-> git add -A
-> git commit -m 'Initial commit'
-> git pushwmill workspace add  {$workspaceStore} {$workspaceStore} {`${$page.url.protocol}//${$page.url.hostname}/`}
+echo 'u/' > .wmillignore
+wmill sync pull --raw --skip-variables --skip-secrets --skip-resources
+git add -A
+git commit -m 'Initial commit'
+git push