faucet_cli/
lib.rs

1#![cfg_attr(docsrs, feature(doc_cfg))]
2
3//! # faucet-cli
4//!
5//! A config-driven runner for [`faucet-stream`](https://docs.rs/faucet-stream)
6//! pipelines.  Define a source, optional transforms, a sink, and (optionally)
7//! a state store in a YAML or JSON file, then run it with the `faucet` binary —
8//! no Rust code required.
9//!
10//! The library half of this crate exposes the same building blocks the binary
11//! uses (config parsing, env interpolation, the connector registry) so that
12//! integrations and tests can reuse them.
13
14pub mod auth_catalog;
15pub mod cli;
16pub mod commands;
17pub mod config;
18pub mod env_config;
19pub mod env_loader;
20pub mod error;
21pub mod executor;
22pub mod expand;
23pub mod init_template;
24pub mod interpolate;
25pub mod merge;
26pub mod obs;
27pub mod registry;
28#[cfg(feature = "schedule")]
29pub mod schedule;
30pub mod secrets;
31#[cfg(feature = "serve")]
32pub mod serve;
33pub mod state;
34pub mod transforms;
35
36pub use error::{CliError, CliResult};
37
38/// Convenience entry point for integration tests and custom hosts: parse a
39/// YAML config string, expand the matrix, and run all rows.
40///
41/// This skips the `install_observability` call that [`commands::run::run`]
42/// performs — callers can wire their own `metrics` recorder / tracing
43/// subscriber before calling this function (or not at all).
44pub async fn run_from_yaml_str(yaml: &str) -> CliResult<executor::RunSummary> {
45    // Parse through the same interpolate → from_text path that the binary uses,
46    // but accept a bare string instead of a file path.
47    let interpolated = interpolate::interpolate(yaml)?;
48    let mut cfg: config::PipelineConfig =
49        serde_yaml::from_str(&interpolated).map_err(|e| CliError::ParseConfig {
50            path: std::path::PathBuf::from("<yaml-string>"),
51            message: e.to_string(),
52        })?;
53    if cfg.version != 1 {
54        return Err(CliError::ParseConfig {
55            path: std::path::PathBuf::from("<yaml-string>"),
56            message: format!(
57                "unsupported pipeline version {}, only version 1 is recognised",
58                cfg.version
59            ),
60        });
61    }
62    crate::secrets::resolve_secrets(&mut cfg).await?;
63    let pipeline_name = cfg.name.clone().unwrap_or_else(|| "unnamed".to_string());
64    let auth = auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
65    let nodes = expand::expand(&cfg)?;
66    executor::run_expanded(
67        nodes,
68        executor::ExecuteOptions {
69            pipeline_name,
70            execution: cfg.execution.clone(),
71            dry_run: false,
72            limit: None,
73            state_path_override: None,
74            auth,
75            clock: chrono::Utc::now().fixed_offset(),
76            cancel: None,
77        },
78    )
79    .await
80}
faucet_cli/lib.rs

faucet_cli/
lib.rs
