Skip to content

Commit 3ea2ae3

Browse files
trouzetrouze
committed
support normalized host
1 parent 9e17b6f commit 3ea2ae3

4 files changed

Lines changed: 118 additions & 62 deletions

File tree

‎src/cli/commands/configure.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,12 @@ pub async fn exec(args: &ConfigureArgs) -> Result<()> {
3434
})?)
3535
};
3636

37+
let host = if !host.starts_with("https://") && !host.starts_with("http://") {
38+
format!("https://{host}")
39+
} else {
40+
host
41+
};
42+
3743
let profile = Profile {
3844
host: Some(host),
3945
token: Some(token),

‎src/cli/commands/jobs.rs‎

Lines changed: 95 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
use clap::{Args, Subcommand};
22
use serde_json::{json, Value};
33

4-
use crate::api::admin::{self, jobs};
4+
use crate::api::admin::{self, jobs, runs};
55
use crate::core::config::Config;
6-
use crate::core::error::Result;
6+
use crate::core::error::{DbtpError, Result};
77
use crate::core::rest_client::RestClient;
88

99
#[derive(Debug, Args)]
@@ -51,10 +51,30 @@ pub enum JobsCommand {
5151
git_branch: Option<String>,
5252
#[arg(long)]
5353
git_sha: Option<String>,
54+
/// Wait for the triggered run to reach a final state
55+
#[arg(long)]
56+
wait: bool,
57+
/// Polling interval in seconds (used with --wait)
58+
#[arg(long, default_value = "10")]
59+
interval: u64,
60+
/// Timeout in seconds (used with --wait)
61+
#[arg(long, default_value = "3600")]
62+
timeout: u64,
5463
},
5564
/// Rerun a job from its point of failure
5665
#[command(name = "trigger-from-failure")]
57-
TriggerFromFailure { id: u64 },
66+
TriggerFromFailure {
67+
id: u64,
68+
/// Wait for the triggered run to reach a final state
69+
#[arg(long)]
70+
wait: bool,
71+
/// Polling interval in seconds (used with --wait)
72+
#[arg(long, default_value = "10")]
73+
interval: u64,
74+
/// Timeout in seconds (used with --wait)
75+
#[arg(long, default_value = "3600")]
76+
timeout: u64,
77+
},
5878
}
5979

6080
pub async fn exec(args: &JobsArgs, client: &RestClient, config: &Config) -> Result<Value> {
@@ -107,6 +127,9 @@ pub async fn exec(args: &JobsArgs, client: &RestClient, config: &Config) -> Resu
107127
cause,
108128
git_branch,
109129
git_sha,
130+
wait,
131+
interval,
132+
timeout,
110133
} => {
111134
let mut body = json!({
112135
"cause": cause.as_deref().unwrap_or("Triggered via dbtp CLI"),
@@ -117,10 +140,76 @@ pub async fn exec(args: &JobsArgs, client: &RestClient, config: &Config) -> Resu
117140
if let Some(sha) = git_sha {
118141
body["git_sha"] = json!(sha);
119142
}
120-
jobs::trigger(client, *id, &body).await
143+
let run = jobs::trigger(client, *id, &body).await?;
144+
if *wait {
145+
let run_id = run["id"]
146+
.as_u64()
147+
.ok_or_else(|| DbtpError::config("Trigger response missing run id"))?;
148+
wait_for_run(client, run_id, *interval, *timeout).await
149+
} else {
150+
Ok(run)
151+
}
152+
}
153+
JobsCommand::TriggerFromFailure {
154+
id,
155+
wait,
156+
interval,
157+
timeout,
158+
} => {
159+
let run = jobs::trigger_from_failure(client, *id).await?;
160+
if *wait {
161+
let run_id = run["id"]
162+
.as_u64()
163+
.ok_or_else(|| DbtpError::config("Trigger response missing run id"))?;
164+
wait_for_run(client, run_id, *interval, *timeout).await
165+
} else {
166+
Ok(run)
167+
}
168+
}
169+
}
170+
}
171+
172+
const STATUS_SUCCESS: u64 = 10;
173+
const STATUS_ERROR: u64 = 20;
174+
const STATUS_CANCELLED: u64 = 30;
175+
176+
fn is_terminal_status(status: u64) -> bool {
177+
matches!(status, STATUS_SUCCESS | STATUS_ERROR | STATUS_CANCELLED)
178+
}
179+
180+
async fn wait_for_run(
181+
client: &RestClient,
182+
run_id: u64,
183+
interval: u64,
184+
timeout: u64,
185+
) -> Result<Value> {
186+
eprintln!("Waiting for run {run_id} (polling every {interval}s, timeout {timeout}s)...");
187+
let start = std::time::Instant::now();
188+
let mut last_status = String::new();
189+
190+
loop {
191+
let run = runs::get(client, run_id).await?;
192+
let status_code = run["status"].as_u64().unwrap_or(0);
193+
let status_human = run["status_humanized"]
194+
.as_str()
195+
.unwrap_or("Unknown")
196+
.to_string();
197+
198+
if status_human != last_status {
199+
eprintln!("Run {run_id}: {status_human}");
200+
last_status = status_human;
201+
}
202+
203+
if is_terminal_status(status_code) {
204+
return Ok(run);
121205
}
122-
JobsCommand::TriggerFromFailure { id } => {
123-
jobs::trigger_from_failure(client, *id).await
206+
207+
if start.elapsed().as_secs() >= timeout {
208+
return Err(DbtpError::config(format!(
209+
"Timeout waiting for run {run_id} after {timeout}s"
210+
)));
124211
}
212+
213+
tokio::time::sleep(std::time::Duration::from_secs(interval)).await;
125214
}
126215
}

‎src/cli/commands/runs.rs‎

Lines changed: 1 addition & 51 deletions
Original file line numberDiff line numberDiff line change
@@ -3,7 +3,7 @@ use serde_json::{json, Value};
33

44
use crate::api::admin::{self, runs};
55
use crate::core::config::Config;
6-
use crate::core::error::{DbtpError, Result};
6+
use crate::core::error::Result;
77
use crate::core::rest_client::RestClient;
88

99
#[derive(Debug, Args)]
@@ -29,16 +29,6 @@ pub enum RunsCommand {
2929
Cancel { id: u64 },
3030
/// Retry a failed run
3131
Retry { id: u64 },
32-
/// Wait for a run to complete
33-
Wait {
34-
id: u64,
35-
/// Polling interval in seconds
36-
#[arg(long, default_value = "10")]
37-
interval: u64,
38-
/// Timeout in seconds
39-
#[arg(long, default_value = "3600")]
40-
timeout: u64,
41-
},
4232
/// Show errors from a run (fetches run steps and extracts failures)
4333
Errors { id: u64 },
4434
}
@@ -77,53 +67,13 @@ pub async fn exec(args: &RunsArgs, client: &RestClient, config: &Config) -> Resu
7767
}
7868
RunsCommand::Cancel { id } => runs::cancel(client, *id).await,
7969
RunsCommand::Retry { id } => runs::retry(client, *id).await,
80-
RunsCommand::Wait {
81-
id,
82-
interval,
83-
timeout,
84-
} => wait_for_run(client, *id, *interval, *timeout).await,
8570
RunsCommand::Errors { id } => {
8671
let run = runs::get_with_steps(client, *id).await?;
8772
Ok(extract_run_errors(&run))
8873
}
8974
}
9075
}
9176

