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/// The PostgreSQL extension enables querying PostgreSQL databases from Akar.
15pub struct PostgresExtension;
16
17impl Default for PostgresExtension {
18    fn default() -> Self {
19        Self::new()
20    }
21}
22
23impl PostgresExtension {
24    pub fn new() -> Self {
25        Self
26    }
27}
28
29impl Extension for PostgresExtension {
30    fn name(&self) -> &'static str {
31        "POSTGRES"
32    }
33
34    fn load(&self, context: &ExtensionContext) -> Result<(), String> {
35        use akar_function::registry::ScalarFunction;
36
37        #[cfg(feature = "native")]
38        {
39            use akar_function::Value;
40
41            // sql_query(conn_str: String, sql: String) → executes SQL against PostgreSQL
42            let query_fn: Arc<dyn Fn(&[Value]) -> Result<Value, String> + Send + Sync> = Arc::new(|args| {
43                if args.len() < 2 {
44                    return Err("sql_query requires (connection_string, sql) arguments".into());
45                }
46                let conn_str = match &args[0] {
47                    Value::String(s) => s.clone(),
48                    _ => return Err("sql_query: first argument must be a connection string".into()),
49                };
50                let sql = match &args[1] {
51                    Value::String(s) => s.clone(),
52                    _ => return Err("sql_query: second argument must be a SQL string".into()),
53                };
54
55                // Create a one-shot tokio runtime for this query
56                let rt = tokio::runtime::Runtime::new().map_err(|e| format!("Failed to create tokio runtime: {e}"))?;
57
58                let (client, connection) = rt
59                    .block_on(async { tokio_postgres::connect(&conn_str, tokio_postgres::NoTls).await })
60                    .map_err(|e| format!("PostgreSQL connect error: {e}"))?;
61
62                // Spawn connection handler
63                rt.spawn(async move {
64                    if let Err(e) = connection.await {
65                        tracing::warn!("PostgreSQL connection error: {e}");
66                    }
67                });
68
69                let rows = rt
70                    .block_on(async { client.query(&sql, &[]).await })
71                    .map_err(|e| format!("PostgreSQL query error: {e}"))?;
72
73                // Collect first row as string result
74                if let Some(row) = rows.first() {
75                    let mut parts = Vec::new();
76                    for i in 0..row.len() {
77                        let val: Option<&str> = row.try_get::<_, &str>(i).ok();
78                        parts.push(val.unwrap_or("NULL").to_string());
79                    }
80                    Ok(Value::String(parts.join(",")))
81                } else {
82                    Ok(Value::String("(empty)".into()))
83                }
84            });
85
86            context.register_scalar_function(
87                "sql_query",
88                ScalarFunction::CustomScalar {
89                    name: "sql_query".into(),
90                    execute: query_fn,
91                },
92            );
93
94            tracing::info!("PostgreSQL extension loaded: 1 function registered (tokio-postgres native)");
95        }
96
97        #[cfg(not(feature = "native"))]
98        {
99            context.register_scalar_function(
100                "sql_query",
101                ScalarFunction::CustomScalar {
102                    name: "sql_query".into(),
103                    execute: Arc::new(|_| Err("PostgreSQL not available (feature 'native' disabled)".into())),
104                },
105            );
106            tracing::info!("PostgreSQL extension loaded: 1 function registered (placeholder)");
107        }
108
109        Ok(())
110    }
111}
112
113#[cfg(test)]
114mod tests {
115    use super::*;
116
117    #[test]
118    fn test_postgres_extension_name() {
119        let ext = PostgresExtension::new();
120        assert_eq!(ext.name(), "POSTGRES");
121    }
122}
akar_postgres/lib.rs

akar_postgres/
lib.rs
