kalugin66/windmill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: track workspace runnables used in flows (#5369)

Browse Source 
* feat: track workspace runnables used in flows

* track script hash

* weird

* do it with lock

* Revert "feat: add support for | None and Optional in python (#5361)"

This reverts commit 69a2d85f25.

* Revert "Revert "feat: add support for | None and Optional in python (#5361)""

This reverts commit bb8f709894.

* update openapi

* delete old in lock_modules + don't track hub scripts
This commit is contained in:
HugoCasa
2025-02-27 10:01:12 +01:00
committed by GitHub
parent afb26f4c8a
commit fd3459568f
16 changed files with 241 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
backend/.sqlx/query-2bb2cf6accb18d3e37a63388cca52a6591e7593b1a7c3d7a6848587679a48187.json generated Normal file
View File

@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE flow_workspace_runnables SET workspace_id = $1 WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "2bb2cf6accb18d3e37a63388cca52a6591e7593b1a7c3d7a6848587679a48187"
}

16
backend/.sqlx/query-35795d27c4ca69d2f145b4dba08a6ed16c25aea4584103c1b9a3651eb31bfe53.json generated Normal file
View File

@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow_workspace_runnables (flow_path, runnable_path, runnable_is_flow, workspace_id) VALUES ($1, $2, TRUE, $3) ON CONFLICT DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar"
]
},
"nullable": []
},
"hash": "35795d27c4ca69d2f145b4dba08a6ed16c25aea4584103c1b9a3651eb31bfe53"
}

24
backend/.sqlx/query-9aeee333b1dbe58ba819ba3b2713242b54d77b46b5f785ae66b8104f89f43219.json generated Normal file
View File

@@ -0,0 +1,24 @@
{
"db_name": "PostgreSQL",
"query": "SELECT f.path\n FROM flow_workspace_runnables fwr \n JOIN flow f \n ON fwr.flow_path = f.path AND fwr.workspace_id = f.workspace_id\n WHERE fwr.runnable_path = $1 AND fwr.runnable_is_flow = $2 AND fwr.workspace_id = $3",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "path",
"type_info": "Varchar"
}
],
"parameters": {
"Left": [
"Text",
"Bool",
"Text"
]
},
"nullable": [
false
]
},
"hash": "9aeee333b1dbe58ba819ba3b2713242b54d77b46b5f785ae66b8104f89f43219"
}

15
backend/.sqlx/query-acbea8740b28c26942c50edcf5618cd141e68cf83a7dcae7d3c1b8a7ba94425b.json generated Normal file
View File

@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "DELETE FROM flow_workspace_runnables WHERE flow_path = $1 AND workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": []
},
"hash": "acbea8740b28c26942c50edcf5618cd141e68cf83a7dcae7d3c1b8a7ba94425b"
}

16
backend/.sqlx/query-b656927cd70b6667f3c72186ec04f0bf040da3af9e2eac3229264ec95b4755d8.json generated Normal file
View File

@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE flow_workspace_runnables SET flow_path = REGEXP_REPLACE(flow_path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1') WHERE flow_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "b656927cd70b6667f3c72186ec04f0bf040da3af9e2eac3229264ec95b4755d8"
}

17
backend/.sqlx/query-c35f44f91b08fa57e29a2b4a685706f62e700695810f23108e975dfcd1fee7a3.json generated Normal file
View File

@@ -0,0 +1,17 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO flow_workspace_runnables (flow_path, runnable_path, script_hash, runnable_is_flow, workspace_id) VALUES ($1, $2, $3, FALSE, $4) ON CONFLICT DO NOTHING",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Int8",
"Varchar"
]
},
"nullable": []
},
"hash": "c35f44f91b08fa57e29a2b4a685706f62e700695810f23108e975dfcd1fee7a3"
}

16
backend/.sqlx/query-dc58e5b4715601a93b3c01a2564a4420f232867a23cacb9a62b386f129a86a4b.json generated Normal file
View File

@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE flow_workspace_runnables SET runnable_path = REGEXP_REPLACE(runnable_path,'u/' || $2 || '/(.*)','u/' || $1 || '/\\1') WHERE runnable_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "dc58e5b4715601a93b3c01a2564a4420f232867a23cacb9a62b386f129a86a4b"
}