92-
async fn wait_for_run(
93-
client: &RestClient,
94-
run_id: u64,
95-
interval: u64,
96-
timeout: u64,
97-
) -> Result<Value> {
98-
let start = std::time::Instant::now();
99-
let mut last_status = String::new();
100-
101-
loop {
102-
let run = runs::get(client, run_id).await?;
103-
let status = run["status_humanized"]
104-
.as_str()
105-
.unwrap_or("Unknown")
106-
.to_string();
107-
108-
if status != last_status {
109-
eprintln!("Run {run_id}: {status}");
110-
last_status = status;
111-
}
112-
113-
if run["is_complete"].as_bool().unwrap_or(false) {
114-
return Ok(run);
115-
}
116-
117-
if start.elapsed().as_secs() >= timeout {
118-
return Err(DbtpError::config(format!(
119-
"Timeout waiting for run {run_id} after {timeout}s"
120-
)));
121-
}
122-
123-
tokio::time::sleep(std::time::Duration::from_secs(interval)).await;
124-
}
125-
}
126-
12777
fn extract_run_errors(run: &Value) -> Value {
12878
let mut error_steps = Vec::new();
12979

‎src/core/config.rs‎

Lines changed: 16 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -95,11 +95,13 @@ pub fn load(overrides: ConfigOverrides) -> Result<Config> {
9595

9696
let profile = file.profile.get(&profile_name).cloned().unwrap_or_default();
9797

98-
let host = overrides
99-
.host
100-
.or_else(|| std::env::var("DBTP_HOST").ok())
101-
.or(profile.host)
102-
.unwrap_or_else(|| "https://cloud.getdbt.com".into());
98+
let host = normalize_host(
99+
overrides
100+
.host
101+
.or_else(|| std::env::var("DBTP_HOST").ok())
102+
.or(profile.host)
103+
.unwrap_or_else(|| "https://cloud.getdbt.com".into()),
104+
);
103105

104106
let token = overrides
105107
.token
@@ -147,6 +149,15 @@ pub fn save_profile(name: &str, profile: &Profile) -> Result<()> {
147149
Ok(())
148150
}
149151

152+
fn normalize_host(host: String) -> String {
153+
let h = host.trim().trim_end_matches('/').to_string();
154+
if h.starts_with("https://") || h.starts_with("http://") {
155+
h
156+
} else {
157+
format!("https://{h}")
158+
}
159+
}
160+
150161
pub fn prompt(label: &str, default: &str) -> String {
151162
if default.is_empty() {
152163
eprint!("{label}: ");

0 commit comments

Comments
 (0)