ironflow_ops_postgres/schema/
columns.rs

1//! Column introspection operation.
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/// A column description.
13#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct ColumnInfo {
15    /// Column name.
16    pub name: String,
17    /// Data type (e.g. `integer`, `text`).
18    pub data_type: String,
19    /// Whether the column is nullable.
20    pub is_nullable: bool,
21    /// Default value expression, if any.
22    pub column_default: Option<String>,
23    /// Ordinal position (1-based).
24    pub ordinal_position: i32,
25}
26
27/// Output of [`ListColumns`].
28#[derive(Debug, Clone, Serialize, Deserialize)]
29pub struct ListColumnsOutput {
30    /// Columns of the table.
31    pub columns: Vec<ColumnInfo>,
32}
33
34/// Describe the columns of a table.
35///
36/// # Examples
37///
38/// ```no_run
39/// use ironflow_ops_postgres::schema::columns::ListColumns;
40/// use ironflow_core::operation::Operation;
41///
42/// # fn example(pool: sqlx::PgPool) {
43/// let op = ListColumns::new(pool, "public", "users");
44/// assert_eq!(op.kind(), "postgres");
45/// # }
46/// ```
47pub struct ListColumns {
48    pool: PgPool,
49    schema: String,
50    table: String,
51}
52
53impl ListColumns {
54    /// Create a new list-columns operation.
55    pub fn new(pool: PgPool, schema: impl Into<String>, table: impl Into<String>) -> Self {
56        Self {
57            pool,
58            schema: schema.into(),
59            table: table.into(),
60        }
61    }
62
63    /// Execute and return a typed result.
64    ///
65    /// # Errors
66    ///
67    /// Returns [`OperationError::External`] on connection errors.
68    pub async fn run(&self, _ctx: &OperationContext) -> Result<ListColumnsOutput, OperationError> {
69        let rows = sqlx::query(
70            "SELECT column_name, data_type, is_nullable, column_default, ordinal_position \
71             FROM information_schema.columns \
72             WHERE table_schema = $1 AND table_name = $2 \
73             ORDER BY ordinal_position",
74        )
75        .bind(&self.schema)
76        .bind(&self.table)
77        .fetch_all(&self.pool)
78        .await
79        .map_err(pg_error)?;
80        let columns = rows
81            .iter()
82            .map(|r| {
83                Ok(ColumnInfo {
84                    name: r.try_get::<String, _>("column_name").map_err(pg_error)?,
85                    data_type: r.try_get::<String, _>("data_type").map_err(pg_error)?,
86                    is_nullable: r.try_get::<String, _>("is_nullable").map_err(pg_error)? == "YES",
87                    column_default: r
88                        .try_get::<Option<String>, _>("column_default")
89                        .map_err(pg_error)?,
90                    ordinal_position: r.try_get::<i32, _>("ordinal_position").map_err(pg_error)?,
91                })
92            })
93            .collect::<Result<Vec<_>, OperationError>>()?;
94        Ok(ListColumnsOutput { columns })
95    }
96}
97
98#[async_trait]
99impl Operation for ListColumns {
100    fn kind(&self) -> &str {
101        "postgres"
102    }
103    async fn execute(&self, ctx: &OperationContext) -> Result<Value, OperationError> {
104        to_value(&self.run(ctx).await?)
105    }
106    fn input(&self) -> Option<Value> {
107        Some(serde_json::json!({ "schema": self.schema, "table": self.table }))
108    }
109}
110
111impl TypedOperation for ListColumns {
112    type Output = ListColumnsOutput;
113}
114
115#[cfg(test)]
116mod tests {
117    use super::*;
118
119    #[tokio::test]
120    async fn kind() {
121        let pool = PgPool::connect_lazy("postgres://localhost/test").unwrap();
122        let op = ListColumns::new(pool, "public", "users");
123        assert_eq!(op.kind(), "postgres");
124    }
125}
ironflow_ops_postgres/schema/columns.rs

ironflow_ops_postgres/schema/
columns.rs
