faucet_cli/commands/
preview.rs

1//! `faucet preview` — run only the source side of the first root row and emit
2//! the first N records to stdout as JSON Lines.
3//!
4//! Child rows can't be previewed in isolation in v1: they need parent records
5//! to resolve `${parent.path}` tokens. Preview the parent first, then point
6//! the child at a `${file:...}` fixture if you need to drive it standalone.
7
8use crate::cli::PreviewArgs;
9use crate::config::PipelineConfig;
10use crate::error::{CliError, CliResult};
11use crate::expand::{NodeRole, expand};
12use crate::registry::build_source;
13use crate::transforms::compile_transforms;
14use faucet_core::stage::{apply_stages, compile_stage};
15
16#[cfg(feature = "sink-stdout")]
17use faucet_core::{Pipeline, Sink};
18
19/// Execute the `preview` subcommand.
20#[cfg(feature = "sink-stdout")]
21pub async fn run(args: PreviewArgs) -> CliResult<()> {
22    use faucet_sink_stdout::{StdoutFormat, StdoutSink, StdoutSinkConfig};
23    let cwd = std::env::current_dir()?;
24    let env_path =
25        crate::env_loader::resolve_env_file(args.env_file.as_deref(), args.no_env_file, &cwd)?;
26    crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
27    let path = match args.config {
28        Some(p) => p,
29        None => crate::env_loader::discover_config_path(&cwd).ok_or(CliError::NoConfigOrFromEnv)?,
30    };
31    let cfg = PipelineConfig::from_path_async(&path, args.profile.as_deref()).await?;
32    let auth = crate::auth_catalog::build_auth_catalog(cfg.auth.as_ref())?;
33    let nodes = expand(&cfg)?;
34    let first_root = nodes
35        .iter()
36        .find(|n| matches!(n.role, NodeRole::Root))
37        .ok_or_else(|| CliError::ParseConfig {
38            path: std::path::PathBuf::from("(preview)"),
39            message: "no root rows in matrix to preview".to_owned(),
40        })?;
41    tracing::info!(row = %first_root.id, "previewing first root row");
42
43    let source = build_source(
44        &first_root.source.kind,
45        first_root.source.config.clone(),
46        &auth,
47        None,
48    )
49    .await?;
50    let stages = compile_transforms(&first_root.transforms)?;
51    let records = source.fetch_all().await?;
52    let records: Vec<_> = if stages.is_empty() {
53        records
54    } else {
55        let compiled = stages
56            .iter()
57            .map(compile_stage)
58            .collect::<Result<Vec<_>, _>>()?;
59        let mut out = Vec::with_capacity(records.len());
60        for r in records {
61            out.extend(apply_stages(r, &compiled)?);
62        }
63        out
64    };
65
66    let limited: Vec<_> = records.into_iter().take(args.limit).collect();
67    let sink = StdoutSink::new(
68        StdoutSinkConfig::new()
69            .format(StdoutFormat::JsonLines)
70            .flush_per_record(true),
71    );
72    sink.write_batch(&limited).await?;
73    sink.flush().await?;
74
75    let _ = std::marker::PhantomData::<Pipeline<'_, dyn faucet_core::Source, dyn Sink>>;
76    Ok(())
77}
78
79#[cfg(not(feature = "sink-stdout"))]
80pub async fn run(_args: PreviewArgs) -> CliResult<()> {
81    Err(CliError::UnknownConnector {
82        kind: "sink",
83        name: "stdout".into(),
84        available: "(preview requires faucet-cli to be built with the 'sink-stdout' feature)"
85            .into(),
86    })
87}
faucet_cli/commands/preview.rs

faucet_cli/commands/
preview.rs
