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

cargo fmt II

Browse Source
This commit is contained in:
Ruben Fiszel
2022-07-29 21:35:50 +02:00
parent 6cf4072f1d
commit 2ecec4b34c
12 changed files with 98 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/rustfmt.toml
+3
View File
@@ -1,2 +1,5 @@
max_width = 100
use_small_heuristics = "Default"
match_arm_leading_pipes="Preserve"
struct_lit_width=100
struct_variant_width=100
backend/src/flows.rs
+8 -25
View File
@@ -176,9 +176,7 @@ async fn list_flows(
}
async fn list_hub_flows(
Authed {
email, username, ..
}: Authed,
Authed { email, username, .. }: Authed,
Extension(http_client): Extension<Client>,
Host(host): Host,
) -> JsonResult<serde_json::Value> {
@@ -194,9 +192,7 @@ async fn list_hub_flows(
}
pub async fn get_hub_flow_by_id(
Authed {
email, username, ..
}: Authed,
Authed { email, username, .. }: Authed,
Path(id): Path<i32>,
Extension(http_client): Extension<Client>,
Host(host): Host,
@@ -416,17 +412,13 @@ mod tests {
let mut hm = HashMap::new();
hm.insert(
"test".to_owned(),
InputTransform::Static {
value: serde_json::json!("test2"),
},
InputTransform::Static { value: serde_json::json!("test2") },
);
let fv = FlowValue {
modules: vec![
FlowModule {
input_transform: hm,
value: FlowModuleValue::Script {
path: "test".to_string(),
},
value: FlowModuleValue::Script { path: "test".to_string() },
stop_after_if_expr: None,
skip_if_stopped: Some(false),
},
@@ -443,19 +435,12 @@ mod tests {
FlowModule {
input_transform: [(
"iterand".to_string(),
InputTransform::Static {
value: serde_json::json!(vec![1, 2, 3]),
},
InputTransform::Static { value: serde_json::json!(vec![1, 2, 3]) },
)]
.into(),
value: FlowModuleValue::ForloopFlow {
iterator: InputTransform::Static {
value: serde_json::json!([1, 2, 3]),
},
value: Box::new(FlowValue {
modules: vec![],
failure_module: None,
}),
iterator: InputTransform::Static { value: serde_json::json!([1, 2, 3]) },
value: Box::new(FlowValue { modules: vec![], failure_module: None }),
skip_failures: true,
},
stop_after_if_expr: Some("previous.res1.isEmpty()".to_string()),
@@ -464,9 +449,7 @@ mod tests {
],
failure_module: Some(FlowModule {
input_transform: HashMap::new(),
value: FlowModuleValue::Flow {
path: "test".to_string(),
},
value: FlowModuleValue::Flow { path: "test".to_string() },
stop_after_if_expr: Some("previous.res1.isEmpty()".to_string()),
skip_if_stopped: None,
}),
backend/src/groups.rs
+2 -6
View File
@@ -258,9 +258,7 @@ async fn add_user(
authed: Authed,
Extension(user_db): Extension<UserDB>,
Path((w_id, name)): Path<(String, String)>,
Json(Username {
username: user_username,
}): Json<Username>,
Json(Username { username: user_username }): Json<Username>,
) -> Result<String> {
let mut tx = user_db.begin(&authed).await?;
@@ -294,9 +292,7 @@ async fn remove_user(
authed: Authed,
Extension(user_db): Extension<UserDB>,
Path((w_id, name)): Path<(String, String)>,
Json(Username {
username: user_username,
}): Json<Username>,
Json(Username { username: user_username }): Json<Username>,
) -> Result<String> {
let mut tx = user_db.begin(&authed).await?;
backend/src/jobs.rs
+11 -43
View File
@@ -248,15 +248,10 @@ pub async fn script_path_to_payload<'c>(
w_id: &String,
) -> Result<JobPayload, Error> {
let job_payload = if script_path.starts_with("hub/") {
JobPayload::ScriptHub {
path: script_path.to_owned(),
}
JobPayload::ScriptHub { path: script_path.to_owned() }
} else {
let script_hash = get_latest_hash_for_path(db, w_id, script_path).await?;
JobPayload::ScriptHash {
hash: script_hash,
path: script_path.to_owned(),
}
JobPayload::ScriptHash { hash: script_hash, path: script_path.to_owned() }
};
Ok(job_payload)
}
@@ -296,10 +291,7 @@ pub async fn run_job_by_hash(
let (uuid, tx) = push(
tx,
&w_id,
JobPayload::ScriptHash {
hash: ScriptHash(hash),
path,
},
JobPayload::ScriptHash { hash: ScriptHash(hash), path },
args,
&authed.username,
owner_to_token_owner(&authed.username, false),
@@ -369,10 +361,7 @@ async fn run_preview_flow_job(
let (uuid, tx) = push(
tx,
&w_id,
JobPayload::RawFlow {
value: raw_flow.value,
path: raw_flow.path,
},
JobPayload::RawFlow { value: raw_flow.value, path: raw_flow.path },
raw_flow.args,
&authed.username,
owner_to_token_owner(&authed.username, false),
@@ -505,10 +494,7 @@ async fn list_jobs(
&w_id,
per_page + offset,
0,
&ListCompletedQuery {
order_desc: Some(true),
..lqc
},
&ListCompletedQuery { order_desc: Some(true), ..lqc },
&[
"'CompletedJob' as typ",
"id",
@@ -797,10 +783,7 @@ pub struct JobUpdate {
async fn get_job_update(
Extension(db): Extension<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
Query(JobUpdateQuery {
running,
log_offset,
}): Query<JobUpdateQuery>,
Query(JobUpdateQuery { running, log_offset }): Query<JobUpdateQuery>,
) -> error::JsonResult<JobUpdate> {
let mut tx = db.begin().await?;
@@ -1024,23 +1007,12 @@ struct PreviewFlow {
#[derive(Debug)]
pub enum JobPayload {
ScriptHub {
path: String,
},
ScriptHash {
hash: ScriptHash,
path: String,
},
ScriptHub { path: String },
ScriptHash { hash: ScriptHash, path: String },
Code(RawCode),
Dependencies {
hash: ScriptHash,
dependencies: Vec<String>,
},
Dependencies { hash: ScriptHash, dependencies: Vec<String> },
Flow(String),
RawFlow {
value: FlowValue,
path: Option<String>,
},
RawFlow { value: FlowValue, path: Option<String> },
}
#[instrument(level = "trace", skip_all)]
@@ -1158,11 +1130,7 @@ pub async fn push<'c>(
Some(ScriptLang::Deno),
)
}
JobPayload::Code(RawCode {
content,
path,
language,
}) => (
JobPayload::Code(RawCode { content, path, language }) => (
None,
path,
Some(content),
backend/src/oauth2.rs
+4 -17
View File
@@ -129,10 +129,7 @@ pub async fn build_oauth_clients(base_url: &str) -> anyhow::Result<AllClients> {
);
(
named_client.0,
ClientWithScopes {
client: named_client.1,
scopes: scopes.unwrap_or(vec![]),
},
ClientWithScopes { client: named_client.1, scopes: scopes.unwrap_or(vec![]) },
)
})
.collect();
@@ -151,10 +148,7 @@ pub async fn build_oauth_clients(base_url: &str) -> anyhow::Result<AllClients> {
);
(
named_client.0,
ClientWithScopes {
client: named_client.1,
scopes: scopes.unwrap_or(vec![]),
},
ClientWithScopes { client: named_client.1, scopes: scopes.unwrap_or(vec![]) },
)
})
.collect();
@@ -174,11 +168,7 @@ pub async fn build_oauth_clients(base_url: &str) -> anyhow::Result<AllClients> {
.1
});
Ok(AllClients {
logins,
connects,
slack,
})
Ok(AllClients { logins, connects, slack })
}
pub fn build_basic_client(
@@ -661,10 +651,7 @@ async fn slack_command(
let (uuid, tx) = jobs::push(
tx,
&settings.workspace_id,
JobPayload::ScriptHash {
hash: script_hash,
path: script.to_owned(),
},
JobPayload::ScriptHash { hash: script_hash, path: script.to_owned() },
Some(map),
&form.user_name,
"g/slack".to_string(),
backend/src/parser.rs
+44 -78
View File
@@ -97,21 +97,20 @@ pub fn parse_python_signature(code: &str) -> error::Result<MainArgSignature> {
Arg {
name: x.arg,
typ: x.annotation.map_or(Typ::Unknown, |e| match *e {
Located {
location: _,
node: ExpressionType::Identifier { name },
} => match name.as_ref() {
"str" => Typ::Str,
"float" => Typ::Float,
"int" => Typ::Int,
"bool" => Typ::Bool,
"dict" => Typ::Dict,
"list" => Typ::List(InnerTyp::Str),
"bytes" => Typ::Bytes,
"datetime" => Typ::Datetime,
"datetime.datetime" => Typ::Datetime,
_ => Typ::Unknown,
},
Located { location: _, node: ExpressionType::Identifier { name } } => {
match name.as_ref() {
"str" => Typ::Str,
"float" => Typ::Float,
"int" => Typ::Int,
"bool" => Typ::Bool,
"dict" => Typ::Dict,
"list" => Typ::List(InnerTyp::Str),
"bytes" => Typ::Bytes,
"datetime" => Typ::Datetime,
"datetime.datetime" => Typ::Datetime,
_ => Typ::Unknown,
}
}
_ => Typ::Unknown,
}),
has_default: default.is_some(),
@@ -161,23 +160,19 @@ pub fn parse_deno_signature(code: &str) -> error::Result<MainArgSignature> {
.body;
// println!("{ast:?}");
let params = ast.into_iter().find_map(|x| match x {
ModuleItem::ModuleDecl(ModuleDecl::ExportDecl(ExportDecl {
decl:
Decl::Fn(FnDecl {
ident:
Ident {
span: _,
sym,
optional: _,
},
declare: _,
function,
}),
span: _,
})) if &sym.to_string() == "main" => Some(function.params),
_ => None,
});
let params =
ast.into_iter().find_map(|x| match x {
ModuleItem::ModuleDecl(ModuleDecl::ExportDecl(ExportDecl {
decl:
Decl::Fn(FnDecl {
ident: Ident { span: _, sym, optional: _ },
declare: _,
function,
}),
span: _,
})) if &sym.to_string() == "main" => Some(function.params),
_ => None,
});
if let Some(params) = params {
Ok(MainArgSignature {
star_args: false,
@@ -187,19 +182,9 @@ pub fn parse_deno_signature(code: &str) -> error::Result<MainArgSignature> {
.map(|x| match x.pat {
Pat::Ident(ident) => {
let (name, typ) = binding_ident_to_arg(&ident)?;
Ok(Arg {
name,
typ,
default: None,
has_default: ident.id.optional,
})
Ok(Arg { name, typ, default: None, has_default: ident.id.optional })
}
Pat::Assign(AssignPat {
span: _,
left,
right,
type_ann: _,
}) => {
Pat::Assign(AssignPat { span: _, left, right, type_ann: _ }) => {
let (name, typ) =
left.as_ident().map(binding_ident_to_arg).ok_or_else(|| {
error::Error::ExecutionErr(format!(
@@ -254,12 +239,7 @@ fn binding_ident_to_arg(
match &**elem_type {
TsType::TsTypeRef(TsTypeRef {
span: _,
type_name:
TsEntityName::Ident(Ident {
span: _,
sym,
optional: _,
}),
type_name: TsEntityName::Ident(Ident { span: _, sym, optional: _ }),
type_params: _,
}) => match sym.to_string().as_str() {
"Base64" => Typ::List(InnerTyp::Bytes),
@@ -272,17 +252,9 @@ fn binding_ident_to_arg(
_ => Typ::List(InnerTyp::Str),
}
}
TsType::TsTypeRef(TsTypeRef {
span: _,
type_name,
type_params,
}) => {
TsType::TsTypeRef(TsTypeRef { span: _, type_name, type_params }) => {
let sym = match type_name {
TsEntityName::Ident(Ident {
span: _,
sym,
optional: _,
}) => sym,
TsEntityName::Ident(Ident { span: _, sym, optional: _ }) => sym,
TsEntityName::TsQualifiedName(p) => &*p.right.sym,
};
match sym.to_string().as_str() {
@@ -618,9 +590,7 @@ const STDIMPORTS: [&str; 301] = [
fn to_value(et: &ExpressionType) -> Option<serde_json::Value> {
match et {
ExpressionType::String {
value: StringGroup::Constant { value },
} => Some(json!(value)),
ExpressionType::String { value: StringGroup::Constant { value } } => Some(json!(value)),
ExpressionType::Number { value } => match value {
Number::Integer { value } => Some(json!(value.to_string().parse::<i64>().unwrap())),
Number::Float { value } => Some(json!(value)),
@@ -655,11 +625,9 @@ fn to_value(et: &ExpressionType) -> Option<serde_json::Value> {
}
ExpressionType::None => Some(json!(null)),
ExpressionType::Call {
function: _,
args: _,
keywords: _,
} => Some(json!("<function call>")),
ExpressionType::Call { function: _, args: _, keywords: _ } => {
Some(json!("<function call>"))
}
_ => None,
}
@@ -696,16 +664,14 @@ pub fn parse_python_imports(code: &str) -> error::Result<Vec<String>> {
.map(|x| x.symbol.split('.').next().unwrap_or("").to_string())
.collect::<Vec<String>>(),
),
StatementType::ImportFrom {
level: _,
module: Some(mod_),
names: _,
} => Some(vec![mod_
.split('.')
.next()
.unwrap_or("")
.to_string()
.replace("_", "-")]),
StatementType::ImportFrom { level: _, module: Some(mod_), names: _ } => {
Some(vec![mod_
.split('.')
.next()
.unwrap_or("")
.to_string()
.replace("_", "-")])
}
_ => None,
},
})
backend/src/scripts.rs
+2 -6
View File
@@ -253,9 +253,7 @@ async fn list_scripts(
}
async fn list_hub_scripts(
Authed {
email, username, ..
}: Authed,
Authed { email, username, .. }: Authed,
Extension(http_client): Extension<Client>,
Host(host): Host,
) -> JsonResult<serde_json::Value> {
@@ -479,9 +477,7 @@ async fn create_script(
}
pub async fn get_hub_script_by_path(
Authed {
email, username, ..
}: Authed,
Authed { email, username, .. }: Authed,
Path(path): Path<StripPath>,
Extension(http_client): Extension<Client>,
Host(host): Host,
backend/src/users.rs
+5 -19
View File
@@ -81,10 +81,7 @@ pub struct AuthCache {
impl AuthCache {
pub fn new(db: DB) -> Self {
AuthCache {
cache: Cache::new(),
db,
}
AuthCache { cache: Cache::new(), db }
}
pub async fn get_authed(&self, w_id: Option<String>, token: &str) -> Option<Authed> {
@@ -587,12 +584,7 @@ async fn logout(
async fn whoami(
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Authed {
username,
email,
is_admin,
groups,
}: Authed,
Authed { username, email, is_admin, groups }: Authed,
) -> JsonResult<UserInfo> {
let user = get_user(&w_id, &username, &db).await?;
if let Some(user) = user {
@@ -844,9 +836,7 @@ async fn add_user_to_workspace<'c>(
}
async fn update_workspace_user(
Authed {
username, is_admin, ..
}: Authed,
Authed { username, is_admin, .. }: Authed,
Extension(db): Extension<DB>,
Path((w_id, username_to_update)): Path<(String, String)>,
Json(eu): Json<EditWorkspaceUser>,
@@ -958,9 +948,7 @@ pub fn owner_to_token_owner(user: &str, is_group: bool) -> String {
}
async fn delete_user(
Authed {
username, is_admin, ..
}: Authed,
Authed { username, is_admin, .. }: Authed,
Extension(db): Extension<DB>,
Path((w_id, username_to_delete)): Path<(String, String)>,
) -> Result<String> {
@@ -1000,9 +988,7 @@ async fn delete_user(
async fn set_password(
Extension(db): Extension<DB>,
Extension(argon2): Extension<Arc<Argon2<'_>>>,
Authed {
username, email, ..
}: Authed,
Authed { username, email, .. }: Authed,
Json(EditPassword { password }): Json<EditPassword>,
) -> Result<String> {
let mut tx = db.begin().await?;
backend/src/variables.rs
+1 -3
View File
@@ -144,9 +144,7 @@ pub fn get_reserved_variables(
async fn list_contextual_variables(
Path(w_id): Path<String>,
Authed {
username, email, ..
}: Authed,
Authed { username, email, .. }: Authed,
) -> JsonResult<Vec<ContextualVariable>> {
Ok(Json(
get_reserved_variables(
backend/src/worker.rs
+1 -3
View File
@@ -354,9 +354,7 @@ async fn handle_job(
e.to_string()
))
})?;
Ok(JobResult {
result: Some(result),
})
Ok(JobResult { result: Some(result) })
} else {
let err = match status {
Ok(_) => {
backend/src/worker_flow.rs
+12 -49
View File
@@ -35,25 +35,11 @@ pub struct Iterator {
#[serde(tag = "type")]
pub enum FlowStatusModule {
WaitingForPriorSteps,
WaitingForEvent {
event: String,
},
WaitingForExecutor {
job: Uuid,
},
InProgress {
job: Uuid,
iterator: Option<Iterator>,
forloop_jobs: Option<Vec<Uuid>>,
},
Success {
job: Uuid,
forloop_jobs: Option<Vec<Uuid>>,
},
Failure {
job: Uuid,
forloop_jobs: Option<Vec<Uuid>>,
},
WaitingForEvent { event: String },
WaitingForExecutor { job: Uuid },
InProgress { job: Uuid, iterator: Option<Iterator>, forloop_jobs: Option<Vec<Uuid>> },
Success { job: Uuid, forloop_jobs: Option<Vec<Uuid>> },
Failure { job: Uuid, forloop_jobs: Option<Vec<Uuid>> },
}
#[async_recursion]
@@ -102,22 +88,13 @@ pub async fn update_flow_status_after_job_completion(
}
module_status @ _ => {
let forloop_jobs = match module_status {
FlowStatusModule::InProgress {
forloop_jobs: Some(jobs),
..
} => Some(jobs.clone()),
FlowStatusModule::InProgress { forloop_jobs: Some(jobs), .. } => Some(jobs.clone()),
_ => None,
};
let new_status = if success || (forloop_jobs.is_some() && skip_loop_failures) {
FlowStatusModule::Success {
job: job.id,
forloop_jobs,
}
FlowStatusModule::Success { job: job.id, forloop_jobs }
} else {
FlowStatusModule::Failure {
job: job.id,
forloop_jobs,
}
FlowStatusModule::Failure { job: job.id, forloop_jobs }
};
(old_status.step + 1, new_status)
}
@@ -165,10 +142,7 @@ pub async fn update_flow_status_after_job_completion(
let done = if !(success || skip_loop_failures) || last_step || stop_early {
let result = match new_status {
FlowStatusModule::Success {
forloop_jobs: Some(jobs),
..
} => {
FlowStatusModule::Success { forloop_jobs: Some(jobs), .. } => {
use futures::TryStreamExt;
let results = sqlx::query_as(
"
@@ -357,10 +331,7 @@ async fn transform_input(
("previous_result".to_string(), previous_result),
("flow_input".to_string(), flow_input),
],
Some(EvalCreds {
workspace: workspace.to_string(),
token: token.to_string(),
}),
Some(EvalCreds { workspace: workspace.to_string(), token: token.to_string() }),
steps.clone(),
)
.await
@@ -507,12 +478,7 @@ async fn push_next_flow_job(
}
}
FlowStatusModule::InProgress {
iterator:
Some(Iterator {
index,
itered,
args,
}),
iterator: Some(Iterator { index, itered, args }),
forloop_jobs: Some(forloop_jobs),
..
} if index.to_owned() + 1 < itered.len() as u8 => {
@@ -562,10 +528,7 @@ async fn push_next_flow_job(
.modules
.into_iter()
.map(|x| match x {
FlowStatusModule::Success {
job,
forloop_jobs: _,
} => job.to_string(),
FlowStatusModule::Success { job, forloop_jobs: _ } => job.to_string(),
_ => "invalid step status".to_string(),
})
.collect();
backend/src/workspaces.rs
+5 -18
View File
@@ -199,9 +199,7 @@ async fn edit_slack_command(
authed: Authed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Authed {
is_admin, username, ..
}: Authed,
Authed { is_admin, username, .. }: Authed,
Json(es): Json<EditCommandScript>,
) -> Result<String> {
require_admin(is_admin, &username)?;
@@ -380,9 +378,7 @@ async fn edit_workspace(
authed: Authed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Authed {
is_admin, username, ..
}: Authed,
Authed { is_admin, username, .. }: Authed,
Json(ew): Json<EditWorkspace>,
) -> Result<String> {
require_admin(is_admin, &username)?;
@@ -421,12 +417,7 @@ async fn edit_workspace(
async fn delete_workspace(
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Authed {
is_admin,
username,
email,
..
}: Authed,
Authed { is_admin, username, email, .. }: Authed,
) -> Result<String> {
require_admin(is_admin, &username)?;
let mut tx = db.begin().await?;
@@ -450,9 +441,7 @@ async fn delete_workspace(
}
async fn invite_user(
Authed {
username, is_admin, ..
}: Authed,
Authed { username, is_admin, .. }: Authed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Json(nu): Json<NewWorkspaceInvite>,
@@ -481,9 +470,7 @@ async fn invite_user(
}
async fn delete_invite(
Authed {
username, is_admin, ..
}: Authed,
Authed { username, is_admin, .. }: Authed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Json(nu): Json<NewWorkspaceInvite>,