1
backend/migrations/20250224131521_flow_workspace_runnables.down.sql Normal file
View File

@@ -0,0 +1 @@
DROP TABLE flow_workspace_runnables;

12
backend/migrations/20250224131521_flow_workspace_runnables.up.sql Normal file
View File

@@ -0,0 +1,12 @@
CREATE TABLE flow_workspace_runnables (
flow_path VARCHAR(255) NOT NULL,
runnable_path VARCHAR(255) NOT NULL,
script_hash BIGINT NULL,
runnable_is_flow BOOLEAN NOT NULL,
workspace_id VARCHAR(50) NOT NULL,
FOREIGN KEY (workspace_id, flow_path) REFERENCES flow (workspace_id, path) ON DELETE CASCADE
);
CREATE UNIQUE INDEX flow_workspace_without_hash_unique_idx ON flow_workspace_runnables (flow_path, runnable_path, runnable_is_flow, workspace_id) WHERE script_hash IS NULL;
CREATE UNIQUE INDEX flow_workspace_with_hash_unique_idx ON flow_workspace_runnables (flow_path, runnable_path, script_hash, runnable_is_flow, workspace_id) WHERE script_hash IS NOT NULL;
CREATE INDEX flow_workspace_runnable_path_is_flow_idx ON flow_workspace_runnables (runnable_path, runnable_is_flow, workspace_id);

20
backend/windmill-api/openapi.yaml
View File

@@ -5292,6 +5292,26 @@ paths:
schema:
$ref: "#/components/schemas/FlowVersion"
/w/{workspace}/flows/list_paths_from_workspace_runnable/{runnable_kind}/{path}:
get:
summary: list flow paths from workspace runnable
operationId: listFlowPathsFromWorkspaceRunnable
tags:
- flow
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/RunnableKind"
- $ref: "#/components/parameters/ScriptPath"
responses:
"200":
description: list of flow paths
content:
application/json:
schema:
type: array
items:
type: string
/w/{workspace}/flows/get/v/{version}/p/{path}:
get:
summary: get flow version

13
backend/windmill-api/src/capture.rs
View File

