ironflow_ops_postgres/
query.rs

1//! Query operations: `SELECT` statements returning rows or scalar values.
2
3use async_trait::async_trait;
4use ironflow_core::error::OperationError;
5use ironflow_core::operation::{Operation, OperationContext, TypedOperation};
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8use sqlx::{PgPool, Row as _};
9
10use crate::helpers::{bind_json_param, column_to_json, pg_error, row_to_json, to_value};
11
12/// Output of [`QueryRows`].
13#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct QueryRowsOutput {
15    /// The rows as a JSON array. Each row is a JSON object with column
16    /// names as keys.
17    pub rows: Vec<Value>,
18}
19
20/// Execute a `SELECT` query and return all matching rows as JSON.
21///
22/// Parameters are passed as a JSON array and bound positionally (`$1`, `$2`, ...).
23///
24/// # Examples
25///
26/// ```no_run
27/// use ironflow_ops_postgres::query::QueryRows;
28/// use ironflow_core::operation::Operation;
29///
30/// # fn example(pool: sqlx::PgPool) {
31/// let op = QueryRows::new(pool, "SELECT id, name FROM users WHERE active = $1", vec![serde_json::json!(true)]);
32/// assert_eq!(op.kind(), "postgres");
33/// # }
34/// ```
35pub struct QueryRows {
36    pool: PgPool,
37    sql: String,
38    params: Vec<Value>,
39}
40
41impl QueryRows {
42    /// Create a new query-rows operation.
43    pub fn new(pool: PgPool, sql: impl Into<String>, params: Vec<Value>) -> Self {
44        Self {
45            pool,
46            sql: sql.into(),
47            params,
48        }
49    }
50
51    /// Execute and return a typed result.
52    ///
53    /// # Errors
54    ///
55    /// Returns [`OperationError::External`] on SQL or connection errors.
56    pub async fn run(&self, _ctx: &OperationContext) -> Result<QueryRowsOutput, OperationError> {
57        let mut query = sqlx::query(&self.sql);
58        for p in &self.params {
59            query = bind_json_param(query, p);
60        }
61        let rows = query.fetch_all(&self.pool).await.map_err(pg_error)?;
62        let json_rows: Vec<Value> = rows.iter().map(row_to_json).collect::<Result<_, _>>()?;
63        Ok(QueryRowsOutput { rows: json_rows })
64    }
65}
66
67#[async_trait]
68impl Operation for QueryRows {
69    fn kind(&self) -> &str {
70        "postgres"
71    }
72    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
73        to_value(&self.run(ctx).await?)
74    }
75    fn input(&self) -> Option<Value> {
76        Some(serde_json::json!({ "sql": self.sql, "params": self.params }))
77    }
78}
79
80impl TypedOperation for QueryRows {
81    type Output = QueryRowsOutput;
82}
83
84/// Output of [`QueryOne`].
85#[derive(Debug, Clone, Serialize, Deserialize)]
86pub struct QueryOneOutput {
87    /// The single row as a JSON object.
88    pub row: Value,
89}
90
91/// Execute a `SELECT` query and return exactly one row.
92///
93/// # Errors
94///
95/// Returns an error if the query returns zero rows or more than one row.
96///
97/// # Examples
98///
99/// ```no_run
100/// use ironflow_ops_postgres::query::QueryOne;
101/// use ironflow_core::operation::Operation;
102///
103/// # fn example(pool: sqlx::PgPool) {
104/// let op = QueryOne::new(pool, "SELECT id, name FROM users WHERE id = $1", vec![serde_json::json!(1)]);
105/// assert_eq!(op.kind(), "postgres");
106/// # }
107/// ```
108pub struct QueryOne {
109    pool: PgPool,
110    sql: String,
111    params: Vec<Value>,
112}
113
114impl QueryOne {
115    /// Create a new query-one operation.
116    pub fn new(pool: PgPool, sql: impl Into<String>, params: Vec<Value>) -> Self {
117        Self {
118            pool,
119            sql: sql.into(),
120            params,
121        }
122    }
123
124    /// Execute and return a typed result.
125    ///
126    /// # Errors
127    ///
128    /// Returns [`OperationError::External`] if the query returns zero rows,
129    /// more than one row, or on SQL/connection errors.
130    pub async fn run(&self, _ctx: &OperationContext) -> Result<QueryOneOutput, OperationError> {
131        let mut query = sqlx::query(&self.sql);
132        for p in &self.params {
133            query = bind_json_param(query, p);
134        }
135        let row = query.fetch_one(&self.pool).await.map_err(pg_error)?;
136        let json = row_to_json(&row)?;
137        Ok(QueryOneOutput { row: json })
138    }
139}
140
141#[async_trait]
142impl Operation for QueryOne {
143    fn kind(&self) -> &str {
144        "postgres"
145    }
146    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
147        to_value(&self.run(ctx).await?)
148    }
149    fn input(&self) -> Option<Value> {
150        Some(serde_json::json!({ "sql": self.sql, "params": self.params }))
151    }
152}
153
154impl TypedOperation for QueryOne {
155    type Output = QueryOneOutput;
156}
157
158/// Output of [`QueryScalar`].
159#[derive(Debug, Clone, Serialize, Deserialize)]
160pub struct QueryScalarOutput {
161    /// The scalar value as JSON.
162    pub value: Value,
163}
164
165/// Execute a `SELECT` query and return a single scalar value.
166///
167/// The query must return exactly one row with one column.
168///
169/// # Examples
170///
171/// ```no_run
172/// use ironflow_ops_postgres::query::QueryScalar;
173/// use ironflow_core::operation::Operation;
174///
175/// # fn example(pool: sqlx::PgPool) {
176/// let op = QueryScalar::new(pool, "SELECT count(*) FROM users", vec![]);
177/// assert_eq!(op.kind(), "postgres");
178/// # }
179/// ```
180pub struct QueryScalar {
181    pool: PgPool,
182    sql: String,
183    params: Vec<Value>,
184}
185
186impl QueryScalar {
187    /// Create a new query-scalar operation.
188    pub fn new(pool: PgPool, sql: impl Into<String>, params: Vec<Value>) -> Self {
189        Self {
190            pool,
191            sql: sql.into(),
192            params,
193        }
194    }
195
196    /// Execute and return a typed result.
197    ///
198    /// # Errors
199    ///
200    /// Returns [`OperationError::External`] on SQL/connection errors or if the
201    /// query does not return exactly one row with one column.
202    pub async fn run(&self, _ctx: &OperationContext) -> Result<QueryScalarOutput, OperationError> {
203        let mut query = sqlx::query(&self.sql);
204        for p in &self.params {
205            query = bind_json_param(query, p);
206        }
207        let row = query.fetch_one(&self.pool).await.map_err(pg_error)?;
208        let columns = row.columns();
209        if columns.is_empty() {
210            return Err(OperationError::External {
211                origin: "postgres".to_string(),
212                message: "query returned no columns".to_string(),
213            });
214        }
215        let val = column_to_json(&row, 0)?;
216        Ok(QueryScalarOutput { value: val })
217    }
218}
219
220#[async_trait]
221impl Operation for QueryScalar {
222    fn kind(&self) -> &str {
223        "postgres"
224    }
225    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
226        to_value(&self.run(ctx).await?)
227    }
228    fn input(&self) -> Option<Value> {
229        Some(serde_json::json!({ "sql": self.sql, "params": self.params }))
230    }
231}
232
233impl TypedOperation for QueryScalar {
234    type Output = QueryScalarOutput;
235}
236
237#[cfg(test)]
238mod tests {
239    use super::*;
240
241    #[tokio::test]
242    async fn query_rows_kind() {
243        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
244        let op = QueryRows::new(pool, "SELECT 1", vec![]);
245        assert_eq!(op.kind(), "postgres");
246    }
247
248    #[tokio::test]
249    async fn query_one_kind() {
250        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
251        let op = QueryOne::new(pool, "SELECT 1", vec![]);
252        assert_eq!(op.kind(), "postgres");
253    }
254
255    #[tokio::test]
256    async fn query_scalar_kind() {
257        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
258        let op = QueryScalar::new(pool, "SELECT 1", vec![]);
259        assert_eq!(op.kind(), "postgres");
260    }
261
262    #[tokio::test]
263    async fn query_rows_input_no_secrets() {
264        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
265        let op = QueryRows::new(pool, "SELECT 1", vec![]);
266        let input = op.input().unwrap();
267        let text = input.to_string();
268        assert!(!text.contains("password"), "leaked secret: {text}");
269        assert!(!text.contains("postgres://"), "leaked URL: {text}");
270    }
271
272    #[tokio::test]
273    async fn query_one_input_no_secrets() {
274        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
275        let op = QueryOne::new(pool, "SELECT 1", vec![]);
276        let input = op.input().unwrap();
277        let text = input.to_string();
278        assert!(!text.contains("password"), "leaked secret: {text}");
279    }
280
281    #[tokio::test]
282    async fn query_scalar_input_no_secrets() {
283        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
284        let op = QueryScalar::new(pool, "SELECT 1", vec![]);
285        let input = op.input().unwrap();
286        let text = input.to_string();
287        assert!(!text.contains("password"), "leaked secret: {text}");
288    }
289}
ironflow_ops_postgres/query.rs

ironflow_ops_postgres/
query.rs
