starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-18 16:02:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: http routes streaming (#6834)

Browse Source 
* feat: http routes streaming

* improve webhook config section for SSE

* feat: add request type config to http routes

* fix check

* sqlx

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
hugocasa
2025-10-17 16:05:17 +02:00
committed by GitHub
parent d2ae0533f4
commit 8cd0006498
21 changed files with 756 additions and 415 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-4410b2d6e52556569cda9d945f791756a80ff2835d74bfab85f53141b6e32351.json → backend/.sqlx/query-1301f873a829db137573b8b39449f6160f2adf44f864f26a99b8eab5818fbd50.json
Generated
+15 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT \n path, \n script_path, \n is_flow, \n route_path, \n authentication_resource_path,\n workspace_id, \n is_async, \n authentication_method AS \"authentication_method: _\", \n edited_by, \n email, \n static_asset_config AS \"static_asset_config: _\",\n wrap_body,\n raw_string,\n workspaced_route,\n is_static_website,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM \n http_trigger \n WHERE \n http_method = $1\n ",
"query": "\n SELECT\n path,\n script_path,\n is_flow,\n route_path,\n authentication_resource_path,\n workspace_id,\n request_type AS \"request_type: _\",\n authentication_method AS \"authentication_method: _\",\n edited_by,\n email,\n static_asset_config AS \"static_asset_config: _\",\n wrap_body,\n raw_string,\n workspaced_route,\n is_static_website,\n error_handler_path,\n error_handler_args as \"error_handler_args: _\",\n retry as \"retry: _\"\n FROM\n http_trigger\n WHERE\n http_method = $1\n ",
"describe": {
"columns": [
{
@@ -35,8 +35,19 @@
},
{
"ordinal": 6,
"name": "is_async",
"type_info": "Bool"
"name": "request_type: _",
"type_info": {
"Custom": {
"name": "request_type",
"kind": {
"Enum": [
"sync",
"async",
"sync_sse"
]
}
}
}
},
{
"ordinal": 7,
@@ -147,5 +158,5 @@
true
]
},
"hash": "4410b2d6e52556569cda9d945f791756a80ff2835d74bfab85f53141b6e32351"
"hash": "1301f873a829db137573b8b39449f6160f2adf44f864f26a99b8eab5818fbd50"
}
backend/.sqlx/query-1f6b773ce34fe51d03d6f9a2345481629c62453eebbb08f82dd2da23389bc117.json
Generated
+73
View File
@@ -0,0 +1,73 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO http_trigger (\n workspace_id,\n path,\n route_path,\n route_path_key,\n workspaced_route,\n authentication_resource_path,\n wrap_body,\n raw_string,\n script_path,\n summary,\n description,\n is_flow,\n request_type,\n authentication_method,\n http_method,\n static_asset_config,\n edited_by,\n email,\n edited_at,\n is_static_website,\n error_handler_path,\n error_handler_args,\n retry\n )\n VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, now(), $19, $20, $21, $22\n )\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Bool",
"Bool",
"Varchar",
"Varchar",
"Text",
"Bool",
{
"Custom": {
"name": "request_type",
"kind": {
"Enum": [
"sync",
"async",
"sync_sse"
]
}
}
},
{
"Custom": {
"name": "authentication_method",
"kind": {
"Enum": [
"none",
"windmill",
"api_key",
"basic_http",
"custom_script",
"signature"
]
}
}
},
{
"Custom": {
"name": "http_method",
"kind": {
"Enum": [
"get",
"post",
"put",
"delete",
"patch"
]
}
}
},
"Jsonb",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "1f6b773ce34fe51d03d6f9a2345481629c62453eebbb08f82dd2da23389bc117"
}
backend/.sqlx/query-465144ea7e2930203618d9814a3e20c77b4363cf9e7c655d395f3fe40c247f61.json
Generated
+74
View File
@@ -0,0 +1,74 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE\n http_trigger\n SET\n route_path = $1,\n route_path_key = $2,\n workspaced_route = $3,\n wrap_body = $4,\n raw_string = $5,\n authentication_resource_path = $6,\n script_path = $7,\n path = $8,\n is_flow = $9,\n http_method = $10,\n static_asset_config = $11,\n edited_by = $12,\n email = $13,\n request_type = $14,\n authentication_method = $15,\n summary = $16,\n description = $17,\n edited_at = now(),\n is_static_website = $18,\n error_handler_path = $19,\n error_handler_args = $20,\n retry = $21\n WHERE\n workspace_id = $22 AND\n path = $23\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Bool",
"Bool",
"Bool",
"Varchar",
"Varchar",
"Varchar",
"Bool",
{
"Custom": {
"name": "http_method",
"kind": {
"Enum": [
"get",
"post",
"put",
"delete",
"patch"
]
}
}
},
"Jsonb",
"Varchar",
"Varchar",
{
"Custom": {
"name": "request_type",
"kind": {
"Enum": [
"sync",
"async",
"sync_sse"
]
}
}
},
{
"Custom": {
"name": "authentication_method",
"kind": {
"Enum": [
"none",
"windmill",
"api_key",
"basic_http",
"custom_script",
"signature"
]
}
}
},
"Varchar",
"Text",
"Bool",
"Varchar",
"Jsonb",
"Jsonb",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "465144ea7e2930203618d9814a3e20c77b4363cf9e7c655d395f3fe40c247f61"
}
backend/.sqlx/query-4c64bd0e364f536597db83161b5a27ff58b5ec7148bc94807423593bcfa27b90.json
Generated
+71
View File
@@ -0,0 +1,71 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE\n http_trigger\n SET\n wrap_body = $1,\n raw_string = $2,\n authentication_resource_path = $3,\n script_path = $4,\n path = $5,\n is_flow = $6,\n http_method = $7,\n static_asset_config = $8,\n edited_by = $9,\n email = $10,\n request_type = $11,\n authentication_method = $12,\n summary = $13,\n description = $14,\n edited_at = now(),\n is_static_website = $15,\n error_handler_path = $16,\n error_handler_args = $17,\n retry = $18\n WHERE\n workspace_id = $19 AND\n path = $20\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Bool",
"Bool",
"Varchar",
"Varchar",
"Varchar",
"Bool",
{
"Custom": {
"name": "http_method",
"kind": {
"Enum": [
"get",
"post",
"put",
"delete",
"patch"
]
}
}
},
"Jsonb",
"Varchar",
"Varchar",
{
"Custom": {
"name": "request_type",
"kind": {
"Enum": [
"sync",
"async",
"sync_sse"
]
}
}
},
{
"Custom": {
"name": "authentication_method",
"kind": {
"Enum": [
"none",
"windmill",
"api_key",
"basic_http",
"custom_script",
"signature"
]
}
}
},
"Varchar",
"Text",
"Bool",
"Varchar",
"Jsonb",
"Jsonb",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "4c64bd0e364f536597db83161b5a27ff58b5ec7148bc94807423593bcfa27b90"
}
backend/.sqlx/query-5cc0d9e3dcd9c20e6e6ec1acf38c8f97b5ece60bfe24fc3783a83de47e3aa583.json
Generated
-62
View File
@@ -1,62 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO http_trigger (\n workspace_id, \n path, \n route_path, \n route_path_key,\n workspaced_route,\n authentication_resource_path,\n wrap_body,\n raw_string,\n script_path, \n summary,\n description,\n is_flow, \n is_async, \n authentication_method, \n http_method, \n static_asset_config, \n edited_by, \n email, \n edited_at, \n is_static_website,\n error_handler_path,\n error_handler_args,\n retry\n ) \n VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, now(), $19, $20, $21, $22\n )\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Bool",
"Bool",
"Varchar",
"Varchar",
"Text",
"Bool",
"Bool",
{
"Custom": {
"name": "authentication_method",
"kind": {
"Enum": [
"none",
"windmill",
"api_key",
"basic_http",
"custom_script",
"signature"
]
}
}
},
{
"Custom": {
"name": "http_method",
"kind": {
"Enum": [
"get",
"post",
"put",
"delete",
"patch"
]
}
}
},
"Jsonb",
"Varchar",
"Varchar",
"Bool",
"Varchar",
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "5cc0d9e3dcd9c20e6e6ec1acf38c8f97b5ece60bfe24fc3783a83de47e3aa583"
}
backend/.sqlx/query-85d945cf5ade707291a161078ff96ddb29140dd72ce5115657418eec503b205d.json
Generated
-63
View File
@@ -1,63 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n http_trigger \n SET \n route_path = $1, \n route_path_key = $2,\n workspaced_route = $3,\n wrap_body = $4,\n raw_string = $5,\n authentication_resource_path = $6,\n script_path = $7, \n path = $8, \n is_flow = $9, \n http_method = $10, \n static_asset_config = $11, \n edited_by = $12, \n email = $13, \n is_async = $14, \n authentication_method = $15, \n summary = $16,\n description = $17,\n edited_at = now(), \n is_static_website = $18,\n error_handler_path = $19,\n error_handler_args = $20,\n retry = $21\n WHERE \n workspace_id = $22 AND \n path = $23\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Bool",
"Bool",
"Bool",
"Varchar",
"Varchar",
"Varchar",
"Bool",
{
"Custom": {
"name": "http_method",
"kind": {
"Enum": [
"get",
"post",
"put",
"delete",
"patch"
]
}
}
},
"Jsonb",
"Varchar",
"Varchar",
"Bool",
{
"Custom": {
"name": "authentication_method",
"kind": {
"Enum": [
"none",
"windmill",
"api_key",
"basic_http",
"custom_script",
"signature"
]
}
}
},
"Varchar",
"Text",
"Bool",
"Varchar",
"Jsonb",
"Jsonb",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "85d945cf5ade707291a161078ff96ddb29140dd72ce5115657418eec503b205d"
}
backend/.sqlx/query-714fb0f66ceb536aee8cb9ae0144757b999d25870fda37fe904e09dd5c742015.json → backend/.sqlx/query-9360d00990822f153ff09c7905ae3180f07d02f38ac12d07a5664d93f160e7ee.json
Generated
+15 -4
View File
@@ -1,6 +1,6 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n route_path,\n http_method AS \"http_method: _\",\n is_async,\n workspaced_route,\n summary,\n description,\n authentication_method AS \"authentication_method: _\",\n authentication_resource_path\n FROM\n http_trigger\n WHERE\n path ~ ANY($1) AND\n route_path ~ ANY($2) AND\n workspace_id = $3\n ",
"query": "\n SELECT\n route_path,\n http_method AS \"http_method: _\",\n request_type AS \"request_type: _\",\n workspaced_route,\n summary,\n description,\n authentication_method AS \"authentication_method: _\",\n authentication_resource_path\n FROM\n http_trigger\n WHERE\n path ~ ANY($1) AND\n route_path ~ ANY($2) AND\n workspace_id = $3\n ",
"describe": {
"columns": [
{
@@ -28,8 +28,19 @@
},
{
"ordinal": 2,
"name": "is_async",
"type_info": "Bool"
"name": "request_type: _",
"type_info": {
"Custom": {
"name": "request_type",
"kind": {
"Enum": [
"sync",
"async",
"sync_sse"
]
}
}
}
},
{
"ordinal": 3,
@@ -89,5 +100,5 @@
true
]
},
"hash": "714fb0f66ceb536aee8cb9ae0144757b999d25870fda37fe904e09dd5c742015"
"hash": "9360d00990822f153ff09c7905ae3180f07d02f38ac12d07a5664d93f160e7ee"
}
backend/.sqlx/query-a7a23229d6915d7fdeea8073d31be6e7d9f7a8581bdbb5914f2c5b49f37dbc36.json
Generated
-60
View File
@@ -1,60 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE \n http_trigger \n SET \n wrap_body = $1,\n raw_string = $2,\n authentication_resource_path = $3,\n script_path = $4, \n path = $5, \n is_flow = $6, \n http_method = $7, \n static_asset_config = $8, \n edited_by = $9, \n email = $10, \n is_async = $11, \n authentication_method = $12, \n summary = $13,\n description = $14,\n edited_at = now(), \n is_static_website = $15,\n error_handler_path = $16,\n error_handler_args = $17,\n retry = $18\n WHERE \n workspace_id = $19 AND \n path = $20\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Bool",
"Bool",
"Varchar",
"Varchar",
"Varchar",
"Bool",
{
"Custom": {
"name": "http_method",
"kind": {
"Enum": [
"get",
"post",
"put",
"delete",
"patch"
]
}
}
},
"Jsonb",
"Varchar",
"Varchar",
"Bool",
{
"Custom": {
"name": "authentication_method",
"kind": {
"Enum": [
"none",
"windmill",
"api_key",
"basic_http",
"custom_script",
"signature"
]
}
}
},
"Varchar",
"Text",
"Bool",
"Varchar",
"Jsonb",
"Jsonb",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "a7a23229d6915d7fdeea8073d31be6e7d9f7a8581bdbb5914f2c5b49f37dbc36"
}
backend/migrations/20251016163921_http_trigger_request_type.down.sql
+19
View File
@@ -0,0 +1,19 @@
-- Add down migration script here
-- Add back the is_async column
ALTER TABLE http_trigger ADD COLUMN is_async BOOLEAN;
-- Migrate request_type values back to is_async
-- 'async' -> TRUE
-- 'sync' or 'sync_sse' -> FALSE
UPDATE http_trigger SET is_async = CASE
WHEN request_type = 'async'::REQUEST_TYPE THEN TRUE
ELSE FALSE
END;
-- Make is_async NOT NULL with default
ALTER TABLE http_trigger ALTER COLUMN is_async SET NOT NULL;
ALTER TABLE http_trigger ALTER COLUMN is_async SET DEFAULT FALSE;
-- Drop the request_type column and type
ALTER TABLE http_trigger DROP COLUMN request_type;
DROP TYPE REQUEST_TYPE;
backend/migrations/20251016163921_http_trigger_request_type.up.sql
+23
View File
@@ -0,0 +1,23 @@
-- Add up migration script here
-- Create the request_type enum type
CREATE TYPE REQUEST_TYPE AS ENUM ('sync', 'async', 'sync_sse');
-- Add the new request_type column with a default value
ALTER TABLE http_trigger ADD COLUMN request_type REQUEST_TYPE;
-- Migrate existing is_async values to request_type
-- is_async = FALSE -> 'sync'
-- is_async = TRUE -> 'async'
UPDATE http_trigger SET request_type = CASE
WHEN is_async = TRUE THEN 'async'::REQUEST_TYPE
ELSE 'sync'::REQUEST_TYPE
END;
-- Make request_type NOT NULL now that all values are populated
ALTER TABLE http_trigger ALTER COLUMN request_type SET NOT NULL;
-- Set default for new rows
ALTER TABLE http_trigger ALTER COLUMN request_type SET DEFAULT 'sync'::REQUEST_TYPE;
-- Drop the old is_async column
ALTER TABLE http_trigger DROP COLUMN is_async;
backend/windmill-api/openapi.yaml
+16 -5
View File
@@ -16894,6 +16894,13 @@ components:
- delete
- patch
HttpRequestType:
type: string
enum:
- sync
- async
- sync_sse
HttpTrigger:
allOf:
- $ref: "#/components/schemas/TriggerExtraProperty"
@@ -16920,8 +16927,8 @@ components:
type: string
description:
type: string
is_async:
type: boolean
request_type:
$ref: "#/components/schemas/HttpRequestType"
authentication_method:
$ref: "#/components/schemas/AuthenticationMethod"
is_static_website:
@@ -16941,7 +16948,7 @@ components:
required:
- route_path
- is_async
- request_type
- authentication_method
- http_method
- is_static_website
@@ -16983,6 +16990,9 @@ components:
type: string
is_async:
type: boolean
description: Deprecated, use request_type instead
request_type:
$ref: "#/components/schemas/HttpRequestType"
authentication_method:
$ref: "#/components/schemas/AuthenticationMethod"
is_static_website:
@@ -17003,7 +17013,6 @@ components:
- script_path
- route_path
- is_flow
- is_async
- authentication_method
- http_method
- is_static_website
@@ -17042,6 +17051,9 @@ components:
$ref: "#/components/schemas/HttpMethod"
is_async:
type: boolean
description: Deprecated, use request_type instead
request_type:
$ref: "#/components/schemas/HttpRequestType"
authentication_method:
$ref: "#/components/schemas/AuthenticationMethod"
is_static_website:
@@ -17061,7 +17073,6 @@ components:
- script_path
- is_flow
- kind
- is_async
- authentication_method
- http_method
- is_static_website
backend/windmill-api/src/jobs.rs
+2 -3
View File
@@ -6896,7 +6896,7 @@ async fn get_job_update_sse(
#[derive(Serialize)]
#[serde(tag = "type", rename_all = "lowercase")]
enum JobUpdateSSEStream {
pub enum JobUpdateSSEStream {
Update(JobUpdate),
Error { error: String },
NotFound,
@@ -6909,8 +6909,7 @@ lazy_static::lazy_static! {
std::env::var("TIMEOUT_SSE_STREAM").unwrap_or("60".to_string()).parse::<u64>().unwrap_or(60);
}
fn start_job_update_sse_stream(
pub fn start_job_update_sse_stream(
opt_authed: Option<ApiAuthed>,
opt_tokened: OptTokened,
db: DB,
backend/windmill-api/src/lib.rs
+13 -2
View File
@@ -101,7 +101,8 @@ mod inkeep_oss;
mod inputs;
mod integration;
mod live_migrations;
pub mod openapi;
#[cfg(feature = "http_trigger")]
mod openapi;
#[cfg(all(feature = "private", feature = "parquet"))]
pub mod s3_proxy_ee;
mod s3_proxy_oss;
@@ -475,7 +476,17 @@ pub async fn run_server(
.nest("/variables", variables::workspaced_service())
.nest("/workspaces", workspaces::workspaced_service())
.nest("/oidc", oidc_oss::workspaced_service())
.nest("/openapi", openapi::openapi_service())
.nest("/openapi", {
#[cfg(feature = "http_trigger")]
{
openapi::openapi_service()
}
#[cfg(not(feature = "http_trigger"))]
{
Router::new()
}
})
.merge(triggers_service),
)
.nest("/workspaces", workspaces::global_service())
backend/windmill-api/src/openapi.rs
+49 -56
View File
@@ -9,6 +9,7 @@ use axum::{
};
use http::{header, HeaderValue, Method, StatusCode};
use indexmap::IndexMap;
use itertools::Itertools;
use serde::{Deserialize, Serialize};
use serde_json::{to_value, Map, Value};
use sqlx::PgConnection;
@@ -20,18 +21,13 @@ use windmill_common::{
DB,
};
use crate::db::ApiAuthed;
#[cfg(feature = "http_trigger")]
use {
crate::{
resources::try_get_resource_from_db_as,
triggers::http::{
http_trigger_args::HttpMethod, http_trigger_auth::ApiKeyAuthentication,
AuthenticationMethod,
},
use crate::{
db::ApiAuthed,
resources::try_get_resource_from_db_as,
triggers::http::{
http_trigger_args::HttpMethod, http_trigger_auth::ApiKeyAuthentication,
AuthenticationMethod, RequestType,
},
itertools::Itertools,
};
lazy_static::lazy_static! {
@@ -49,6 +45,7 @@ const BASIC_HTTP_AUTH_SCHEME: &'static str = "BasicHttp";
const DEFAULT_REQUEST_KEY: &'static str = "defaultRequest";
const DEFAULT_ASYNC_RESPONSE_KEY: &'static str = "AsyncResponse";
const DEFAULT_SYNC_RESPONSE_KEY: &'static str = "SyncResponse";
const DEFAULT_SYNC_SSE_RESPONSE_KEY: &'static str = "SyncSseResponse";
const DEFAULT_PAYLOAD_PARAM_KEY: &'static str = "PayloadParam";
pub fn openapi_service() -> Router {
@@ -166,7 +163,7 @@ pub enum Kind {
pub struct FuturePath {
route_path: String,
kind: Kind,
is_async: Option<bool>,
request_type: Option<RequestType>,
summary: Option<String>,
description: Option<String>,
security_scheme: Option<SecurityScheme>,
@@ -176,12 +173,12 @@ impl FuturePath {
pub fn new(
route_path: String,
kind: Kind,
is_async: Option<bool>,
request_type: Option<RequestType>,
summary: Option<String>,
description: Option<String>,
security_scheme: Option<SecurityScheme>,
) -> FuturePath {
FuturePath { route_path, kind, is_async, summary, description, security_scheme }
FuturePath { route_path, kind, request_type, summary, description, security_scheme }
}
}
@@ -241,6 +238,7 @@ fn from_route_path_to_openapi_path(
vec![
format!("/run/{}", &normalized_path),
format!("/run_wait_result/{}", &normalized_path),
format!("/run_and_stream/{}", &normalized_path),
]
};
@@ -282,19 +280,23 @@ fn generate_paths(
})
};
let generate_response = |is_async: bool| {
let responses = if is_async {
serde_json::json!({
let generate_response = |request_type: RequestType| {
let responses = match request_type {
RequestType::Async => serde_json::json!({
"200": {
"$ref": format!("#/components/responses/{DEFAULT_ASYNC_RESPONSE_KEY}")
}
})
} else {
serde_json::json!(serde_json::json!({
}),
RequestType::Sync => serde_json::json!({
"200": {
"$ref": format!("#/components/responses/{DEFAULT_SYNC_RESPONSE_KEY}")
}
}))
}),
RequestType::SyncSse => serde_json::json!({
"200": {
"$ref": format!("#/components/responses/{DEFAULT_SYNC_SSE_RESPONSE_KEY}")
}
}),
};
responses
@@ -347,12 +349,19 @@ fn generate_paths(
path_object
});
let is_async;
let request_type;
let (methods, is_webhook) = match &path.kind {
Kind::Webhook(_) => {
is_async = route_path.starts_with("/run/");
let methods = if is_async {
request_type = if route_path.starts_with("/run/") {
RequestType::Async
} else if route_path.starts_with("/run_and_stream/") {
RequestType::SyncSse
} else {
RequestType::Sync
};
let methods = if request_type == RequestType::Async {
vec![Method::POST]
} else {
vec![Method::GET, Method::POST]
@@ -369,7 +378,7 @@ fn generate_paths(
)
.into());
}
is_async = path.is_async.unwrap_or(true);
request_type = path.request_type.unwrap_or(RequestType::Sync);
(vec![method.to_owned()], false)
}
};
@@ -401,7 +410,7 @@ fn generate_paths(
);
}
method_map.insert("responses", generate_response(is_async));
method_map.insert("responses", generate_response(request_type));
path_object.insert(method.to_string().to_lowercase(), to_value(&method_map)?);
}
@@ -428,19 +437,6 @@ pub fn transform_to_minified_postgres_regex(glob: &str) -> String {
regex
}
#[derive(Debug, Default)]
pub struct ServerToSet {
pub http_route: bool,
pub webhook_flow: bool,
pub webhook_script: bool,
}
impl ServerToSet {
pub fn new(http_route: bool, webhook_flow: bool, webhook_script: bool) -> ServerToSet {
ServerToSet { http_route, webhook_flow, webhook_script }
}
}
fn header_to_pascal_case(header: &str) -> String {
header
.split(|c: char| c == '-' || c == '_' || c == ' ')
@@ -582,7 +578,17 @@ fn generate_components(future_paths: &[FuturePath]) -> Map<String, Value> {
"application/octet-stream": {}
}
},
DEFAULT_SYNC_SSE_RESPONSE_KEY: {
"description": "Returns an SSE stream.",
"content": {
"text/event-stream": {
"schema": {
"type": "string",
"description": "Stream of SSE"
},
}
}
}
}));
components
@@ -652,7 +658,6 @@ struct GenerateOpenAPI {
openapi_spec_format: Format,
}
#[cfg(feature = "http_trigger")]
async fn http_routes_to_future_paths(
db: &DB,
user_db: UserDB,
@@ -683,7 +688,7 @@ async fn http_routes_to_future_paths(
struct MinifiedHttpTrigger {
route_path: String,
http_method: HttpMethod,
is_async: bool,
request_type: RequestType,
workspaced_route: bool,
summary: Option<String>,
description: Option<String>,
@@ -697,7 +702,7 @@ async fn http_routes_to_future_paths(
SELECT
route_path,
http_method AS "http_method: _",
is_async,
request_type AS "request_type: _",
workspaced_route,
summary,
description,
@@ -765,7 +770,7 @@ async fn http_routes_to_future_paths(
let future_path = FuturePath::new(
route_path,
Kind::HttpRoute(HttpRouteConfig::new(method)),
Some(http_route.is_async),
Some(http_route.request_type),
http_route.summary,
http_route.description,
auth_method,
@@ -777,18 +782,6 @@ async fn http_routes_to_future_paths(
Ok(openapi_future_paths)
}
#[cfg(not(feature = "http_trigger"))]
async fn http_routes_to_future_paths(
_db: &DB,
_user_db: UserDB,
_authed: &ApiAuthed,
_pg_pool: &mut PgConnection,
_http_route_filters: Option<&[HttpRouteFilter]>,
_w_id: &str,
) -> Result<Vec<FuturePath>> {
Ok(Vec::new())
}
async fn webhook_to_future_paths(
pg_pool: &mut PgConnection,
webhook_filters: Option<&[WebhookFilter]>,
backend/windmill-api/src/triggers/http/handler.rs
+127 -57
View File
@@ -1,10 +1,11 @@
use super::{
http_trigger_args::RawHttpTriggerArgs, AuthenticationMethod, HttpMethod, TriggerRoute,
HTTP_ACCESS_CACHE, HTTP_AUTH_CACHE, HTTP_ROUTERS_CACHE,
http_trigger_args::RawHttpTriggerArgs, AuthenticationMethod, HttpMethod, RequestType,
TriggerRoute, HTTP_ACCESS_CACHE, HTTP_AUTH_CACHE, HTTP_ROUTERS_CACHE,
};
use crate::{
auth::{AuthCache, OptTokened},
db::{ApiAuthed, DB},
jobs::start_job_update_sse_stream,
resources::try_get_resource_from_db_as,
triggers::{
http::{
@@ -12,7 +13,8 @@ use crate::{
RouteExists, ROUTE_PATH_KEY_RE, VALID_ROUTE_PATH_RE,
},
trigger_helpers::{
get_runnable_format, trigger_runnable, trigger_runnable_and_wait_for_result, RunnableId,
get_runnable_format, trigger_runnable, trigger_runnable_and_wait_for_result,
trigger_runnable_inner, RunnableId,
},
Trigger, TriggerCrud, TriggerData,
},
@@ -26,6 +28,7 @@ use axum::{
routing::{get, post},
Extension, Json, Router,
};
use futures::StreamExt;
use http::{HeaderMap, StatusCode};
use sqlx::PgConnection;
use std::{
@@ -174,33 +177,35 @@ pub async fn insert_new_trigger_into_db(
) -> Result<()> {
require_admin(authed.is_admin, &authed.username)?;
let request_type = trigger.config.request_type;
sqlx::query!(
r#"
INSERT INTO http_trigger (
workspace_id,
path,
route_path,
workspace_id,
path,
route_path,
route_path_key,
workspaced_route,
authentication_resource_path,
wrap_body,
raw_string,
script_path,
script_path,
summary,
description,
is_flow,
is_async,
authentication_method,
http_method,
static_asset_config,
edited_by,
email,
edited_at,
is_flow,
request_type,
authentication_method,
http_method,
static_asset_config,
edited_by,
email,
edited_at,
is_static_website,
error_handler_path,
error_handler_args,
retry
)
)
VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, now(), $19, $20, $21, $22
)
@@ -217,7 +222,7 @@ pub async fn insert_new_trigger_into_db(
trigger.config.summary,
trigger.config.description,
trigger.base.is_flow,
trigger.config.is_async,
request_type as _,
trigger.config.authentication_method as _,
trigger.config.http_method as _,
trigger.config.static_asset_config as _,
@@ -358,7 +363,7 @@ impl TriggerCrud for HttpTrigger {
const ADDITIONAL_SELECT_FIELDS: &[&'static str] = &[
"route_path",
"route_path_key",
"is_async",
"request_type",
"authentication_method",
"http_method",
"summary",
@@ -459,35 +464,37 @@ impl TriggerCrud for HttpTrigger {
let route_path_key =
check_if_route_exist(db, &trigger.config, workspace_id, Some(path)).await?;
let request_type = trigger.config.request_type;
sqlx::query!(
r#"
UPDATE
http_trigger
SET
route_path = $1,
UPDATE
http_trigger
SET
route_path = $1,
route_path_key = $2,
workspaced_route = $3,
wrap_body = $4,
raw_string = $5,
authentication_resource_path = $6,
script_path = $7,
path = $8,
is_flow = $9,
http_method = $10,
static_asset_config = $11,
edited_by = $12,
email = $13,
is_async = $14,
authentication_method = $15,
script_path = $7,
path = $8,
is_flow = $9,
http_method = $10,
static_asset_config = $11,
edited_by = $12,
email = $13,
request_type = $14,
authentication_method = $15,
summary = $16,
description = $17,
edited_at = now(),
edited_at = now(),
is_static_website = $18,
error_handler_path = $19,
error_handler_args = $20,
retry = $21
WHERE
workspace_id = $22 AND
WHERE
workspace_id = $22 AND
path = $23
"#,
route_path,
@@ -503,7 +510,7 @@ impl TriggerCrud for HttpTrigger {
trigger.config.static_asset_config as _,
&authed.username,
&authed.email,
trigger.config.is_async,
request_type as _,
trigger.config.authentication_method as _,
trigger.config.summary,
trigger.config.description,
@@ -517,32 +524,34 @@ impl TriggerCrud for HttpTrigger {
.execute(&mut *tx)
.await?;
} else {
let request_type = trigger.config.request_type;
sqlx::query!(
r#"
UPDATE
http_trigger
SET
UPDATE
http_trigger
SET
wrap_body = $1,
raw_string = $2,
authentication_resource_path = $3,
script_path = $4,
path = $5,
is_flow = $6,
http_method = $7,
static_asset_config = $8,
edited_by = $9,
email = $10,
is_async = $11,
authentication_method = $12,
script_path = $4,
path = $5,
is_flow = $6,
http_method = $7,
static_asset_config = $8,
edited_by = $9,
email = $10,
request_type = $11,
authentication_method = $12,
summary = $13,
description = $14,
edited_at = now(),
edited_at = now(),
is_static_website = $15,
error_handler_path = $16,
error_handler_args = $17,
retry = $18
WHERE
workspace_id = $19 AND
WHERE
workspace_id = $19 AND
path = $20
"#,
trigger.config.wrap_body,
@@ -555,7 +564,7 @@ impl TriggerCrud for HttpTrigger {
trigger.config.static_asset_config as _,
&authed.username,
&authed.email,
trigger.config.is_async,
request_type as _,
trigger.config.authentication_method as _,
trigger.config.summary,
trigger.config.description,
@@ -793,6 +802,7 @@ async fn route_job(
args: RawHttpTriggerArgs,
) -> std::result::Result<impl IntoResponse, Response> {
let route_path = route_path.to_path().trim_end_matches("/");
let (trigger, called_path, params, authed) = get_http_route_trigger(
route_path,
&auth_cache,
@@ -1021,8 +1031,69 @@ async fn route_job(
)
.map_err(|e| e.into_response())?;
if trigger.is_async {
trigger_runnable(
// Handle execution based on the execution mode
match trigger.request_type {
RequestType::SyncSse => {
// Trigger the job (always async when streaming)
let (uuid, _, _) = trigger_runnable_inner(
&db,
Some(user_db.clone()),
authed.clone(),
&trigger.workspace_id,
&trigger.script_path,
trigger.is_flow,
args,
trigger.retry.as_ref(),
trigger.error_handler_path.as_deref(),
trigger.error_handler_args.as_ref(),
format!("http_trigger/{}", trigger.path),
None,
)
.await
.map_err(|e| e.into_response())?;
// Set up SSE stream
let opt_authed = Some(authed.clone());
let opt_tokened = OptTokened { token: None };
let (tx, rx) = tokio::sync::mpsc::channel(32);
let stream = tokio_stream::wrappers::ReceiverStream::new(rx).map(|x| {
format!(
"data: {}\n\n",
serde_json::to_string(&x).unwrap_or_default()
)
});
start_job_update_sse_stream(
opt_authed,
opt_tokened,
db.clone(),
trigger.workspace_id.clone(),
uuid,
None,
None,
None,
None,
Some(true),
Some(true),
None,
None,
tx,
None,
);
let body = axum::body::Body::from_stream(
stream.map(std::result::Result::<_, std::convert::Infallible>::Ok),
);
Ok(Response::builder()
.status(200)
.header("Content-Type", "text/event-stream")
.header("Cache-Control", "no-cache")
.body(body)
.map_err(|e| Error::internal_err(e.to_string()).into_response())?)
}
RequestType::Async => trigger_runnable(
&db,
Some(user_db),
authed,
@@ -1037,9 +1108,8 @@ async fn route_job(
None,
)
.await
.map_err(|e| e.into_response())
} else {
trigger_runnable_and_wait_for_result(
.map_err(|e| e.into_response()),
RequestType::Sync => trigger_runnable_and_wait_for_result(
&db,
Some(user_db),
authed,
@@ -1053,6 +1123,6 @@ async fn route_job(
format!("http_trigger/{}", trigger.path),
)
.await
.map_err(|e| e.into_response())
.map_err(|e| e.into_response()),
}
}
backend/windmill-api/src/triggers/http/mod.rs
+144 -17
View File
@@ -35,7 +35,7 @@ pub struct TriggerRoute {
is_flow: bool,
route_path: String,
workspace_id: String,
is_async: bool,
request_type: RequestType,
authentication_method: AuthenticationMethod,
edited_by: String,
email: String,
@@ -66,6 +66,15 @@ pub enum HttpMethod {
Patch,
}
#[derive(Serialize, Deserialize, sqlx::Type, Debug, Clone, Copy, PartialEq)]
#[sqlx(type_name = "REQUEST_TYPE", rename_all = "snake_case")]
#[serde(rename_all = "snake_case")]
pub enum RequestType {
Sync,
Async,
SyncSse,
}
impl TryFrom<&http::Method> for HttpMethod {
type Error = Error;
fn try_from(method: &http::Method) -> Result<Self> {
@@ -96,7 +105,7 @@ pub enum AuthenticationMethod {
pub struct HttpConfig {
pub route_path: String,
pub route_path_key: String,
pub is_async: bool,
pub request_type: RequestType,
pub authentication_method: AuthenticationMethod,
pub http_method: HttpMethod,
pub summary: Option<String>,
@@ -109,11 +118,11 @@ pub struct HttpConfig {
pub raw_string: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize)]
pub struct HttpConfigRequest {
#[serde(default)]
pub route_path: String,
pub is_async: bool,
pub request_type: RequestType,
pub authentication_method: AuthenticationMethod,
pub http_method: HttpMethod,
pub summary: Option<String>,
@@ -126,6 +135,61 @@ pub struct HttpConfigRequest {
pub raw_string: Option<bool>,
}
#[derive(Deserialize)]
struct HttpConfigRequestHelper {
#[serde(default)]
route_path: String,
request_type: Option<RequestType>,
is_async: Option<bool>,
authentication_method: AuthenticationMethod,
http_method: HttpMethod,
summary: Option<String>,
description: Option<String>,
static_asset_config: Option<SqlxJson<S3Object>>,
is_static_website: bool,
authentication_resource_path: Option<String>,
workspaced_route: Option<bool>,
wrap_body: Option<bool>,
raw_string: Option<bool>,
}
impl<'de> Deserialize<'de> for HttpConfigRequest {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
let helper = HttpConfigRequestHelper::deserialize(deserializer)?;
// Determine request_type with backward compatibility
let request_type = if let Some(mode) = helper.request_type {
mode
} else if let Some(is_async) = helper.is_async {
if is_async {
RequestType::Async
} else {
RequestType::Sync
}
} else {
RequestType::Sync
};
Ok(HttpConfigRequest {
route_path: helper.route_path,
request_type,
authentication_method: helper.authentication_method,
http_method: helper.http_method,
summary: helper.summary,
description: helper.description,
static_asset_config: helper.static_asset_config,
is_static_website: helper.is_static_website,
authentication_resource_path: helper.authentication_resource_path,
workspaced_route: helper.workspaced_route,
wrap_body: helper.wrap_body,
raw_string: helper.raw_string,
})
}
}
// Regex patterns for route validation
lazy_static::lazy_static! {
static ref ROUTE_PATH_KEY_RE: regex::Regex = regex::Regex::new(r"/?:[-\w]+").unwrap();
@@ -174,17 +238,17 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route
let triggers = sqlx::query_as!(
TriggerRoute,
r#"
SELECT
path,
script_path,
is_flow,
route_path,
SELECT
path,
script_path,
is_flow,
route_path,
authentication_resource_path,
workspace_id,
is_async,
authentication_method AS "authentication_method: _",
edited_by,
email,
workspace_id,
request_type AS "request_type: _",
authentication_method AS "authentication_method: _",
edited_by,
email,
static_asset_config AS "static_asset_config: _",
wrap_body,
raw_string,
@@ -193,9 +257,9 @@ pub async fn refresh_routers(db: &DB) -> Result<(bool, RwLockReadGuard<'_, Route
error_handler_path,
error_handler_args as "error_handler_args: _",
retry as "retry: _"
FROM
http_trigger
WHERE
FROM
http_trigger
WHERE
http_method = $1
"#,
&http_method as &HttpMethod
@@ -277,3 +341,66 @@ pub async fn refresh_routers_loop(
}
});
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_request_type_backward_compatibility() {
// Test with new request_type field
let json_new = r#"{
"route_path": "/test",
"request_type": "sync_sse",
"authentication_method": "none",
"http_method": "get",
"is_static_website": false
}"#;
let config: HttpConfigRequest = serde_json::from_str(json_new).unwrap();
assert_eq!(config.request_type, RequestType::SyncSse);
// Test with legacy is_async = true
let json_legacy_async = r#"{
"route_path": "/test",
"is_async": true,
"authentication_method": "none",
"http_method": "get",
"is_static_website": false
}"#;
let config: HttpConfigRequest = serde_json::from_str(json_legacy_async).unwrap();
assert_eq!(config.request_type, RequestType::Async);
// Test with legacy is_async = false
let json_legacy_sync = r#"{
"route_path": "/test",
"is_async": false,
"authentication_method": "none",
"http_method": "get",
"is_static_website": false
}"#;
let config: HttpConfigRequest = serde_json::from_str(json_legacy_sync).unwrap();
assert_eq!(config.request_type, RequestType::Sync);
// Test with neither field (default to sync)
let json_default = r#"{
"route_path": "/test",
"authentication_method": "none",
"http_method": "get",
"is_static_website": false
}"#;
let config: HttpConfigRequest = serde_json::from_str(json_default).unwrap();
assert_eq!(config.request_type, RequestType::Sync);
// Test that request_type takes precedence over is_async
let json_both = r#"{
"route_path": "/test",
"request_type": "sync_sse",
"is_async": true,
"authentication_method": "none",
"http_method": "get",
"is_static_website": false
}"#;
let config: HttpConfigRequest = serde_json::from_str(json_both).unwrap();
assert_eq!(config.request_type, RequestType::SyncSse);
}
}
backend/windmill-api/src/triggers/trigger_helpers.rs
+1 -1
View File
@@ -477,7 +477,7 @@ pub trait TriggerJobArgs {
}
#[allow(dead_code)]
async fn trigger_runnable_inner(
pub async fn trigger_runnable_inner(
db: &DB,
user_db: Option<UserDB>,
authed: ApiAuthed,
frontend/src/lib/components/triggers/http/RouteEditorInner.svelte
+17 -10
View File
@@ -76,7 +76,7 @@
let isValid = $state(false)
let dirtyRoutePath = $state(false)
let dirtyPath = $state(false)
let is_async = $state(false)
let request_type = $state<'sync' | 'async' | 'sync_sse'>('sync')
let authentication_method = $state<AuthenticationMethod>('none')
let route_path = $state('')
let http_method = $state<'get' | 'post' | 'put' | 'patch' | 'delete'>('post')
@@ -223,7 +223,7 @@
is_flow = defaultValues?.is_flow ?? nis_flow
edit = false
itemKind = nis_flow ? 'flow' : 'script'
is_async = defaultValues?.is_async ?? false
request_type = defaultValues?.request_type ?? 'sync'
authentication_method = defaultValues?.authentication_method ?? 'none'
route_path = defaultValues?.route_path ?? ''
dirtyRoutePath = false
@@ -263,7 +263,7 @@
path = cfg?.path ?? ''
route_path = cfg?.route_path ?? ''
http_method = cfg?.http_method ?? 'post'
is_async = cfg?.is_async ?? false
request_type = cfg?.request_type ?? 'sync'
workspaced_route = cfg?.workspaced_route ?? false
wrap_body = cfg?.wrap_body ?? false
raw_string = cfg?.raw_string ?? false
@@ -342,7 +342,7 @@
path,
route_path,
http_method,
is_async,
request_type,
workspaced_route,
wrap_body,
raw_string,
@@ -464,7 +464,7 @@
initialScriptPath = ''
is_flow = false
http_method = 'get'
is_async = false
request_type = 'sync'
is_static_website = ev.detail === 'static_website'
if (is_static_website) {
authentication_method = 'none'
@@ -617,13 +617,20 @@
{#snippet action()}
<ToggleButtonGroup
class="w-auto h-full"
selected={is_async ? 'async' : 'sync'}
selected={request_type}
on:selected={({ detail }) => {
is_async = detail === 'async'
request_type = detail
}}
disabled={!can_write || !!static_asset_config}
>
{#snippet children({ item, disabled })}
<ToggleButton
label="Sync"
value="sync"
tooltip="Triggers the execution, wait for the job to complete and return it as a response."
{item}
{disabled}
/>
<ToggleButton
label="Async"
value="async"
@@ -632,9 +639,9 @@
{disabled}
/>
<ToggleButton
label="Sync"
value="sync"
tooltip="Triggers the execution, wait for the job to complete and return it as a response."
label="Sync SSE"
value="sync_sse"
tooltip="Triggers the execution and returns an SSE stream."
{item}
{disabled}
/>
frontend/src/lib/components/triggers/http/utils.ts
+2 -2
View File
@@ -46,7 +46,7 @@ export async function saveHttpRouteFromCfg(
path: routeCfg.path,
script_path: routeCfg.script_path,
is_flow: routeCfg.is_flow,
is_async: routeCfg.is_async,
request_type: routeCfg.request_type,
authentication_method: routeCfg.authentication_method,
route_path: routeCfg.route_path,
http_method: routeCfg.http_method,
@@ -139,7 +139,7 @@ function processOpenApiDocument(
http_method: method,
authentication_method: 'none',
workspaced_route: false,
is_async: true,
request_type: 'async',
script_path: '',
raw_string: false,
is_flow: false,
frontend/src/lib/components/triggers/webhook/WebhooksConfigSection.svelte
+93 -67
View File
@@ -10,7 +10,7 @@
} from '$lib/consts'
import bash from 'svelte-highlight/languages/bash'
import { Tabs, Tab, TabContent, Button } from '$lib/components/common'
import { ArrowDownRight, ArrowUpRight, Clipboard, RssIcon } from 'lucide-svelte'
import { ArrowDownRight, ArrowUpRight, Clipboard } from 'lucide-svelte'
import { Highlight } from 'svelte-highlight'
import { typescript } from 'svelte-highlight/languages'
import ClipboardPanel from '../../details/ClipboardPanel.svelte'
@@ -49,7 +49,6 @@
hash?: string
path: string
}
sse: {}
}
sync: {
get: {
@@ -59,7 +58,13 @@
hash?: string
path: string
}
sse: {
}
sync_sse: {
get: {
hash?: string
path: string
}
post: {
hash?: string
path: string
}
@@ -67,23 +72,17 @@
} = $derived(isFlow ? computeFlowWebhooks(path) : computeScriptWebhooks(hash, path))
let selectedTab: string = $state('rest')
let userSettings: UserSettings | undefined = $state()
let requestType = $state(DEFAULT_WEBHOOK_TYPE) as 'async' | 'sync'
let callMethod = $state('post') as 'get' | 'post' | 'sse'
let requestType = $state(DEFAULT_WEBHOOK_TYPE) as 'async' | 'sync' | 'sync_sse'
let callMethod = $state('post') as 'get' | 'post'
let runnableId = $state('path') as 'hash' | 'path'
let tokenType = $state('headers') as 'query' | 'headers'
$effect(() => {
if (requestType === 'async' && (callMethod === 'get' || callMethod === 'sse')) {
if (requestType === 'async' && callMethod === 'get') {
callMethod = 'post'
}
})
$effect(() => {
if (callMethod === 'sse' && tokenType === 'headers') {
tokenType = 'query'
}
})
let cleanedRunnableArgs = $derived.by(() => {
readFieldsRecursively(runnableArgs)
return isObject(runnableArgs) && 'wm_trigger' in runnableArgs
@@ -94,7 +93,7 @@
webhooks[requestType][callMethod][runnableId] +
(tokenType === 'query'
? `?token=${token}${
callMethod === 'get' || callMethod === 'sse'
callMethod === 'get' || requestType === 'sync_sse'
? `&payload=${encodeURIComponent(btoa(JSON.stringify(cleanedRunnableArgs ?? {})))}`
: ''
}`
@@ -113,8 +112,7 @@
post: {
hash: `${webhookBase}/run/h/${hash}`,
path: `${webhookBase}/run/p/${path}`
},
sse: {}
}
},
sync: {
get: {
@@ -123,8 +121,14 @@
post: {
hash: `${webhookBase}/run_wait_result/h/${hash}`,
path: `${webhookBase}/run_wait_result/p/${path}`
}
},
sync_sse: {
get: {
path: `${webhookBase}/run_and_stream/p/${path}`,
hash: `${webhookBase}/run_and_stream/h/${hash}`
},
sse: {
post: {
hash: `${webhookBase}/run_and_stream/h/${hash}`,
path: `${webhookBase}/run_and_stream/p/${path}`
}
@@ -143,8 +147,7 @@
get: {},
post: {
path: urlAsync
},
sse: {}
}
},
sync: {
get: {
@@ -152,8 +155,13 @@
},
post: {
path: urlSync
}
},
sync_sse: {
get: {
path: urlStream
},
sse: {
post: {
path: urlStream
}
}
@@ -173,31 +181,57 @@
}
function fetchCode() {
if (callMethod === 'sse') {
if (requestType === 'sync_sse') {
return `
import { EventSource } from "eventsource";
export async function main() {
const endpoint = \`${url}\`;
const response = await fetch(\`${url}\`, {
method: '${callMethod === 'get' ? 'GET' : 'POST'}',
headers: ${JSON.stringify(headers(), null, 2).replaceAll('\n', '\n ')},
body: ${callMethod === 'get' ? 'undefined' : `JSON.stringify(${JSON.stringify(cleanedRunnableArgs ?? {}, null, 2).replaceAll('\n', '\n ')})`}
});
if (!response.ok) {
const text = await response.text()
throw new Error(\`\${response.status} \${text}\`)
}
return new Promise((resolve, reject) => {
const eventSource = new EventSource(endpoint);
if (!response.body) {
throw new Error("Response body is empty");
}
eventSource.onmessage = (event) => {
const data = JSON.parse(event.data);
console.log(data);
if (data.completed) {
eventSource.close();
resolve();
}
};
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = "";
eventSource.onerror = (error) => {
console.error('EventSource error:', error);
eventSource.close();
reject(error);
};
});
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split("\\n");
// Keep the last incomplete line in buffer
buffer = lines.pop() || "";
for (const line of lines) {
if (line.startsWith("data: ")) {
const jsonData = line.slice(6); // Remove 'data: ' prefix
const data = JSON.parse(jsonData);
console.log(data);
if (data.completed) {
reader.cancel();
return;
}
}
}
}
} catch (error) {
console.error("Stream error:", error);
throw error;
}
}`
}
if (requestType === 'sync') {
@@ -292,16 +326,17 @@ function waitForJobCompletion(UUID) {
return `TOKEN='${token}'
${callMethod !== 'get' ? `BODY='${JSON.stringify(cleanedRunnableArgs ?? {})}'` : ''}
URL='${url}'
${requestType === 'sync' ? 'RESULT' : 'UUID'}=$(curl -s ${
${requestType === 'sync' ? 'RESULT=$(' : requestType === 'async' ? 'UUID=$(' : ''}curl -s ${
callMethod != 'get' ? "-H 'Content-Type: application/json'" : ''
} ${tokenType === 'headers' ? `-H "Authorization: Bearer $TOKEN"` : ''} -X ${
callMethod === 'get' ? 'GET' : 'POST'
} ${callMethod !== 'get' ? `-d "$BODY" ` : ''}$URL)
} ${callMethod !== 'get' ? `-d "$BODY" ` : ''}$URL${requestType === 'sync' || requestType === 'async' ? ')' : ''}
${
requestType === 'sync'
? 'echo -E $RESULT | jq'
: `
: requestType === 'async'
? `
URL="${location.origin}/api/w/${$workspaceStore}/jobs_u/completed/get_result_maybe/$UUID"
while true; do
curl -s -H "Authorization: Bearer $TOKEN" $URL -o res.json
@@ -313,6 +348,7 @@ while true; do
sleep 1
fi
done`
: ''
}`
}
</script>
@@ -370,6 +406,15 @@ done`
tooltip="Triggers the execution, wait for the job to complete and return it as a response."
{item}
/>
<ToggleButton
label="Sync SSE"
value="sync_sse"
tooltip={'Triggers the execution and returns an SSE stream. ' +
(isFlow
? 'Only useful if the last step of the flow returns a stream.'
: 'Only useful if the script returns a stream.')}
{item}
/>
{/snippet}
</ToggleButtonGroup>
</div>
@@ -390,19 +435,7 @@ done`
selectedColor="#fb923c"
value="get"
{item}
disabled={requestType !== 'sync'}
/>
<ToggleButton
label="SSE"
value="sse"
icon={RssIcon}
selectedColor="#3B82F6"
disabled={requestType !== 'sync'}
tooltip={'Returns an SSE stream. ' +
(isFlow
? 'Only useful if the last step of the flow returns a stream.'
: 'Only useful if the script returns a stream.')}
{item}
disabled={requestType !== 'sync' && requestType !== 'sync_sse'}
/>
{/snippet}
</ToggleButtonGroup>
@@ -426,12 +459,7 @@ done`
>
<ToggleButtonGroup class="h-[30px] w-auto" bind:selected={tokenType}>
{#snippet children({ item })}
<ToggleButton
label="Token in Headers"
value="headers"
{item}
disabled={callMethod === 'sse'}
/>
<ToggleButton label="Token in Headers" value="headers" {item} />
<ToggleButton label="Token in Query" value="query" {item} />
{/snippet}
</ToggleButtonGroup>
@@ -443,12 +471,10 @@ done`
<div>
<Tabs bind:selected={selectedTab}>
<Tab value="rest" size="xs">REST</Tab>
{#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL && callMethod !== 'sse'}
{#if SCRIPT_VIEW_SHOW_EXAMPLE_CURL}
<Tab value="curl" size="xs">Curl</Tab>
{/if}
<Tab value="fetch" size="xs">
{callMethod === 'sse' ? 'Event Source' : 'Fetch'}
</Tab>
<Tab value="fetch" size="xs">Fetch</Tab>
{#snippet content()}
{#key token}
@@ -486,7 +512,7 @@ done`
}}
>
<Highlight language={bash} code={curlCode()} />
<Clipboard size={14} class="w-8 top-2 right-2 absolute" />
<Clipboard size={14} class="w-8 top-2 right-2 absolute cursor-pointer" />
</div>
{/key}
{/key}
@@ -508,7 +534,7 @@ done`
}}
>
<Highlight language={typescript} code={fetchCode()} />
<Clipboard size={14} class="w-8 top-2 right-2 absolute" />
<Clipboard size={14} class="w-8 top-2 right-2 absolute cursor-pointer" />
</div>
{/key}{/key}{/key}{/key}
{/key}
frontend/src/lib/utils_deployable.ts
+2 -2
View File
@@ -359,7 +359,7 @@ export async function getTriggerValue(kind: TriggerKind, path: string, workspace
http_method,
route_path,
static_asset_config,
is_async,
request_type,
authentication_method,
is_static_website,
authentication_resource_path
@@ -374,7 +374,7 @@ export async function getTriggerValue(kind: TriggerKind, path: string, workspace
http_method,
route_path,
static_asset_config,
is_async,
request_type,
authentication_method,
is_static_website,
authentication_resource_path