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 compose;
18pub mod config;
19pub mod env_config;
20pub mod env_loader;
21pub mod error;
22pub mod executor;
23pub mod expand;
24pub mod init_template;
25pub mod interpolate;
26#[cfg(feature = "lineage")]
27pub mod lineage_glue;
28pub mod merge;
29pub mod obs;
30pub mod registry;
31#[cfg(feature = "schedule")]
32pub mod schedule;
33pub mod secrets;
34#[cfg(feature = "serve")]
35pub mod serve;
36pub mod state;
37pub mod transforms;
38
39pub use error::{CliError, CliResult};
40
41/// Convenience entry point for integration tests and custom hosts: parse a
42/// YAML config string, expand the matrix, and run all rows.
43///
44/// This skips the `install_observability` call that [`commands::run::run`]
45/// performs — callers can wire their own `metrics` recorder / tracing
46/// subscriber before calling this function (or not at all).
47pub async fn run_from_yaml_str(yaml: &str) -> CliResult<executor::RunSummary> {
48    // Parse through the same interpolate → from_text path that the binary uses,
49    // but accept a bare string instead of a file path.
50    let interpolated = interpolate::interpolate(yaml)?;
51    let mut cfg: config::PipelineConfig =
52        serde_yaml::from_str(&interpolated).map_err(|e| CliError::ParseConfig {
53            path: std::path::PathBuf::from("<yaml-string>"),
54            message: e.to_string(),
55        })?;
56    if cfg.version != 1 {
57        return Err(CliError::ParseConfig {
58            path: std::path::PathBuf::from("<yaml-string>"),
59            message: format!(
60                "unsupported pipeline version {}, only version 1 is recognised",
61                cfg.version
62            ),
63        });
64    }
65    crate::secrets::resolve_secrets(&mut cfg).await?;
66    let pipeline_name = cfg.name.clone().unwrap_or_else(|| "unnamed".to_string());
67    let auth = auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
68    let nodes = expand::expand(&cfg)?;
69    executor::run_expanded(
70        nodes,
71        executor::ExecuteOptions {
72            pipeline_name,
73            execution: cfg.execution.clone(),
74            dry_run: false,
75            limit: None,
76            state_path_override: None,
77            auth,
78            clock: chrono::Utc::now().fixed_offset(),
79            cancel: None,
80            #[cfg(feature = "lineage")]
81            lineage: None,
82            #[cfg(feature = "lineage")]
83            lineage_cfg: None,
84        },
85    )
86    .await
87}
faucet_cli/lib.rs

faucet_cli/
lib.rs
