noetl-executor 0.3.1

NoETL shared execution core — utilities and types shared between the noetl CLI's local-mode runner and the noetl-worker NATS pull consumer.
Documentation 
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

//! Per-execution runtime context shared between sources, dispatch, and
//! event emission.
//!
//! `ExecutionContext` is the durable carrier for one playbook execution.
//! Both the CLI's local-mode runner and the worker's NATS daemon
//! construct one of these per invocation; the dispatch layer reads from
//! it to resolve credentials, look up rendered step input, and emit
//! events into the configured sink.
//!
//! This module deliberately starts minimal — the skeleton commits the
//! shape, R-1.2 (extraction PR) fills in the concrete fields by porting
//! them from `repos/cli/src/playbook_runner.rs`.

use anyhow::Result;
use std::collections::HashMap;
use std::sync::Arc;

/// Identifies one playbook execution end-to-end.
///
/// R-1.2 PR-2a: aliased to `i64` to match the Python `noetl.event` /
/// `noetl.command` table columns (bigint), the CLI's
/// `BridgeContext.execution_id` (i64), `noetl_tools::context::
/// ExecutionContext.execution_id` (i64), and the worker's
/// `CommandNotification.execution_id` (i64).  The `pub struct
/// ExecutionId(pub String)` newtype in 0.1.x was a placeholder while
/// nothing crossed the boundary; R-1.2 starts cross-binary work so
/// the type alignment matters.
pub type ExecutionId = i64;

/// Trait for resolving keychain credential aliases at step-execution
/// time.  Mirrors the Python-side `noetl.core.credential_refs`
/// resolution flow; see noetl/noetl wiki page `credential-resolution-flow`.
#[async_trait::async_trait]
pub trait CredentialResolver: Send + Sync {
    /// Resolve `alias` (e.g. `"duffel"`) to its concrete credential
    /// envelope.  Implementations consult the NoETL keychain (worker
    /// path) or the CLI's local keychain cache (CLI path).
    async fn resolve(&self, alias: &str) -> Result<serde_json::Value>;
}

/// Carries everything one execution needs.  Cheap to clone via `Arc`s
/// on the inner heavy fields.
#[derive(Clone)]
pub struct ExecutionContext {
    pub execution_id: ExecutionId,
    pub credentials: Arc<dyn CredentialResolver>,
    /// Step-level results accumulated during execution.  Used by the
    /// template layer to render `{{ step_name.result }}` references.
    pub step_results: Arc<tokio::sync::RwLock<HashMap<String, serde_json::Value>>>,
    /// Workload payload (`workload:` block from the YAML, plus any
    /// `--set` overrides from the CLI or runtime defaults from NATS).
    pub workload: Arc<serde_json::Value>,
}

impl ExecutionContext {
    /// Construct a fresh context.  R-1.2 will add a builder that
    /// reads the workload from the YAML's `workload:` block and
    /// applies CLI overrides.
    pub fn new(
        execution_id: ExecutionId,
        credentials: Arc<dyn CredentialResolver>,
        workload: serde_json::Value,
    ) -> Self {
        Self {
            execution_id,
            credentials,
            step_results: Arc::new(tokio::sync::RwLock::new(HashMap::new())),
            workload: Arc::new(workload),
        }
    }
}