faucet_cli/commands/
preview.rs1use 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#[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
34 if crate::topology::is_topology(&cfg) {
36 return crate::topology::preview(&cfg, &auth, args.limit).await;
37 }
38
39 let nodes = expand(&cfg)?;
40 let selection =
43 crate::select::RunSelection::from_args(&args.selection, cfg.selection.as_ref())?;
44 let nodes = crate::select::select_nodes(nodes, &selection, !cfg.matrix.is_empty())?;
45 let first_root = nodes
46 .iter()
47 .find(|n| matches!(n.role, NodeRole::Root))
48 .ok_or_else(|| CliError::ParseConfig {
49 path: std::path::PathBuf::from("(preview)"),
50 message: "no root rows in matrix to preview".to_owned(),
51 })?;
52 tracing::info!(row = %first_root.id, "previewing first root row");
53
54 let source = build_source(
55 &first_root.source.kind,
56 first_root.source.config.clone(),
57 &auth,
58 None,
59 )
60 .await?;
61 let stages = compile_transforms(&first_root.transforms)?;
62 let records = source.fetch_all().await?;
63 let records: Vec<_> = if stages.is_empty() {
64 records
65 } else {
66 let compiled = stages
67 .iter()
68 .map(compile_stage)
69 .collect::<Result<Vec<_>, _>>()?;
70 let mut out = Vec::with_capacity(records.len());
71 for r in records {
72 out.extend(apply_stages(r, &compiled)?);
73 }
74 out
75 };
76
77 let limited: Vec<_> = records.into_iter().take(args.limit).collect();
78 let sink = StdoutSink::new(
79 StdoutSinkConfig::new()
80 .format(StdoutFormat::JsonLines)
81 .flush_per_record(true),
82 );
83 sink.write_batch(&limited).await?;
84 sink.flush().await?;
85
86 let _ = std::marker::PhantomData::<Pipeline<'_, dyn faucet_core::Source, dyn Sink>>;
87 Ok(())
88}
89
90#[cfg(not(feature = "sink-stdout"))]
91pub async fn run(_args: PreviewArgs) -> CliResult<()> {
92 Err(CliError::UnknownConnector {
93 kind: "sink",
94 name: "stdout".into(),
95 available: "(preview requires faucet-cli to be built with the 'sink-stdout' feature)"
96 .into(),
97 })
98}