mirror of
https://github.com/mountain-loop/yaak.git
synced 2026-09-10 20:01:55 +02:00
extract shared yaak send flow and wire CLI
This commit is contained in:
Gregory Schier
@@ -0,0 +1,14 @@
|
||||
[package]
|
||||
name = "yaak"
|
||||
version = "0.1.0"
|
||||
edition = "2024"
|
||||
publish = false
|
||||
|
||||
[dependencies]
|
||||
log = { workspace = true }
|
||||
serde_json = { workspace = true }
|
||||
thiserror = { workspace = true }
|
||||
tokio = { workspace = true, features = ["sync", "rt"] }
|
||||
yaak-http = { workspace = true }
|
||||
yaak-models = { workspace = true }
|
||||
yaak-templates = { workspace = true }
|
||||
@@ -0,0 +1,9 @@
|
||||
use thiserror::Error;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum Error {
|
||||
#[error(transparent)]
|
||||
Send(#[from] crate::send::SendHttpRequestError),
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, Error>;
|
||||
@@ -0,0 +1,6 @@
|
||||
pub mod error;
|
||||
pub mod render;
|
||||
pub mod send;
|
||||
|
||||
pub use error::Error;
|
||||
pub type Result<T> = error::Result<T>;
|
||||
@@ -0,0 +1,157 @@
|
||||
use log::info;
|
||||
use serde_json::Value;
|
||||
use std::collections::BTreeMap;
|
||||
use yaak_http::path_placeholders::apply_path_placeholders;
|
||||
use yaak_models::models::{Environment, HttpRequest, HttpRequestHeader, HttpUrlParameter};
|
||||
use yaak_models::render::make_vars_hashmap;
|
||||
use yaak_templates::{RenderOptions, TemplateCallback, parse_and_render, render_json_value_raw};
|
||||
|
||||
pub async fn render_http_request<T: TemplateCallback>(
|
||||
request: &HttpRequest,
|
||||
environment_chain: Vec<Environment>,
|
||||
callback: &T,
|
||||
options: &RenderOptions,
|
||||
) -> yaak_templates::error::Result<HttpRequest> {
|
||||
let vars = &make_vars_hashmap(environment_chain);
|
||||
|
||||
let mut url_parameters = Vec::new();
|
||||
for parameter in request.url_parameters.clone() {
|
||||
if !parameter.enabled {
|
||||
continue;
|
||||
}
|
||||
|
||||
url_parameters.push(HttpUrlParameter {
|
||||
enabled: parameter.enabled,
|
||||
name: parse_and_render(parameter.name.as_str(), vars, callback, options).await?,
|
||||
value: parse_and_render(parameter.value.as_str(), vars, callback, options).await?,
|
||||
id: parameter.id,
|
||||
})
|
||||
}
|
||||
|
||||
let mut headers = Vec::new();
|
||||
for header in request.headers.clone() {
|
||||
if !header.enabled {
|
||||
continue;
|
||||
}
|
||||
|
||||
headers.push(HttpRequestHeader {
|
||||
enabled: header.enabled,
|
||||
name: parse_and_render(header.name.as_str(), vars, callback, options).await?,
|
||||
value: parse_and_render(header.value.as_str(), vars, callback, options).await?,
|
||||
id: header.id,
|
||||
})
|
||||
}
|
||||
|
||||
let mut body = BTreeMap::new();
|
||||
for (key, value) in request.body.clone() {
|
||||
let value = if key == "form" { strip_disabled_form_entries(value) } else { value };
|
||||
body.insert(key, render_json_value_raw(value, vars, callback, options).await?);
|
||||
}
|
||||
|
||||
let authentication = {
|
||||
let mut disabled = false;
|
||||
let mut auth = BTreeMap::new();
|
||||
|
||||
match request.authentication.get("disabled") {
|
||||
Some(Value::Bool(true)) => {
|
||||
disabled = true;
|
||||
}
|
||||
Some(Value::String(template)) => {
|
||||
disabled = parse_and_render(template.as_str(), vars, callback, options)
|
||||
.await
|
||||
.unwrap_or_default()
|
||||
.is_empty();
|
||||
info!(
|
||||
"Rendering authentication.disabled as a template: {disabled} from \"{template}\""
|
||||
);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
|
||||
if disabled {
|
||||
auth.insert("disabled".to_string(), Value::Bool(true));
|
||||
} else {
|
||||
for (key, value) in request.authentication.clone() {
|
||||
if key == "disabled" {
|
||||
auth.insert(key, Value::Bool(false));
|
||||
} else {
|
||||
auth.insert(key, render_json_value_raw(value, vars, callback, options).await?);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
auth
|
||||
};
|
||||
|
||||
let url = parse_and_render(request.url.clone().as_str(), vars, callback, options).await?;
|
||||
let (url, url_parameters) = apply_path_placeholders(&url, &url_parameters);
|
||||
|
||||
Ok(HttpRequest { url, url_parameters, headers, body, authentication, ..request.to_owned() })
|
||||
}
|
||||
|
||||
fn strip_disabled_form_entries(v: Value) -> Value {
|
||||
match v {
|
||||
Value::Array(items) => Value::Array(
|
||||
items
|
||||
.into_iter()
|
||||
.filter(|item| item.get("enabled").and_then(|e| e.as_bool()).unwrap_or(true))
|
||||
.collect(),
|
||||
),
|
||||
v => v,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use serde_json::json;
|
||||
|
||||
#[test]
|
||||
fn test_strip_disabled_form_entries() {
|
||||
let input = json!([
|
||||
{"enabled": true, "name": "foo", "value": "bar"},
|
||||
{"enabled": false, "name": "disabled", "value": "gone"},
|
||||
{"enabled": true, "name": "baz", "value": "qux"},
|
||||
]);
|
||||
let result = strip_disabled_form_entries(input);
|
||||
assert_eq!(
|
||||
result,
|
||||
json!([
|
||||
{"enabled": true, "name": "foo", "value": "bar"},
|
||||
{"enabled": true, "name": "baz", "value": "qux"},
|
||||
])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_strip_disabled_form_entries_all_disabled() {
|
||||
let input = json!([
|
||||
{"enabled": false, "name": "a", "value": "b"},
|
||||
{"enabled": false, "name": "c", "value": "d"},
|
||||
]);
|
||||
let result = strip_disabled_form_entries(input);
|
||||
assert_eq!(result, json!([]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_strip_disabled_form_entries_missing_enabled_defaults_to_kept() {
|
||||
let input = json!([
|
||||
{"name": "no_enabled_field", "value": "kept"},
|
||||
{"enabled": false, "name": "disabled", "value": "gone"},
|
||||
]);
|
||||
let result = strip_disabled_form_entries(input);
|
||||
assert_eq!(
|
||||
result,
|
||||
json!([
|
||||
{"name": "no_enabled_field", "value": "kept"},
|
||||
])
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_strip_disabled_form_entries_non_array_passthrough() {
|
||||
let input = json!("just a string");
|
||||
let result = strip_disabled_form_entries(input.clone());
|
||||
assert_eq!(result, input);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,310 @@
|
||||
use crate::render::render_http_request;
|
||||
use log::warn;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::Instant;
|
||||
use thiserror::Error;
|
||||
use tokio::sync::mpsc;
|
||||
use yaak_http::sender::{HttpResponseEvent as SenderHttpResponseEvent, HttpSender, ReqwestSender};
|
||||
use yaak_http::types::{SendableBody, SendableHttpRequest, SendableHttpRequestOptions};
|
||||
use yaak_models::blob_manager::BlobManager;
|
||||
use yaak_models::models::{HttpResponse, HttpResponseEvent, HttpResponseHeader, HttpResponseState};
|
||||
use yaak_models::query_manager::QueryManager;
|
||||
use yaak_models::util::UpdateSource;
|
||||
use yaak_templates::{RenderOptions, TemplateCallback};
|
||||
|
||||
const HTTP_EVENT_CHANNEL_CAPACITY: usize = 100;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum SendHttpRequestError {
|
||||
#[error("Failed to load request: {0}")]
|
||||
LoadRequest(#[source] yaak_models::error::Error),
|
||||
|
||||
#[error("Failed to resolve environments: {0}")]
|
||||
ResolveEnvironments(#[source] yaak_models::error::Error),
|
||||
|
||||
#[error("Failed to resolve inherited request settings: {0}")]
|
||||
ResolveRequestInheritance(#[source] yaak_models::error::Error),
|
||||
|
||||
#[error("Failed to render request templates: {0}")]
|
||||
RenderRequest(#[source] yaak_templates::error::Error),
|
||||
|
||||
#[error("Failed to persist response metadata: {0}")]
|
||||
PersistResponse(#[source] yaak_models::error::Error),
|
||||
|
||||
#[error("Failed to create HTTP client: {0}")]
|
||||
CreateHttpClient(#[source] yaak_http::error::Error),
|
||||
|
||||
#[error("Failed to build sendable request: {0}")]
|
||||
BuildSendableRequest(#[source] yaak_http::error::Error),
|
||||
|
||||
#[error("Failed to send request: {0}")]
|
||||
SendRequest(#[source] yaak_http::error::Error),
|
||||
|
||||
#[error("Failed to read response body: {0}")]
|
||||
ReadResponseBody(#[source] yaak_http::error::Error),
|
||||
|
||||
#[error("Failed to create response directory {path:?}: {source}")]
|
||||
CreateResponseDirectory {
|
||||
path: PathBuf,
|
||||
#[source]
|
||||
source: std::io::Error,
|
||||
},
|
||||
|
||||
#[error("Failed to write response body to {path:?}: {source}")]
|
||||
WriteResponseBody {
|
||||
path: PathBuf,
|
||||
#[source]
|
||||
source: std::io::Error,
|
||||
},
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, SendHttpRequestError>;
|
||||
|
||||
pub struct SendHttpRequestByIdParams<'a, T: TemplateCallback> {
|
||||
pub query_manager: &'a QueryManager,
|
||||
pub blob_manager: &'a BlobManager,
|
||||
pub request_id: &'a str,
|
||||
pub environment_id: Option<&'a str>,
|
||||
pub template_callback: &'a T,
|
||||
pub send_options: SendableHttpRequestOptions,
|
||||
pub update_source: UpdateSource,
|
||||
pub response_dir: &'a Path,
|
||||
pub persist_events: bool,
|
||||
pub emit_events_to: Option<mpsc::Sender<SenderHttpResponseEvent>>,
|
||||
}
|
||||
|
||||
pub struct SendHttpRequestResult {
|
||||
pub rendered_request: yaak_models::models::HttpRequest,
|
||||
pub response: HttpResponse,
|
||||
pub response_body: Vec<u8>,
|
||||
}
|
||||
|
||||
pub async fn send_http_request_by_id<T: TemplateCallback>(
|
||||
params: SendHttpRequestByIdParams<'_, T>,
|
||||
) -> Result<SendHttpRequestResult> {
|
||||
let db = params.query_manager.connect();
|
||||
let request =
|
||||
db.get_http_request(params.request_id).map_err(SendHttpRequestError::LoadRequest)?;
|
||||
let environment_chain = db
|
||||
.resolve_environments(
|
||||
&request.workspace_id,
|
||||
request.folder_id.as_deref(),
|
||||
params.environment_id,
|
||||
)
|
||||
.map_err(SendHttpRequestError::ResolveEnvironments)?;
|
||||
|
||||
let (authentication_type, authentication, _auth_context_id) = db
|
||||
.resolve_auth_for_http_request(&request)
|
||||
.map_err(SendHttpRequestError::ResolveRequestInheritance)?;
|
||||
let resolved_headers = db
|
||||
.resolve_headers_for_http_request(&request)
|
||||
.map_err(SendHttpRequestError::ResolveRequestInheritance)?;
|
||||
drop(db);
|
||||
|
||||
let mut resolved_request = request.clone();
|
||||
resolved_request.authentication_type = authentication_type;
|
||||
resolved_request.authentication = authentication;
|
||||
resolved_request.headers = resolved_headers;
|
||||
|
||||
let rendered_request = render_http_request(
|
||||
&resolved_request,
|
||||
environment_chain,
|
||||
params.template_callback,
|
||||
&RenderOptions::throw(),
|
||||
)
|
||||
.await
|
||||
.map_err(SendHttpRequestError::RenderRequest)?;
|
||||
|
||||
let sendable_request =
|
||||
SendableHttpRequest::from_http_request(&rendered_request, params.send_options)
|
||||
.await
|
||||
.map_err(SendHttpRequestError::BuildSendableRequest)?;
|
||||
let request_content_length = sendable_body_length(sendable_request.body.as_ref());
|
||||
|
||||
let mut response = params
|
||||
.query_manager
|
||||
.connect()
|
||||
.upsert_http_response(
|
||||
&HttpResponse {
|
||||
request_id: request.id.clone(),
|
||||
workspace_id: request.workspace_id.clone(),
|
||||
request_content_length,
|
||||
request_headers: sendable_request
|
||||
.headers
|
||||
.iter()
|
||||
.map(|(name, value)| HttpResponseHeader {
|
||||
name: name.clone(),
|
||||
value: value.clone(),
|
||||
})
|
||||
.collect(),
|
||||
url: sendable_request.url.clone(),
|
||||
..Default::default()
|
||||
},
|
||||
¶ms.update_source,
|
||||
params.blob_manager,
|
||||
)
|
||||
.map_err(SendHttpRequestError::PersistResponse)?;
|
||||
|
||||
let (event_tx, mut event_rx) =
|
||||
mpsc::channel::<SenderHttpResponseEvent>(HTTP_EVENT_CHANNEL_CAPACITY);
|
||||
let event_query_manager = params.query_manager.clone();
|
||||
let event_response_id = response.id.clone();
|
||||
let event_workspace_id = request.workspace_id.clone();
|
||||
let event_update_source = params.update_source.clone();
|
||||
let emit_events_to = params.emit_events_to.clone();
|
||||
let persist_events = params.persist_events;
|
||||
let event_handle = tokio::spawn(async move {
|
||||
while let Some(event) = event_rx.recv().await {
|
||||
if persist_events {
|
||||
let db_event = HttpResponseEvent::new(
|
||||
&event_response_id,
|
||||
&event_workspace_id,
|
||||
event.clone().into(),
|
||||
);
|
||||
if let Err(err) = event_query_manager
|
||||
.connect()
|
||||
.upsert_http_response_event(&db_event, &event_update_source)
|
||||
{
|
||||
warn!("Failed to persist HTTP response event: {}", err);
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(tx) = emit_events_to.as_ref() {
|
||||
let _ = tx.try_send(event);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
let sender = ReqwestSender::new().map_err(SendHttpRequestError::CreateHttpClient)?;
|
||||
let started_at = Instant::now();
|
||||
let request_started_url = sendable_request.url.clone();
|
||||
|
||||
let http_response = match sender.send(sendable_request, event_tx).await {
|
||||
Ok(response) => response,
|
||||
Err(err) => {
|
||||
let _ = params
|
||||
.query_manager
|
||||
.connect()
|
||||
.upsert_http_response(
|
||||
&HttpResponse {
|
||||
state: HttpResponseState::Closed,
|
||||
elapsed: duration_to_i32(started_at.elapsed()),
|
||||
elapsed_headers: duration_to_i32(started_at.elapsed()),
|
||||
error: Some(err.to_string()),
|
||||
url: request_started_url,
|
||||
..response
|
||||
},
|
||||
¶ms.update_source,
|
||||
params.blob_manager,
|
||||
)
|
||||
.map_err(SendHttpRequestError::PersistResponse)?;
|
||||
|
||||
if let Err(join_err) = event_handle.await {
|
||||
warn!("Failed to join response event task: {}", join_err);
|
||||
}
|
||||
|
||||
return Err(SendHttpRequestError::SendRequest(err));
|
||||
}
|
||||
};
|
||||
|
||||
let headers_elapsed = duration_to_i32(started_at.elapsed());
|
||||
response = params
|
||||
.query_manager
|
||||
.connect()
|
||||
.upsert_http_response(
|
||||
&HttpResponse {
|
||||
state: HttpResponseState::Connected,
|
||||
elapsed_headers: headers_elapsed,
|
||||
status: i32::from(http_response.status),
|
||||
status_reason: http_response.status_reason.clone(),
|
||||
url: http_response.url.clone(),
|
||||
remote_addr: http_response.remote_addr.clone(),
|
||||
version: http_response.version.clone(),
|
||||
headers: http_response
|
||||
.headers
|
||||
.iter()
|
||||
.map(|(name, value)| HttpResponseHeader {
|
||||
name: name.clone(),
|
||||
value: value.clone(),
|
||||
})
|
||||
.collect(),
|
||||
request_headers: http_response
|
||||
.request_headers
|
||||
.iter()
|
||||
.map(|(name, value)| HttpResponseHeader {
|
||||
name: name.clone(),
|
||||
value: value.clone(),
|
||||
})
|
||||
.collect(),
|
||||
..response
|
||||
},
|
||||
¶ms.update_source,
|
||||
params.blob_manager,
|
||||
)
|
||||
.map_err(SendHttpRequestError::PersistResponse)?;
|
||||
|
||||
let (response_body, body_stats) =
|
||||
http_response.bytes().await.map_err(SendHttpRequestError::ReadResponseBody)?;
|
||||
|
||||
std::fs::create_dir_all(params.response_dir).map_err(|source| {
|
||||
SendHttpRequestError::CreateResponseDirectory {
|
||||
path: params.response_dir.to_path_buf(),
|
||||
source,
|
||||
}
|
||||
})?;
|
||||
|
||||
let body_path = params.response_dir.join(&response.id);
|
||||
std::fs::write(&body_path, &response_body).map_err(|source| {
|
||||
SendHttpRequestError::WriteResponseBody { path: body_path.clone(), source }
|
||||
})?;
|
||||
|
||||
response = params
|
||||
.query_manager
|
||||
.connect()
|
||||
.upsert_http_response(
|
||||
&HttpResponse {
|
||||
body_path: Some(body_path.to_string_lossy().to_string()),
|
||||
content_length: Some(usize_to_i32(response_body.len())),
|
||||
content_length_compressed: Some(u64_to_i32(body_stats.size_compressed)),
|
||||
elapsed: duration_to_i32(started_at.elapsed()),
|
||||
elapsed_headers: headers_elapsed,
|
||||
state: HttpResponseState::Closed,
|
||||
..response
|
||||
},
|
||||
¶ms.update_source,
|
||||
params.blob_manager,
|
||||
)
|
||||
.map_err(SendHttpRequestError::PersistResponse)?;
|
||||
|
||||
if let Err(join_err) = event_handle.await {
|
||||
warn!("Failed to join response event task: {}", join_err);
|
||||
}
|
||||
|
||||
Ok(SendHttpRequestResult { rendered_request, response, response_body })
|
||||
}
|
||||
|
||||
fn sendable_body_length(body: Option<&SendableBody>) -> Option<i32> {
|
||||
match body {
|
||||
Some(SendableBody::Bytes(bytes)) => Some(usize_to_i32(bytes.len())),
|
||||
Some(SendableBody::Stream { content_length: Some(length), .. }) => {
|
||||
Some(u64_to_i32(*length))
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn duration_to_i32(duration: std::time::Duration) -> i32 {
|
||||
u128_to_i32(duration.as_millis())
|
||||
}
|
||||
|
||||
fn usize_to_i32(value: usize) -> i32 {
|
||||
if value > i32::MAX as usize { i32::MAX } else { value as i32 }
|
||||
}
|
||||
|
||||
fn u64_to_i32(value: u64) -> i32 {
|
||||
if value > i32::MAX as u64 { i32::MAX } else { value as i32 }
|
||||
}
|
||||
|
||||
fn u128_to_i32(value: u128) -> i32 {
|
||||
if value > i32::MAX as u128 { i32::MAX } else { value as i32 }
|
||||
}
|
||||
Reference in New Issue
Block a user