akar_postgres/
lib.rs

1//! PostgreSQL extension for Akar.
2//!
3//! Provides integration with PostgreSQL databases using the native
4//! `tokio-postgres` crate with a synchronous wrapper via `block_on`.
5//!
6//! ## Native Rust approach
7//! Uses `tokio-postgres` (v0.7) with `futures::executor::block_on()` to bridge
8//! the async PostgreSQL client with Akar's synchronous runtime.
9//! Most complex extension due to catalog binding + table enumeration + type mapping.
10
11use akar_extension::{Extension, ExtensionContext};
12use std::sync::Arc;
13
14/// Render a single cell of a PostgreSQL row as text.
15///
16/// `try_get` needs a concrete type at compile time, so we cascade through the
17/// common representations. A NULL cell yields `Ok(None)` for every `Option<T>`
18/// probe and is reported as "NULL"; cells whose type matches none of the probes
19/// (e.g. bytea, timestamps) fall back to their column type name.
20#[cfg(feature = "native")]
21fn postgres_value_to_string(row: &tokio_postgres::Row, i: usize) -> String {
22    let type_name = row.columns().get(i).map(|c| c.type_().name()).unwrap_or("unknown").to_string();
23    if let Ok(Some(v)) = row.try_get::<_, Option<String>>(i) {
24        return v;
25    }
26    if let Ok(Some(v)) = row.try_get::<_, Option<i64>>(i) {
27        return v.to_string();
28    }
29    if let Ok(Some(v)) = row.try_get::<_, Option<i32>>(i) {
30        return v.to_string();
31    }
32    if let Ok(Some(v)) = row.try_get::<_, Option<f64>>(i) {
33        return v.to_string();
34    }
35    if let Ok(Some(v)) = row.try_get::<_, Option<bool>>(i) {
36        return v.to_string();
37    }
38    // A NULL cell succeeds with None for any Option<T> probe.
39    if row.try_get::<_, Option<i64>>(i).is_ok() {
40        "NULL".to_string()
41    } else {
42        format!("<{type_name}>")
43    }
44}
45
46/// The PostgreSQL extension enables querying PostgreSQL databases from Akar.
47pub struct PostgresExtension;
48
49impl Default for PostgresExtension {
50    fn default() -> Self {
51        Self::new()
52    }
53}
54
55impl PostgresExtension {
56    pub fn new() -> Self {
57        Self
58    }
59}
60
61impl Extension for PostgresExtension {
62    fn name(&self) -> &'static str {
63        "POSTGRES"
64    }
65
66    fn load(&self, context: &ExtensionContext) -> Result<(), String> {
67        use akar_function::registry::ScalarFunction;
68
69        #[cfg(feature = "native")]
70        {
71            use akar_function::Value;
72
73            // sql_query(conn_str: String, sql: String) → executes SQL against PostgreSQL
74            let query_fn: Arc<dyn Fn(&[Value]) -> Result<Value, String> + Send + Sync> = Arc::new(|args| {
75                if args.len() < 2 {
76                    return Err("sql_query requires (connection_string, sql) arguments".into());
77                }
78                let conn_str = match &args[0] {
79                    Value::String(s) => s.clone(),
80                    _ => return Err("sql_query: first argument must be a connection string".into()),
81                };
82                let sql = match &args[1] {
83                    Value::String(s) => s.clone(),
84                    _ => return Err("sql_query: second argument must be a SQL string".into()),
85                };
86
87                // Create a one-shot tokio runtime for this query
88                let rt = tokio::runtime::Runtime::new().map_err(|e| format!("Failed to create tokio runtime: {e}"))?;
89
90                let (client, connection) = rt
91                    .block_on(async { tokio_postgres::connect(&conn_str, tokio_postgres::NoTls).await })
92                    .map_err(|e| format!("PostgreSQL connect error: {e}"))?;
93
94                // Spawn connection handler
95                rt.spawn(async move {
96                    if let Err(e) = connection.await {
97                        tracing::warn!("PostgreSQL connection error: {e}");
98                    }
99                });
100
101                let rows = rt
102                    .block_on(async { client.query(&sql, &[]).await })
103                    .map_err(|e| format!("PostgreSQL query error: {e}"))?;
104
105                // Collect every row (all columns) as strings, not just the first.
106                let mut parts = Vec::new();
107                for row in &rows {
108                    for i in 0..row.len() {
109                        parts.push(postgres_value_to_string(row, i));
110                    }
111                }
112                if parts.is_empty() {
113                    Ok(Value::String("(empty)".into()))
114                } else {
115                    Ok(Value::String(parts.join(",")))
116                }
117            });
118
119            context.register_scalar_function(
120                "sql_query",
121                ScalarFunction::CustomScalar {
122                    name: "sql_query".into(),
123                    execute: query_fn,
124                },
125            );
126
127            tracing::info!("PostgreSQL extension loaded: 1 function registered (tokio-postgres native)");
128        }
129
130        #[cfg(not(feature = "native"))]
131        {
132            context.register_scalar_function(
133                "sql_query",
134                ScalarFunction::CustomScalar {
135                    name: "sql_query".into(),
136                    execute: Arc::new(|_| Err("PostgreSQL not available (feature 'native' disabled)".into())),
137                },
138            );
139            tracing::info!("PostgreSQL extension loaded: 1 function registered (placeholder)");
140        }
141
142        Ok(())
143    }
144}
145
146#[cfg(test)]
147mod tests {
148    use super::*;
149
150    #[test]
151    fn test_postgres_extension_name() {
152        let ext = PostgresExtension::new();
153        assert_eq!(ext.name(), "POSTGRES");
154    }
155}
akar_postgres/lib.rs

akar_postgres/
lib.rs
