diff --git a/Cargo.lock b/Cargo.lock index d86231b8..c127be68 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -524,7 +524,7 @@ dependencies = [ [[package]] name = "config" -version = "0.3.62" +version = "0.3.63" dependencies = [ "base64", "chrono", @@ -631,7 +631,7 @@ checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" [[package]] name = "crypto" -version = "0.3.62" +version = "0.3.63" dependencies = [ "aes-gcm", "base64", @@ -3331,7 +3331,7 @@ dependencies = [ [[package]] name = "testutils" -version = "0.3.62" +version = "0.3.63" dependencies = [ "pem", "rsa", @@ -3586,7 +3586,7 @@ checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" [[package]] name = "tower" -version = "0.3.62" +version = "0.3.63" dependencies = [ "config", "pyo3", @@ -3614,7 +3614,7 @@ dependencies = [ [[package]] name = "tower-api" -version = "0.3.62" +version = "0.3.63" dependencies = [ "reqwest", "serde", @@ -3626,7 +3626,7 @@ dependencies = [ [[package]] name = "tower-cmd" -version = "0.3.62" +version = "0.3.63" dependencies = [ "axum", "bytes", @@ -3697,7 +3697,7 @@ checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" [[package]] name = "tower-package" -version = "0.3.62" +version = "0.3.63" dependencies = [ "async-compression", "flate2", @@ -3721,7 +3721,7 @@ dependencies = [ [[package]] name = "tower-runtime" -version = "0.3.62" +version = "0.3.63" dependencies = [ "async-trait", "chrono", @@ -3745,7 +3745,7 @@ checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" [[package]] name = "tower-telemetry" -version = "0.3.62" +version = "0.3.63" dependencies = [ "tracing", "tracing-appender", @@ -3754,7 +3754,7 @@ dependencies = [ [[package]] name = "tower-uv" -version = "0.3.62" +version = "0.3.63" dependencies = [ "async-compression", "async_zip", @@ -3774,7 +3774,7 @@ dependencies = [ [[package]] name = "tower-version" -version = "0.3.62" +version = "0.3.63" dependencies = [ "anyhow", "chrono", diff --git a/Cargo.toml b/Cargo.toml index cddd184b..fa80245f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,7 +4,7 @@ resolver = "2" [workspace.package] edition = "2021" -version = "0.3.62" +version = "0.3.63" description = "Tower is the best way to host Python data apps in production" rust-version = "1.81" authors = ["Brad Heller ", "Ben Lovell "] diff --git a/crates/tower-cmd/src/api.rs b/crates/tower-cmd/src/api.rs index 07f0dca8..54454477 100644 --- a/crates/tower-cmd/src/api.rs +++ b/crates/tower-cmd/src/api.rs @@ -30,38 +30,62 @@ trait PaginatedResponse { impl PaginatedResponse for tower_api::models::ListAppsResponse { type Item = tower_api::models::AppSummary; - fn pagination(&self) -> &Pagination { &self.pages } - fn into_items(self) -> Vec { self.apps } + fn pagination(&self) -> &Pagination { + &self.pages + } + fn into_items(self) -> Vec { + self.apps + } } impl PaginatedResponse for tower_api::models::ListTeamsResponse { type Item = tower_api::models::Team; - fn pagination(&self) -> &Pagination { &self.pages } - fn into_items(self) -> Vec { self.teams } + fn pagination(&self) -> &Pagination { + &self.pages + } + fn into_items(self) -> Vec { + self.teams + } } impl PaginatedResponse for tower_api::models::ListSecretsResponse { type Item = tower_api::models::Secret; - fn pagination(&self) -> &Pagination { &self.pages } - fn into_items(self) -> Vec { self.secrets } + fn pagination(&self) -> &Pagination { + &self.pages + } + fn into_items(self) -> Vec { + self.secrets + } } impl PaginatedResponse for tower_api::models::ListCatalogsResponse { type Item = tower_api::models::Catalog; - fn pagination(&self) -> &Pagination { &self.pages } - fn into_items(self) -> Vec { self.catalogs } + fn pagination(&self) -> &Pagination { + &self.pages + } + fn into_items(self) -> Vec { + self.catalogs + } } impl PaginatedResponse for tower_api::models::ListEnvironmentsResponse { type Item = tower_api::models::Environment; - fn pagination(&self) -> &Pagination { &self.pages } - fn into_items(self) -> Vec { self.environments } + fn pagination(&self) -> &Pagination { + &self.pages + } + fn into_items(self) -> Vec { + self.environments + } } impl PaginatedResponse for tower_api::models::ListSchedulesResponse { type Item = tower_api::models::Schedule; - fn pagination(&self) -> &Pagination { &self.pages } - fn into_items(self) -> Vec { self.schedules } + fn pagination(&self) -> &Pagination { + &self.pages + } + fn into_items(self) -> Vec { + self.schedules + } } /// Fetches pages from a paginated API endpoint, honoring the caller's @@ -375,8 +399,7 @@ pub async fn list_secrets( config: &Config, env: &str, all: bool, -) -> Result, Error> -{ +) -> Result, Error> { let api_config: configuration::Configuration = config.into(); let env = env.to_string(); @@ -650,8 +673,16 @@ pub async fn stream_run_logs( builder = builder.header(reqwest::header::USER_AGENT, user_agent.clone()); } + // Mirrors the generated tower-api client: prefer a bearer token (interactive session), + // otherwise fall back to the API key header set when TOWER_API_KEY is configured. if let Some(ref token) = api_config.bearer_access_token { builder = builder.bearer_auth(token.to_owned()); + } else if let Some(ref apikey) = api_config.api_key { + let value = match &apikey.prefix { + Some(prefix) => format!("{} {}", prefix, apikey.key), + None => apikey.key.clone(), + }; + builder = builder.header("X-API-Key", value); }; // Now let's try to open the event source with the server. diff --git a/crates/tower-cmd/src/apps.rs b/crates/tower-cmd/src/apps.rs index e04c3dc9..5fc707c1 100644 --- a/crates/tower-cmd/src/apps.rs +++ b/crates/tower-cmd/src/apps.rs @@ -833,9 +833,7 @@ mod tests { #[test] fn list_defaults_to_no_environment_filter() { - let matches = apps_cmd() - .try_get_matches_from(["apps", "list"]) - .unwrap(); + let matches = apps_cmd().try_get_matches_from(["apps", "list"]).unwrap(); let (_, list_args) = matches.subcommand().unwrap(); assert_eq!(list_args.get_one::("environment"), None); @@ -849,7 +847,9 @@ mod tests { let (_, list_args) = matches.subcommand().unwrap(); assert_eq!( - list_args.get_one::("environment").map(|s| s.as_str()), + list_args + .get_one::("environment") + .map(|s| s.as_str()), Some("production") ); } diff --git a/crates/tower-cmd/src/run.rs b/crates/tower-cmd/src/run.rs index be8050e6..3a3cfa27 100644 --- a/crates/tower-cmd/src/run.rs +++ b/crates/tower-cmd/src/run.rs @@ -713,9 +713,10 @@ async fn monitor_cli_status( ); match handle.lock().await.status().await { - Ok(status) => { + Ok(exec_status) => { // We reset the error count to indicate that we can intermittently get statuses. err_count = 0; + let status = exec_status.status; match status { Status::Exited => { diff --git a/crates/tower-cmd/src/teams.rs b/crates/tower-cmd/src/teams.rs index 36783a9e..59ae9636 100644 --- a/crates/tower-cmd/src/teams.rs +++ b/crates/tower-cmd/src/teams.rs @@ -60,10 +60,7 @@ async fn do_list_via_api(config: &Config) { let headers = vec!["Name".to_string()]; - let teams_data: Vec> = teams - .iter() - .map(|team| vec![team.name.clone()]) - .collect(); + let teams_data: Vec> = teams.iter().map(|team| vec![team.name.clone()]).collect(); output::newline(); output::table(headers, teams_data, None::<&Vec>); diff --git a/crates/tower-cmd/src/util/deploy.rs b/crates/tower-cmd/src/util/deploy.rs index 23c56a3c..cc19fcb9 100644 --- a/crates/tower-cmd/src/util/deploy.rs +++ b/crates/tower-cmd/src/util/deploy.rs @@ -57,9 +57,17 @@ pub async fn upload_file_with_progress( .header("Content-Encoding", "gzip") .body(Body::wrap_stream(progress_stream)); - // Add authorization if available + // Add authorization if available. Mirrors the generated tower-api client: prefer a + // bearer token (interactive session), otherwise fall back to the API key header set + // when TOWER_API_KEY is configured. if let Some(token) = &api_config.bearer_access_token { req = req.header("Authorization", format!("Bearer {}", token)); + } else if let Some(apikey) = &api_config.api_key { + let value = match &apikey.prefix { + Some(prefix) => format!("{} {}", prefix, apikey.key), + None => apikey.key.clone(), + }; + req = req.header("X-API-Key", value); } // Send the request diff --git a/crates/tower-package/src/towerfile.rs b/crates/tower-package/src/towerfile.rs index f86b2d53..1d18aba4 100644 --- a/crates/tower-package/src/towerfile.rs +++ b/crates/tower-package/src/towerfile.rs @@ -76,9 +76,28 @@ impl Towerfile { return Err(Error::MissingRequiredAppField { field: "name".to_string(), }); - } else { - Ok(towerfile) } + + for import_path in &towerfile.app.import_paths { + let as_str = import_path.to_string_lossy(); + if as_str.is_empty() { + return Err(Error::InvalidTowerfile { + message: "import_paths entries must not be empty".to_string(), + }); + } + // PATH-style separators in a single entry would break PYTHONPATH construction at + // runtime, since each entry is joined with the platform path separator. + if as_str.contains(|c: char| c == ':' || c == ';') { + return Err(Error::InvalidTowerfile { + message: format!( + "import_paths entry {:?} contains an illegal character (':' or ';')", + as_str + ), + }); + } + } + + Ok(towerfile) } /// set_parameter upserts a parameter by lookup name. If a parameter with the given name diff --git a/crates/tower-runtime/src/execution.rs b/crates/tower-runtime/src/execution.rs index 47d6dd26..f69e51f9 100644 --- a/crates/tower-runtime/src/execution.rs +++ b/crates/tower-runtime/src/execution.rs @@ -5,6 +5,7 @@ //! Kubernetes pods, etc.) through a uniform interface. use async_trait::async_trait; +use chrono::{DateTime, Utc}; use std::collections::HashMap; use std::path::PathBuf; use tokio::io::AsyncRead; @@ -198,14 +199,37 @@ pub struct BackendCapabilities { // Execution Handle Trait // ============================================================================ +/// Result of querying execution status, including optional timing and +/// backend-specific metadata. The metadata map allows backends to surface +/// arbitrary key-value data (e.g. node_type, scheduling_latency_ms) without +/// requiring trait changes. +#[derive(Clone, Debug)] +pub struct ExecutionStatus { + pub status: Status, + pub started_at: Option>, + pub ended_at: Option>, + pub metadata: HashMap, +} + +impl From for ExecutionStatus { + fn from(status: Status) -> Self { + Self { + status, + started_at: None, + ended_at: None, + metadata: HashMap::new(), + } + } +} + /// ExecutionHandle represents a running execution #[async_trait] pub trait ExecutionHandle: Send + Sync { /// Get a unique identifier for this execution fn id(&self) -> &str; - /// Get current execution status - async fn status(&self) -> Result; + /// Get current execution status with optional timing metadata + async fn status(&self) -> Result; /// Subscribe to log stream async fn logs(&self) -> Result; diff --git a/crates/tower-runtime/src/lib.rs b/crates/tower-runtime/src/lib.rs index 446264fe..fffb9447 100644 --- a/crates/tower-runtime/src/lib.rs +++ b/crates/tower-runtime/src/lib.rs @@ -83,10 +83,7 @@ impl Status { pub fn is_terminal(&self) -> bool { matches!( self, - Status::Exited - | Status::Crashed { .. } - | Status::Cancelled - | Status::Failed(_) + Status::Exited | Status::Crashed { .. } | Status::Cancelled | Status::Failed(_) ) } } diff --git a/crates/tower-runtime/src/local.rs b/crates/tower-runtime/src/local.rs index 07b368cf..cafc2bc7 100644 --- a/crates/tower-runtime/src/local.rs +++ b/crates/tower-runtime/src/local.rs @@ -161,34 +161,17 @@ async fn inner_execute_local_app( let manifest = &package.manifest; let secrets = opts.secrets; let params = opts.parameters; - let mut other_env_vars = opts.env_vars; - - if !package.manifest.import_paths.is_empty() { - debug!(ctx: &ctx, "adding import paths to PYTHONPATH: {:?}", package.manifest.import_paths); - - let import_paths = package - .manifest - .import_paths - .iter() - .map(|p| package_path.join(p)) - .collect::>(); - - let import_paths = std::env::join_paths(import_paths)? - .to_string_lossy() - .to_string(); - - if other_env_vars.contains_key("PYTHONPATH") { - // If we already have a PYTHONPATH, we need to append to it. - let existing = other_env_vars.get("PYTHONPATH").unwrap(); - let pythonpath = std::env::join_paths(vec![existing, &import_paths])? - .to_string_lossy() - .to_string(); - - other_env_vars.insert("PYTHONPATH".to_string(), pythonpath); - } else { - // Otherwise, we just set it. - other_env_vars.insert("PYTHONPATH".to_string(), import_paths); - } + let other_env_vars = opts.env_vars; + + let import_paths: Vec = package + .manifest + .import_paths + .iter() + .map(|p| package_path.join(p)) + .collect(); + + if !import_paths.is_empty() { + debug!(ctx: &ctx, "adding import paths to PYTHONPATH: {:?}", import_paths); } // We insert these checks for cancellation along the way to see if the process was @@ -210,6 +193,7 @@ async fn inner_execute_local_app( secrets, params, other_env_vars, + &import_paths, ) .await?; @@ -233,6 +217,7 @@ async fn inner_execute_local_app( &secrets, ¶ms, &other_env_vars, + &import_paths, ); // Now we also need to find the program to execute. @@ -303,7 +288,8 @@ async fn inner_execute_local_app( } } Ok(child) => { - let mut res = run_setup_child(&ctx, &cancel_token, &opts.output_sender, child).await; + let mut res = + run_setup_child(&ctx, &cancel_token, &opts.output_sender, child).await; // If sync was cancelled, don't bother retrying — bail out // cleanly so the receiver sees `Status::Cancelled` instead of @@ -328,7 +314,8 @@ async fn inner_execute_local_app( let retry_child = uv .sync_with_legacy_setuptools_pin(&working_dir, &env_vars) .await?; - res = run_setup_child(&ctx, &cancel_token, &opts.output_sender, retry_child).await; + res = run_setup_child(&ctx, &cancel_token, &opts.output_sender, retry_child) + .await; if cancel_token.is_cancelled() { return Err(Error::Cancelled); } @@ -419,7 +406,9 @@ impl App for LocalApp { Ok(Ok(code)) => AppCompletion::Exit(code), Ok(Err(Error::Cancelled)) => AppCompletion::Cancelled, Ok(Err(e)) => AppCompletion::Failed(AppFailure::Runtime(e)), - Err(panic) => AppCompletion::Failed(AppFailure::Panic(panic_payload_message(&panic))), + Err(panic) => { + AppCompletion::Failed(AppFailure::Panic(panic_payload_message(&panic))) + } }; let _ = sx.send(completion); }); @@ -493,6 +482,7 @@ async fn execute_bash_program( secrets: HashMap, params: HashMap, other_env_vars: HashMap, + import_paths: &[PathBuf], ) -> Result { let bash_path = find_bash().await?; debug!(ctx: &ctx, "using bash at {:?}", bash_path); @@ -512,6 +502,7 @@ async fn execute_bash_program( &secrets, ¶ms, &other_env_vars, + import_paths, )) .kill_on_drop(true) .spawn()?; @@ -538,6 +529,7 @@ fn make_env_vars( secs: &HashMap, params: &HashMap, other_env_vars: &HashMap, + import_paths: &[PathBuf], ) -> HashMap { let mut res = HashMap::new(); @@ -558,19 +550,20 @@ fn make_env_vars( let added_keys = res.keys().map(|s| &**s).collect::>().join(", "); debug!(ctx: &ctx, "added keys {}", &added_keys); - // We also need a PYTHONPATH that is set to the current working directory to help with the - // dependency resolution problem at runtime. - let pythonpath = cwd.to_string_lossy().to_string(); - let pythonpath = if res.contains_key("PYTHONPATH") { - // If we already have a PYTHONPATH, we need to append to it. - let existing = res.get("PYTHONPATH").unwrap(); - let joined_paths = std::env::join_paths([existing, &pythonpath]).unwrap(); - joined_paths.to_string_lossy().to_string() - } else { - // There was no previously set PYTHONPATH, so we just include our current directory. - pythonpath - }; - + let mut path_entries: Vec = res + .remove("PYTHONPATH") + .as_deref() + .map(env::split_paths) + .into_iter() + .flatten() + .collect(); + path_entries.extend(import_paths.iter().cloned()); + path_entries.push(cwd.clone()); + + let pythonpath = env::join_paths(&path_entries) + .unwrap() + .to_string_lossy() + .to_string(); res.insert("PYTHONPATH".to_string(), pythonpath); // Inject a TOWER_ENVIRONMENT parameter so you know what environment you're running in. Empty diff --git a/crates/tower-runtime/src/subprocess.rs b/crates/tower-runtime/src/subprocess.rs index decdb1b0..23c7d4ef 100644 --- a/crates/tower-runtime/src/subprocess.rs +++ b/crates/tower-runtime/src/subprocess.rs @@ -4,7 +4,7 @@ use crate::auto_cleanup; use crate::errors::Error; use crate::execution::{ BackendCapabilities, CacheBackend, ExecutionBackend, ExecutionHandle, ExecutionSpec, - ServiceEndpoint, + ExecutionStatus, ServiceEndpoint, }; use crate::local::LocalApp; use crate::{App, OutputReceiver, StartOptions, Status}; @@ -212,9 +212,9 @@ impl ExecutionHandle for SubprocessHandle { &self.id } - async fn status(&self) -> Result { + async fn status(&self) -> Result { let app = self.app.lock().await; - app.status().await + Ok(app.status().await?.into()) } async fn logs(&self) -> Result { @@ -247,7 +247,7 @@ impl ExecutionHandle for SubprocessHandle { async fn wait_for_completion(&self) -> Result { loop { - let status = self.status().await?; + let status = self.status().await?.status; match status { Status::None | Status::Running => { tokio::time::sleep(Duration::from_millis(100)).await; diff --git a/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/Towerfile b/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/Towerfile new file mode 100644 index 00000000..d9698aa7 --- /dev/null +++ b/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/Towerfile @@ -0,0 +1,7 @@ +[app] +name = "06-multiple-import-paths" +script = "./task.py" +import_paths = [ + "./lib_a", + "./lib_b", +] diff --git a/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/lib_a/mod_a.py b/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/lib_a/mod_a.py new file mode 100644 index 00000000..86016b51 --- /dev/null +++ b/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/lib_a/mod_a.py @@ -0,0 +1,2 @@ +def greet(): + print("hello from mod_a") diff --git a/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/lib_b/mod_b.py b/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/lib_b/mod_b.py new file mode 100644 index 00000000..b6807b0d --- /dev/null +++ b/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/lib_b/mod_b.py @@ -0,0 +1,2 @@ +def greet(): + print("hello from mod_b") diff --git a/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/task.py b/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/task.py new file mode 100644 index 00000000..c23a71f0 --- /dev/null +++ b/crates/tower-runtime/tests/example-apps/06-multiple-import-paths/task.py @@ -0,0 +1,5 @@ +import mod_a +import mod_b + +mod_a.greet() +mod_b.greet() diff --git a/crates/tower-runtime/tests/local_test.rs b/crates/tower-runtime/tests/local_test.rs index b0ab679a..fdd634cf 100644 --- a/crates/tower-runtime/tests/local_test.rs +++ b/crates/tower-runtime/tests/local_test.rs @@ -381,3 +381,53 @@ async fn test_abort_on_dependency_installation_failure() { } } } + +// Regression test: when a Towerfile declares multiple `import_paths`, the runtime used to panic +// while building PYTHONPATH because the already-joined value was fed back into +// `std::env::join_paths`, which rejects components containing the path separator. +#[tokio::test] +async fn test_running_app_with_multiple_import_paths() { + debug!("Running 06-multiple-import-paths"); + let dir = get_example_app_dir("06-multiple-import-paths"); + let package = build_package_from_dir(&dir).await; + let (sender, mut receiver) = unbounded_channel(); + + let opts = StartOptions { + ctx: tower_telemetry::Context::new("runner-id".to_string()), + package, + output_sender: sender, + cwd: None, + environment: "local".to_string(), + secrets: HashMap::new(), + parameters: HashMap::new(), + env_vars: HashMap::new(), + cache_dir: Some(config::default_cache_dir()), + }; + + let app = LocalApp::start(opts) + .await + .expect("Failed to start app with multiple import_paths"); + + let mut outputs = Vec::new(); + while let Some(output) = receiver.recv().await { + outputs.push(output.line); + } + + assert!( + outputs.iter().any(|line| line.contains("hello from mod_a")), + "expected output from lib_a/mod_a; got {:?}", + outputs + ); + assert!( + outputs.iter().any(|line| line.contains("hello from mod_b")), + "expected output from lib_b/mod_b; got {:?}", + outputs + ); + + let status = app.status().await.expect("Failed to get app status"); + assert!( + status == Status::Exited, + "App should have exited cleanly, got {:?}", + status + ); +} diff --git a/pyproject.toml b/pyproject.toml index a6eb82fa..07c007d1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "maturin" [project] name = "tower" -version = "0.3.62" +version = "0.3.63" description = "Tower CLI and runtime environment for Tower." authors = [{ name = "Tower Computing Inc.", email = "brad@tower.dev" }] readme = "README.md" diff --git a/uv.lock b/uv.lock index 05fcc6ee..9e390c39 100644 --- a/uv.lock +++ b/uv.lock @@ -2642,7 +2642,7 @@ wheels = [ [[package]] name = "tower" -version = "0.3.62" +version = "0.3.63" source = { editable = "." } dependencies = [ { name = "attrs" },