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

feat: Custom content type for script and flow results (#2767)

Browse Source
This commit is contained in:
Guillaume Bouvignies
2023-12-04 13:19:40 +01:00
committed by GitHub
parent 5de6973bba
commit 36db047a86
2 changed files with 76 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api/src/jobs.rs
+55 -13
View File
@@ -6,6 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
use axum::http::HeaderValue;
use serde_json::value::RawValue;
use std::collections::HashMap;
use windmill_common::flow_status::RestartedFrom;
@@ -1874,15 +1875,16 @@ impl Drop for Guard {
}
#[derive(Deserialize)]
pub struct WindmillStatusCode {
pub struct WindmillCompositeResult {
windmill_status_code: Option<u16>,
windmill_content_type: Option<String>,
result: Option<Box<RawValue>>,
}
async fn run_wait_result<T>(
db: &DB,
uuid: Uuid,
Path((w_id, _)): Path<(String, T)>,
node_id: Option<String>,
node_id_for_empty_return: Option<String>,
) -> error::Result<Response> {
let mut result;
let timeout = SERVER_CONFIG.read().await.timeout_wait_result.clone();
@@ -1898,10 +1900,16 @@ async fn run_wait_result<T>(
let mut accumulated_delay = 0 as u64;
loop {
if let Some(node_id) = node_id.as_ref() {
result = get_result_by_id_from_running_flow(&db, &w_id, &uuid, node_id, None)
.await
.ok();
if let Some(node_id_for_empty_return) = node_id_for_empty_return.as_ref() {
result = get_result_by_id_from_running_flow(
&db,
&w_id,
&uuid,
node_id_for_empty_return,
None,
)
.await
.ok();
} else {
let row =
sqlx::query("SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2")
@@ -1934,13 +1942,47 @@ async fn run_wait_result<T>(
if let Some(result) = result {
g.done = true;
let status_code = serde_json::from_str::<WindmillStatusCode>(result.get());
match status_code {
Ok(WindmillStatusCode { windmill_status_code: Some(status_code), result }) => {
Err(Error::CustomStatusCode(
StatusCode::from_u16(status_code).unwrap_or_else(|_| StatusCode::IM_A_TEAPOT),
result,
))
let composite_result = serde_json::from_str::<WindmillCompositeResult>(result.get());
match composite_result {
Ok(WindmillCompositeResult {
windmill_status_code,
windmill_content_type,
result: result_value,
}) => {
if windmill_content_type.is_none() && windmill_status_code.is_none() {
return Ok(Json(result).into_response());
}
let status_code_or_default = windmill_status_code
.map(|val| match StatusCode::from_u16(val) {
Ok(sc) => Ok(sc),
Err(_) => Err(Error::ExecutionErr("Invalid status code".to_string())),
})
.unwrap_or(if result_value.is_some() {
Ok(StatusCode::OK)
} else {
Ok(StatusCode::NO_CONTENT)
})?;
if windmill_content_type.is_some() {
let serialized_result = result_value
.map(|val| val.get().to_owned())
.unwrap_or_else(String::new);
return Ok((
status_code_or_default,
[(
http::header::CONTENT_TYPE,
HeaderValue::from_str(windmill_content_type.unwrap().as_str()).unwrap(),
)],
serialized_result,
)
.into_response());
}
return Ok((
status_code_or_default,
Json(result_value), // default to JSON result if no content type is provided
)
.into_response());
}
_ => Ok(Json(result).into_response()),
}
backend/windmill-common/src/error.rs
+21 -35
View File
@@ -15,7 +15,6 @@ use axum::{
};
use hyper::StatusCode;
use serde_json::value::RawValue;
#[cfg(feature = "sqlx")]
use sqlx::migrate::MigrateError;
use thiserror::Error;
@@ -65,8 +64,6 @@ pub enum Error {
Anyhow(#[from] anyhow::Error),
#[error("Error: {0:#?}")]
JsonErr(serde_json::Value),
#[error("Custom Status Code: {0:#?}")]
CustomStatusCode(StatusCode, Option<Box<RawValue>>),
#[error("{0}")]
OpenAIError(String),
}
@@ -85,41 +82,30 @@ pub fn to_anyhow<T: 'static + std::error::Error + Send + Sync>(e: T) -> anyhow::
#[cfg(feature = "axum")]
impl IntoResponse for Error {
fn into_response(self) -> axum::response::Response<BoxBody> {
match self {
Self::CustomStatusCode(code, result) => {
let mut res = Json(result).into_response();
let status_mut = res.status_mut();
*status_mut = code;
res
let e = &self;
let body = body::boxed(body::Full::from(e.to_string()));
let status = match self {
Self::NotFound(_) => axum::http::StatusCode::NOT_FOUND,
Self::NotAuthorized(_) => axum::http::StatusCode::UNAUTHORIZED,
Self::RequireAdmin(_) => axum::http::StatusCode::FORBIDDEN,
Self::SqlErr(_) | Self::BadRequest(_) | Self::OpenAIError(_) => {
axum::http::StatusCode::BAD_REQUEST
}
_ => {
let e = &self;
let body = body::boxed(body::Full::from(e.to_string()));
_ => axum::http::StatusCode::INTERNAL_SERVER_ERROR,
};
let status = match self {
Self::NotFound(_) => axum::http::StatusCode::NOT_FOUND,
Self::NotAuthorized(_) => axum::http::StatusCode::UNAUTHORIZED,
Self::RequireAdmin(_) => axum::http::StatusCode::FORBIDDEN,
Self::CustomStatusCode(code, _) => code,
Self::SqlErr(_) | Self::BadRequest(_) | Self::OpenAIError(_) => {
axum::http::StatusCode::BAD_REQUEST
}
_ => axum::http::StatusCode::INTERNAL_SERVER_ERROR,
};
if matches!(status, axum::http::StatusCode::NOT_FOUND) {
tracing::warn!(not_found = e.to_string());
} else {
tracing::error!(error = e.to_string());
};
if matches!(status, axum::http::StatusCode::NOT_FOUND) {
tracing::warn!(not_found = e.to_string());
} else {
tracing::error!(error = e.to_string());
};
axum::response::Response::builder()
.header("Content-Type", "text/plain")
.status(status)
.body(body)
.unwrap()
}
}
axum::response::Response::builder()
.header("Content-Type", "text/plain")
.status(status)
.body(body)
.unwrap()
}
}