@@ -52,6 +52,7 @@ use crate::{
args::WebhookArgs,
db::{ApiAuthed, DB},
users::fetch_api_authed,
utils::RunnableKind,
};
const KEEP_LAST: i64 = 20;
@@ -139,7 +140,7 @@ pub struct KafkaTriggerConfig {
pub struct SqsTriggerConfig {
pub queue_url: String,
pub aws_resource_path: String,
pub message_attributes: Option<Vec<String>>
pub message_attributes: Option<Vec<String>>,
}
#[cfg(all(feature = "enterprise", feature = "nats"))]
@@ -300,8 +301,7 @@ async fn set_config(
#[cfg(feature = "postgres_trigger")]
let nc = if let TriggerKind::Postgres = nc.trigger_kind {
set_postgres_trigger_config(&w_id, authed.clone(), &db, user_db.clone(), nc).await?
}
else {
} else {
nc
};
@@ -362,13 +362,6 @@ struct Capture {
trigger_extra: Option<SqlxJson<Box<serde_json::value::RawValue>>>,
}
#[derive(Deserialize)]
#[serde(rename_all = "lowercase")]
enum RunnableKind {
Script,
Flow,
}
#[derive(Deserialize)]
struct ListCapturesQuery {
trigger_kind: Option<TriggerKind>,

28
backend/windmill-api/src/flows.rs
View File

@@ -12,7 +12,7 @@ use crate::db::ApiAuthed;
use crate::triggers::{
get_triggers_count_internal, list_tokens_internal, TriggersCount, TruncatedTokenWithEmail,
};
use crate::utils::WithStarredInfoQuery;
use crate::utils::{RunnableKind, WithStarredInfoQuery};
use crate::{
db::DB,
schedule::clear_schedule,
@@ -65,6 +65,10 @@ pub fn workspaced_service() -> Router {
.route("/list_paths", get(list_paths))
.route("/history/p/*path", get(get_flow_history))
.route("/get_latest_version/*path", get(get_latest_version))
.route(
"/list_paths_from_workspace_runnable/:runnable_kind/*path",
get(list_paths_from_workspace_runnable),
)
.route(
"/history_update/v/:version/p/*path",
post(update_flow_history),
@@ -324,6 +328,28 @@ async fn check_path_conflict<'c>(
return Ok(());
}
async fn list_paths_from_workspace_runnable(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Path((w_id, runnable_kind, path)): Path<(String, RunnableKind, StripPath)>,
) -> JsonResult<Vec<String>> {
let mut tx = user_db.begin(&authed).await?;
let runnables = sqlx::query_scalar!(
r#"SELECT f.path
FROM flow_workspace_runnables fwr
JOIN flow f
ON fwr.flow_path = f.path AND fwr.workspace_id = f.workspace_id
WHERE fwr.runnable_path = $1 AND fwr.runnable_is_flow = $2 AND fwr.workspace_id = $3"#,
path.to_path(),
matches!(runnable_kind, RunnableKind::Flow),
w_id
)
.fetch_all(&mut *tx)
.await?;
tx.commit().await?;
Ok(Json(runnables))
}
async fn create_flow(
authed: ApiAuthed,
Extension(db): Extension<DB>,

16
backend/windmill-api/src/users.rs
View File

@@ -2478,6 +2478,22 @@ async fn update_username_in_workpsace<'c>(
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"UPDATE flow_workspace_runnables SET flow_path = REGEXP_REPLACE(flow_path,'u/' || $2 || '/(.*)','u/' || $1 || '/\1') WHERE flow_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
new_username,
old_username,
w_id
).execute(&mut **tx)
.await?;
sqlx::query!(
r#"UPDATE flow_workspace_runnables SET runnable_path = REGEXP_REPLACE(runnable_path,'u/' || $2 || '/(.*)','u/' || $1 || '/\1') WHERE runnable_path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
new_username,
old_username,
w_id
).execute(&mut **tx)
.await?;
sqlx::query!(
r#"UPDATE flow_node SET path = REGEXP_REPLACE(path,'u/' || $2 || '/(.*)','u/' || $1 || '/\1') WHERE path LIKE ('u/' || $2 || '/%') AND workspace_id = $3"#,
new_username,

7
backend/windmill-api/src/utils.rs
View File

@@ -29,6 +29,13 @@ pub struct WithStarredInfoQuery {
pub with_starred_info: Option<bool>,
}
#[derive(Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum RunnableKind {
Script,
Flow,
}
pub async fn require_super_admin(db: &DB, email: &str) -> error::Result<()> {
let is_admin = is_super_admin_email(db, email).await?;

8
backend/windmill-api/src/workspaces_extra.rs
View File

@@ -189,6 +189,14 @@ pub(crate) async fn change_workspace_id(
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE flow_workspace_runnables SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
sqlx::query!(
"UPDATE flow_node SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,

28
backend/windmill-worker/src/worker_lockfiles.rs
View File

@@ -620,6 +620,13 @@ pub async fn handle_flow_dependency_job(
tx = clear_dependency_parent_path(&parent_path, &job_path, &job.workspace_id, "flow", tx)
.await?;
sqlx::query!(
"DELETE FROM flow_workspace_runnables WHERE flow_path = $1 AND workspace_id = $2",
job_path,
job.workspace_id
)
.execute(&mut *tx)
.await?;
let modified_ids;
let errors;
(flow.modules, tx, modified_ids, errors) = lock_modules(
@@ -981,6 +988,27 @@ async fn lock_modules<'c>(
}
.into();
}
FlowModuleValue::Script { path, hash, .. } if !path.starts_with("hub/") => {
sqlx::query!(
"INSERT INTO flow_workspace_runnables (flow_path, runnable_path, script_hash, runnable_is_flow, workspace_id) VALUES ($1, $2, $3, FALSE, $4) ON CONFLICT DO NOTHING",
job_path,
path,
hash.map(|h| h.0),
job.workspace_id
)
.execute(&mut *tx)
.await?;
}
FlowModuleValue::Flow { path, .. } => {
sqlx::query!(
"INSERT INTO flow_workspace_runnables (flow_path, runnable_path, runnable_is_flow, workspace_id) VALUES ($1, $2, TRUE, $3) ON CONFLICT DO NOTHING",
job_path,
path,
job.workspace_id
)
.execute(&mut *tx)
.await?;
}
_ => (),
};
modified_ids.extend(nmodified_ids);