a3s_flow/nodes/
context_get.rs

1//! `"context-get"` node — read values from the shared execution context.
2//!
3//! Reads one or more keys from [`ExecContext::context`] and returns them as a
4//! JSON object. Missing keys produce a `null` value. The returned object can be
5//! referenced by downstream nodes in the usual way (`{{ context_get_1.key }}`).
6//!
7//! # Config schema
8//!
9//! ```json
10//! { "keys": ["user_id", "attempt", "metadata"] }
11//! ```
12//!
13//! | Field | Type | Required | Description |
14//! |-------|------|:--------:|-------------|
15//! | `keys` | string[] | ✅ | List of context keys to read; missing keys resolve to `null` |
16//!
17//! # Output schema
18//!
19//! ```json
20//! { "user_id": "u_123", "attempt": 1, "metadata": null }
21//! ```
22
23use async_trait::async_trait;
24use serde_json::Value;
25
26use crate::error::{FlowError, Result};
27use crate::node::{ExecContext, Node};
28
29/// Context-get node — reads key-value pairs from the shared execution context.
30pub struct ContextGetNode;
31
32#[async_trait]
33impl Node for ContextGetNode {
34    fn node_type(&self) -> &str {
35        "context-get"
36    }
37
38    async fn execute(&self, ctx: ExecContext) -> Result<Value> {
39        let keys = ctx.data["keys"].as_array().ok_or_else(|| {
40            FlowError::InvalidDefinition("context-get: missing or non-array data.keys".into())
41        })?;
42
43        let context = ctx.context.read().unwrap();
44        let mut out = serde_json::Map::new();
45
46        for key_val in keys {
47            let key = key_val.as_str().ok_or_else(|| {
48                FlowError::InvalidDefinition(
49                    "context-get: each entry in data.keys must be a string".into(),
50                )
51            })?;
52            let value = context.get(key).cloned().unwrap_or(Value::Null);
53            out.insert(key.to_string(), value);
54        }
55
56        Ok(Value::Object(out))
57    }
58}
59
60// ── Tests ──────────────────────────────────────────────────────────────────
61
62#[cfg(test)]
63mod tests {
64    use super::*;
65    use serde_json::json;
66    use std::collections::HashMap;
67    use std::sync::{Arc, RwLock};
68
69    fn ctx_with_context(data: Value, shared: Arc<RwLock<HashMap<String, Value>>>) -> ExecContext {
70        ExecContext {
71            data,
72            context: shared,
73            ..Default::default()
74        }
75    }
76
77    #[tokio::test]
78    async fn reads_existing_keys() {
79        let shared = Arc::new(RwLock::new(HashMap::from([
80            ("user_id".into(), json!("u_42")),
81            ("count".into(), json!(7)),
82        ])));
83        let node = ContextGetNode;
84        let out = node
85            .execute(ctx_with_context(
86                json!({ "keys": ["user_id", "count"] }),
87                Arc::clone(&shared),
88            ))
89            .await
90            .unwrap();
91
92        assert_eq!(out["user_id"], json!("u_42"));
93        assert_eq!(out["count"], json!(7));
94    }
95
96    #[tokio::test]
97    async fn missing_key_returns_null() {
98        let shared = Arc::new(RwLock::new(HashMap::new()));
99        let node = ContextGetNode;
100        let out = node
101            .execute(ctx_with_context(
102                json!({ "keys": ["nonexistent"] }),
103                Arc::clone(&shared),
104            ))
105            .await
106            .unwrap();
107
108        assert_eq!(out["nonexistent"], json!(null));
109    }
110
111    #[tokio::test]
112    async fn empty_keys_returns_empty_object() {
113        let shared = Arc::new(RwLock::new(HashMap::new()));
114        let node = ContextGetNode;
115        let out = node
116            .execute(ctx_with_context(json!({ "keys": [] }), Arc::clone(&shared)))
117            .await
118            .unwrap();
119
120        assert_eq!(out, json!({}));
121    }
122
123    #[tokio::test]
124    async fn missing_keys_field_returns_error() {
125        let node = ContextGetNode;
126        let err = node
127            .execute(ExecContext {
128                data: json!({}),
129                ..Default::default()
130            })
131            .await
132            .unwrap_err();
133        assert!(matches!(err, FlowError::InvalidDefinition(_)));
134    }
135
136    #[tokio::test]
137    async fn non_string_key_entry_returns_error() {
138        let shared = Arc::new(RwLock::new(HashMap::new()));
139        let node = ContextGetNode;
140        let err = node
141            .execute(ctx_with_context(
142                json!({ "keys": [42] }),
143                Arc::clone(&shared),
144            ))
145            .await
146            .unwrap_err();
147        assert!(matches!(err, FlowError::InvalidDefinition(_)));
148    }
149}
a3s_flow/nodes/context_get.rs

a3s_flow/nodes/
context_get.rs
