faucet-cli 1.13.1

Config-driven CLI runner for faucet-stream pipelines (YAML / JSON, Meltano-style)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

//! `GET /v1/schemas` (catalog) and `GET /v1/schemas/{kind}/{name}` (one JSON
//! Schema). Read-only; reuses the CLI's schema introspection so the catalog
//! always reflects the compiled connectors/transforms.

use crate::serve::error::ServeError;
use crate::serve::state::ServerState;
use axum::Json;
use axum::extract::{Path, State};
use serde::Serialize;
use serde_json::Value;

#[derive(Debug, Serialize)]
pub struct SchemaItem {
    pub name: String,
    pub description: String,
}

#[derive(Debug, Serialize)]
pub struct SchemasResponse {
    pub sources: Vec<SchemaItem>,
    pub sinks: Vec<SchemaItem>,
    pub transforms: Vec<SchemaItem>,
    pub state: Vec<String>,
    /// Config blocks a submitted pipeline can add beside its source, sink, and
    /// transforms; fetch each one's schema from `/v1/schemas/block/{name}`.
    pub blocks: Vec<crate::commands::schema::PipelineBlock>,
}

fn items(pairs: Vec<(&'static str, &'static str)>) -> Vec<SchemaItem> {
    pairs
        .into_iter()
        .map(|(name, description)| SchemaItem {
            name: name.to_string(),
            description: description.to_string(),
        })
        .collect()
}

/// `GET /v1/schemas` → catalog of compiled connectors/transforms + state kinds.
pub async fn list_schemas(State(_state): State<ServerState>) -> Json<SchemasResponse> {
    Json(SchemasResponse {
        sources: items(crate::registry::source_descriptions()),
        sinks: items(crate::registry::sink_descriptions()),
        transforms: items(crate::transforms::transform_descriptions()),
        state: crate::state::available_state_kinds()
            .into_iter()
            .map(String::from)
            .collect(),
        blocks: crate::commands::schema::pipeline_blocks(),
    })
}

/// `GET /v1/schemas/{kind}/{name}` → the JSON Schema for one connector,
/// transform, or pipeline block (`kind` = `source` | `sink` | `transform` | `block`).
/// Unknown kind or name → 404.
pub async fn get_schema(
    State(_state): State<ServerState>,
    Path((kind, name)): Path<(String, String)>,
) -> Result<Json<Value>, ServeError> {
    let schema = match kind.as_str() {
        "source" => crate::registry::source_schema(&name),
        "sink" => crate::registry::sink_schema(&name),
        "transform" => crate::transforms::transform_schema(&name),
        "block" => {
            return crate::commands::schema::block_schema(&name)
                .map(Json)
                .ok_or(ServeError::NotFound);
        }
        _ => return Err(ServeError::NotFound),
    }
    .map_err(|_| ServeError::NotFound)?;
    Ok(Json(schema))
}