sqlx: implement working read queries

libsql/src/connection.rs

@@ -245,3 +245,9 @@ impl Connection {
         self.conn.load_extension(dylib_path.as_ref(), entry_point)
     }
 }
+
+impl fmt::Debug for Connection {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Connection").finish()
+    }
+}

sqlx-libsql/Cargo.toml

@@ -9,3 +9,4 @@ sqlx-core = "=0.8.2"
 futures-core = "0.3"
 futures-util = "0.3"
 log = "0.4"
+async-stream = "0.3"

sqlx-libsql/examples/todos/Cargo.toml

@@ -12,3 +12,4 @@ sqlx = { version = "0.8", features = [ "runtime-tokio", "tls-native-tls" ] }
 sqlx-libsql = { path = "../.." }
 clap = { version = "4", features = ["derive"] }
 tokio = { version = "1.20.0", features = ["rt", "macros"]}
+tokio-stream = "0.1"

sqlx-libsql/examples/todos/src/main.rs

@@ -19,6 +19,20 @@ async fn main() -> anyhow::Result<()> {
     let args = Args::parse();
     let pool = LibsqlPool::connect(&env::var("DATABASE_URL")?).await?;
 
+    // Migrations currently do not work, so we must do them manually
+    sqlx::query(
+        r#"
+CREATE TABLE IF NOT EXISTS todos
+(
+    id          INTEGER PRIMARY KEY NOT NULL,
+    description TEXT                NOT NULL,
+    done        BOOLEAN             NOT NULL DEFAULT 0
+);
+        "#,
+    )
+    .execute(&pool)
+    .await?;
+
     match args.cmd {
         Some(Command::Add { description }) => {
             println!("Adding new todo with description '{description}'");
@@ -87,14 +101,12 @@ ORDER BY id
     .fetch_all(pool)
     .await?;
 
-    for _rec in recs {
-        // TODO(lucio): impl this
-        // println!(
-        //     "- [{}] {}: {}",
-        //     if rec.done { "x" } else { " " },
-        //     rec.id,
-        //     &rec.description,
-        // );
+    for rec in recs {
+        let done = rec.get::<bool>(2).unwrap();
+        let id = rec.get::<u64>(0).unwrap();
+        let desc = rec.get::<String>(1).unwrap();
+
+        println!("- [{}] {}: {}", if done { "x" } else { " " }, id, &desc,);
     }
 
     Ok(())

sqlx-libsql/src/lib.rs

@@ -3,6 +3,7 @@
 use std::{
     iter::{Extend, IntoIterator},
     marker::PhantomData,
+    ops::Deref,
     str::FromStr,
 };
 
@@ -20,6 +21,7 @@ use sqlx_core::{
     type_info::TypeInfo,
     types::Type,
     value::{Value, ValueRef},
+    Either,
 };
 
 #[derive(Debug)]
@@ -42,67 +44,6 @@ impl Database for Libsql {
     const URL_SCHEMES: &'static [&'static str] = &["libsql"];
 }
 
-// #[derive(Debug)]
-// pub struct LibsqlExecutor<'a> {
-//     _a: PhantomData<&'a ()>,
-// }
-
-// impl<'a> Executor<'a> for LibsqlExecutor<'a> {
-//     type Database = Libsql;
-
-//     fn fetch_many<'e, 'q: 'e, E>(
-//         self,
-//         query: E,
-//     ) -> futures_core::stream::BoxStream<
-//         'e,
-//         Result<
-//             sqlx_core::Either<
-//                 <Self::Database as Database>::QueryResult,
-//                 <Self::Database as Database>::Row,
-//             >,
-//             sqlx_core::Error,
-//         >,
-//     >
-//     where
-//         'a: 'e,
-//         E: 'q + sqlx_core::executor::Execute<'q, Self::Database>,
-//     {
-//         todo!()
-//     }
-
-//     fn fetch_optional<'e, 'q: 'e, E>(
-//         self,
-//         query: E,
-//     ) -> BoxFuture<'e, Result<Option<<Self::Database as Database>::Row>, sqlx_core::Error>>
-//     where
-//         'a: 'e,
-//         E: 'q + sqlx_core::executor::Execute<'q, Self::Database>,
-//     {
-//         todo!()
-//     }
-
-//     fn prepare_with<'e, 'q: 'e>(
-//         self,
-//         sql: &'q str,
-//         parameters: &'e [<Self::Database as Database>::TypeInfo],
-//     ) -> BoxFuture<'e, Result<<Self::Database as Database>::Statement<'q>, sqlx_core::Error>>
-//     where
-//         'a: 'e,
-//     {
-//         todo!()
-//     }
-
-//     fn describe<'e, 'q: 'e>(
-//         self,
-//         sql: &'q str,
-//     ) -> BoxFuture<'e, Result<sqlx_core::describe::Describe<Self::Database>, sqlx_core::Error>>
-//     where
-//         'a: 'e,
-//     {
-//         todo!()
-//     }
-// }
-
 impl<'a> Executor<'a> for &'a mut LibsqlConnection {
     type Database = Libsql;
 
@@ -123,7 +64,15 @@ impl<'a> Executor<'a> for &'a mut LibsqlConnection {
         'a: 'e,
         E: 'q + sqlx_core::executor::Execute<'q, Self::Database>,
     {
-        todo!()
+        let sql = query.sql();
+
+        Box::pin(async_stream::stream! {
+            let mut rows = self.conn.query(sql, ()).await.unwrap();
+
+            while let Some(row) = rows.next().await.unwrap() {
+                yield Ok(Either::Right(LibsqlRow { row }));
+            }
+        })
     }
 
     fn fetch_optional<'e, 'q: 'e, E>(
@@ -171,7 +120,9 @@ sqlx_core::impl_acquire!(Libsql, LibsqlConnection);
 sqlx_core::impl_encode_for_option!(Libsql);
 
 #[derive(Debug)]
-pub struct LibsqlConnection {}
+pub struct LibsqlConnection {
+    conn: libsql::Connection,
+}
 
 impl Connection for LibsqlConnection {
     type Database = Libsql;
@@ -187,7 +138,7 @@ impl Connection for LibsqlConnection {
     }
 
     fn ping(&mut self) -> BoxFuture<'_, Result<(), sqlx_core::Error>> {
-        todo!()
+        Box::pin(async { Ok(()) })
     }
 
     fn begin(
@@ -216,7 +167,9 @@ impl Connection for LibsqlConnection {
 }
 
 #[derive(Debug, Clone)]
-pub struct LibsqlConnectionOptions {}
+pub struct LibsqlConnectionOptions {
+    url: String,
+}
 
 impl ConnectOptions for LibsqlConnectionOptions {
     type Connection = LibsqlConnection;
@@ -229,7 +182,16 @@ impl ConnectOptions for LibsqlConnectionOptions {
     where
         Self::Connection: Sized,
     {
-        todo!()
+        let url = self.url.clone();
+
+        Box::pin(async {
+            let db = libsql::Builder::new_remote(url, "".to_string())
+                .build()
+                .await
+                .unwrap();
+            let conn = db.connect().unwrap();
+            Ok(LibsqlConnection { conn })
+        })
     }
 
     fn log_statements(self, level: log::LevelFilter) -> Self {
@@ -245,7 +207,7 @@ impl FromStr for LibsqlConnectionOptions {
     type Err = sqlx_core::Error;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        todo!()
+        Ok(LibsqlConnectionOptions { url: s.to_string() })
     }
 }
 
@@ -277,7 +239,17 @@ impl TransactionManager for LibsqlTransactionManager {
     }
 }
 
-pub struct LibsqlRow {}
+pub struct LibsqlRow {
+    row: libsql::Row,
+}
+
+impl Deref for LibsqlRow {
+    type Target = libsql::Row;
+
+    fn deref(&self) -> &Self::Target {
+        &self.row
+    }
+}
 
 impl Row for LibsqlRow {
     type Database = Libsql;