ironflow_ops_postgres/schema/
tables.rs

1//! Table listing and existence check operations.
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};
9
10use crate::helpers::{pg_error, to_value};
11
12/// Output of [`ListTables`].
13#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct ListTablesOutput {
15    /// Table names.
16    pub tables: Vec<String>,
17}
18
19/// List all tables in a given schema.
20///
21/// # Examples
22///
23/// ```no_run
24/// use ironflow_ops_postgres::schema::tables::ListTables;
25/// use ironflow_core::operation::Operation;
26///
27/// # fn example(pool: sqlx::PgPool) {
28/// let op = ListTables::new(pool, "public");
29/// assert_eq!(op.kind(), "postgres");
30/// # }
31/// ```
32pub struct ListTables {
33    pool: PgPool,
34    schema: String,
35}
36
37impl ListTables {
38    /// Create a new list-tables operation.
39    pub fn new(pool: PgPool, schema: impl Into<String>) -> Self {
40        Self {
41            pool,
42            schema: schema.into(),
43        }
44    }
45
46    /// Execute and return a typed result.
47    ///
48    /// # Errors
49    ///
50    /// Returns [`OperationError::External`] on connection errors.
51    pub async fn run(&self, _ctx: &OperationContext) -> Result<ListTablesOutput, OperationError> {
52        let rows = sqlx::query(
53            "SELECT table_name FROM information_schema.tables \
54             WHERE table_schema = $1 AND table_type = 'BASE TABLE' \
55             ORDER BY table_name",
56        )
57        .bind(&self.schema)
58        .fetch_all(&self.pool)
59        .await
60        .map_err(pg_error)?;
61        let tables = rows
62            .iter()
63            .map(|r| r.try_get::<String, _>("table_name").map_err(pg_error))
64            .collect::<Result<Vec<_>, _>>()?;
65        Ok(ListTablesOutput { tables })
66    }
67}
68
69#[async_trait]
70impl Operation for ListTables {
71    fn kind(&self) -> &str {
72        "postgres"
73    }
74    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
75        to_value(&self.run(ctx).await?)
76    }
77    fn input(&self) -> Option<Value> {
78        Some(serde_json::json!({ "schema": self.schema }))
79    }
80}
81
82impl TypedOperation for ListTables {
83    type Output = ListTablesOutput;
84}
85
86/// Output of [`TableExists`].
87#[derive(Debug, Clone, Serialize, Deserialize)]
88pub struct TableExistsOutput {
89    /// Whether the table exists.
90    pub exists: bool,
91}
92
93/// Check whether a table exists in a schema.
94///
95/// # Examples
96///
97/// ```no_run
98/// use ironflow_ops_postgres::schema::tables::TableExists;
99/// use ironflow_core::operation::Operation;
100///
101/// # fn example(pool: sqlx::PgPool) {
102/// let op = TableExists::new(pool, "public", "users");
103/// assert_eq!(op.kind(), "postgres");
104/// # }
105/// ```
106pub struct TableExists {
107    pool: PgPool,
108    schema: String,
109    table: String,
110}
111
112impl TableExists {
113    /// Create a new table-exists operation.
114    pub fn new(pool: PgPool, schema: impl Into<String>, table: impl Into<String>) -> Self {
115        Self {
116            pool,
117            schema: schema.into(),
118            table: table.into(),
119        }
120    }
121
122    /// Execute and return a typed result.
123    ///
124    /// # Errors
125    ///
126    /// Returns [`OperationError::External`] on connection errors.
127    pub async fn run(&self, _ctx: &OperationContext) -> Result<TableExistsOutput, OperationError> {
128        let row = sqlx::query(
129            "SELECT EXISTS( \
130                SELECT 1 FROM information_schema.tables \
131                WHERE table_schema = $1 AND table_name = $2 \
132            ) AS exists",
133        )
134        .bind(&self.schema)
135        .bind(&self.table)
136        .fetch_one(&self.pool)
137        .await
138        .map_err(pg_error)?;
139        let exists: bool = row.try_get("exists").map_err(pg_error)?;
140        Ok(TableExistsOutput { exists })
141    }
142}
143
144#[async_trait]
145impl Operation for TableExists {
146    fn kind(&self) -> &str {
147        "postgres"
148    }
149    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
150        to_value(&self.run(ctx).await?)
151    }
152    fn input(&self) -> Option<Value> {
153        Some(serde_json::json!({ "schema": self.schema, "table": self.table }))
154    }
155}
156
157impl TypedOperation for TableExists {
158    type Output = TableExistsOutput;
159}
160
161#[cfg(test)]
162mod tests {
163    use super::*;
164
165    #[tokio::test]
166    async fn list_tables_kind() {
167        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
168        let op = ListTables::new(pool, "public");
169        assert_eq!(op.kind(), "postgres");
170    }
171
172    #[tokio::test]
173    async fn table_exists_kind() {
174        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
175        let op = TableExists::new(pool, "public", "users");
176        assert_eq!(op.kind(), "postgres");
177    }
178
179    #[tokio::test]
180    async fn list_tables_input_no_secrets() {
181        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
182        let op = ListTables::new(pool, "public");
183        let input = op.input().unwrap();
184        let text = input.to_string();
185        assert!(!text.contains("postgres://"), "leaked URL: {text}");
186    }
187}
ironflow_ops_postgres/schema/tables.rs

ironflow_ops_postgres/schema/
tables.rs
