ironflow_ops_postgres/
execute.rs

1//! Mutating operations: `INSERT`, `UPDATE`, `DELETE`, batch execution, and
2//! transactions.
3
4use async_trait::async_trait;
5use ironflow_core::error::OperationError;
6use ironflow_core::operation::{Operation, OperationContext, TypedOperation};
7use serde::{Deserialize, Serialize};
8use serde_json::Value;
9use sqlx::{Executor, PgPool};
10
11use crate::helpers::{bind_json_param, pg_error, to_value};
12
13/// Output of [`Execute`].
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct ExecuteOutput {
16    /// Number of rows affected by the statement.
17    pub rows_affected: u64,
18}
19
20/// Execute a single mutating SQL statement (`INSERT`, `UPDATE`, `DELETE`).
21///
22/// Returns the number of rows affected.
23///
24/// # Examples
25///
26/// ```no_run
27/// use ironflow_ops_postgres::execute::Execute;
28/// use ironflow_core::operation::Operation;
29///
30/// # fn example(pool: sqlx::PgPool) {
31/// let op = Execute::new(pool, "INSERT INTO users (name) VALUES ($1)", vec![serde_json::json!("Alice")]);
32/// assert_eq!(op.kind(), "postgres");
33/// # }
34/// ```
35pub struct Execute {
36    pool: PgPool,
37    sql: String,
38    params: Vec<Value>,
39}
40
41impl Execute {
42    /// Create a new execute 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<ExecuteOutput, 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 result = query.execute(&self.pool).await.map_err(pg_error)?;
62        Ok(ExecuteOutput {
63            rows_affected: result.rows_affected(),
64        })
65    }
66}
67
68#[async_trait]
69impl Operation for Execute {
70    fn kind(&self) -> &str {
71        "postgres"
72    }
73    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
74        to_value(&self.run(ctx).await?)
75    }
76    fn input(&self) -> Option<Value> {
77        Some(serde_json::json!({ "sql": self.sql, "params": self.params }))
78    }
79}
80
81impl TypedOperation for Execute {
82    type Output = ExecuteOutput;
83}
84
85/// Output of [`ExecuteBatch`].
86#[derive(Debug, Clone, Serialize, Deserialize)]
87pub struct ExecuteBatchOutput {
88    /// Number of rows affected by each statement, in order.
89    pub rows_affected: Vec<u64>,
90}
91
92/// Execute multiple SQL statements in sequence (not transactional).
93///
94/// Each statement is executed independently. If a statement fails, the
95/// remaining statements are not executed.
96///
97/// # Safety
98///
99/// Statements are executed as raw SQL. Never build them from untrusted input
100/// without proper parameterization. Use [`Execute`] with bind parameters for
101/// user-supplied values.
102///
103/// # Examples
104///
105/// ```no_run
106/// use ironflow_ops_postgres::execute::ExecuteBatch;
107/// use ironflow_core::operation::Operation;
108///
109/// # fn example(pool: sqlx::PgPool) {
110/// let op = ExecuteBatch::new(pool, vec![
111///     "CREATE TABLE IF NOT EXISTS t (id int)".to_string(),
112///     "INSERT INTO t VALUES (1)".to_string(),
113/// ]);
114/// assert_eq!(op.kind(), "postgres");
115/// # }
116/// ```
117pub struct ExecuteBatch {
118    pool: PgPool,
119    statements: Vec<String>,
120}
121
122impl ExecuteBatch {
123    /// Create a new batch-execute operation.
124    pub fn new(pool: PgPool, statements: Vec<String>) -> Self {
125        Self { pool, statements }
126    }
127
128    /// Execute and return a typed result.
129    ///
130    /// # Errors
131    ///
132    /// Returns [`OperationError::External`] on SQL or connection errors.
133    /// Stops at the first failing statement.
134    pub async fn run(&self, _ctx: &OperationContext) -> Result<ExecuteBatchOutput, OperationError> {
135        let mut results = Vec::with_capacity(self.statements.len());
136        for stmt in &self.statements {
137            let result = self.pool.execute(stmt.as_str()).await.map_err(pg_error)?;
138            results.push(result.rows_affected());
139        }
140        Ok(ExecuteBatchOutput {
141            rows_affected: results,
142        })
143    }
144}
145
146#[async_trait]
147impl Operation for ExecuteBatch {
148    fn kind(&self) -> &str {
149        "postgres"
150    }
151    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
152        to_value(&self.run(ctx).await?)
153    }
154    fn input(&self) -> Option<Value> {
155        Some(serde_json::json!({ "statements": self.statements }))
156    }
157}
158
159impl TypedOperation for ExecuteBatch {
160    type Output = ExecuteBatchOutput;
161}
162
163/// Output of [`Transaction`].
164#[derive(Debug, Clone, Serialize, Deserialize)]
165pub struct TransactionOutput {
166    /// Number of rows affected by each statement, in order.
167    pub rows_affected: Vec<u64>,
168}
169
170/// Execute multiple SQL statements inside an atomic transaction.
171///
172/// All statements succeed together or are rolled back on the first error.
173///
174/// # Safety
175///
176/// Statements are executed as raw SQL. Never build them from untrusted input
177/// without proper parameterization. Use [`Execute`] with bind parameters for
178/// user-supplied values.
179///
180/// # Examples
181///
182/// ```no_run
183/// use ironflow_ops_postgres::execute::Transaction;
184/// use ironflow_core::operation::Operation;
185///
186/// # fn example(pool: sqlx::PgPool) {
187/// let op = Transaction::new(pool, vec![
188///     "INSERT INTO accounts (id, balance) VALUES (1, 100)".to_string(),
189///     "UPDATE accounts SET balance = balance - 50 WHERE id = 1".to_string(),
190/// ]);
191/// assert_eq!(op.kind(), "postgres");
192/// # }
193/// ```
194pub struct Transaction {
195    pool: PgPool,
196    statements: Vec<String>,
197}
198
199impl Transaction {
200    /// Create a new transaction operation.
201    pub fn new(pool: PgPool, statements: Vec<String>) -> Self {
202        Self { pool, statements }
203    }
204
205    /// Execute and return a typed result.
206    ///
207    /// # Errors
208    ///
209    /// Returns [`OperationError::External`] on SQL or connection errors.
210    /// The transaction is rolled back on any failure.
211    pub async fn run(&self, _ctx: &OperationContext) -> Result<TransactionOutput, OperationError> {
212        let mut tx = self.pool.begin().await.map_err(pg_error)?;
213        let mut results = Vec::with_capacity(self.statements.len());
214        for stmt in &self.statements {
215            let result = tx.execute(stmt.as_str()).await.map_err(pg_error)?;
216            results.push(result.rows_affected());
217        }
218        tx.commit().await.map_err(pg_error)?;
219        Ok(TransactionOutput {
220            rows_affected: results,
221        })
222    }
223}
224
225#[async_trait]
226impl Operation for Transaction {
227    fn kind(&self) -> &str {
228        "postgres"
229    }
230    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
231        to_value(&self.run(ctx).await?)
232    }
233    fn input(&self) -> Option<Value> {
234        Some(serde_json::json!({ "statements": self.statements }))
235    }
236}
237
238impl TypedOperation for Transaction {
239    type Output = TransactionOutput;
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245
246    #[tokio::test]
247    async fn execute_kind() {
248        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
249        let op = Execute::new(pool, "INSERT INTO t VALUES (1)", vec![]);
250        assert_eq!(op.kind(), "postgres");
251    }
252
253    #[tokio::test]
254    async fn execute_batch_kind() {
255        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
256        let op = ExecuteBatch::new(pool, vec!["SELECT 1".to_string()]);
257        assert_eq!(op.kind(), "postgres");
258    }
259
260    #[tokio::test]
261    async fn transaction_kind() {
262        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
263        let op = Transaction::new(pool, vec!["SELECT 1".to_string()]);
264        assert_eq!(op.kind(), "postgres");
265    }
266
267    #[tokio::test]
268    async fn execute_input_no_secrets() {
269        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
270        let op = Execute::new(pool, "INSERT INTO t VALUES ($1)", vec![]);
271        let input = op.input().unwrap();
272        let text = input.to_string();
273        assert!(!text.contains("postgres://"), "leaked URL: {text}");
274    }
275
276    #[tokio::test]
277    async fn execute_batch_input_no_secrets() {
278        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
279        let op = ExecuteBatch::new(pool, vec!["SELECT 1".to_string()]);
280        let input = op.input().unwrap();
281        let text = input.to_string();
282        assert!(!text.contains("postgres://"), "leaked URL: {text}");
283    }
284
285    #[tokio::test]
286    async fn transaction_input_no_secrets() {
287        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
288        let op = Transaction::new(pool, vec!["SELECT 1".to_string()]);
289        let input = op.input().unwrap();
290        let text = input.to_string();
291        assert!(!text.contains("postgres://"), "leaked URL: {text}");
292    }
293}
ironflow_ops_postgres/execute.rs

ironflow_ops_postgres/
execute.rs
