Implement async support

butane/Cargo.toml

@@ -40,11 +40,12 @@ log_for_test = {package="log", version = "0.4"}
 quote = "1.0"
 proc-macro2="1.0"
 once_cell="1.5.2"
-postgres = { version = "0.19", features=["with-geo-types-0_7"] }
+tokio-postgres = { version = "0.7", features=["with-geo-types-0_7"] }
+# tokio-postgres = { version = "0.7" }
 r2d2_for_test = {package="r2d2", version = "0.8"}
 rusqlite = {workspace=true}
 serde_json = "1.0"
 uuid_for_test = {package="uuid", version = "1.2", features=["v4"] }
 
 [package.metadata.docs.rs]
-all-features = true
+all-features = true

butane/tests/basic.rs

@@ -4,7 +4,7 @@ use butane::{butane_type, find, model, query};
 use butane::{ForeignKey, ObjectState};
 use paste;
 #[cfg(feature = "pg")]
-use postgres;
+use tokio_postgres as postgres;
 #[cfg(feature = "sqlite")]
 use rusqlite;
 

butane/tests/custom_pg.rs

@@ -9,7 +9,7 @@ mod custom_pg {
     use butane::{butane_type, db::Connection, model, ObjectState};
     use butane::{FieldType, FromSql, SqlType, SqlVal, SqlValRef, ToSql};
     use geo_types;
-    use postgres;
+    use tokio_postgres as postgres;
     use std::result::Result;
 
     // newtype so we can implement traits for it.

butane_core/Cargo.toml

@@ -15,7 +15,7 @@ debug = ["log"]
 sqlite = ["rusqlite"]
 sqlite-bundled = ["rusqlite/bundled"]
 tls = ["postgres-native-tls", "native-tls"]
-pg = ["postgres", "bytes"]
+pg = ["tokio-postgres", "bytes"]
 
 
 [dependencies]
@@ -25,10 +25,9 @@ fallible-iterator = "0.2"
 fallible-streaming-iterator = "0.1"
 fs2 = "0.4" # for file locks
 hex = "0.4"
-once_cell="1.5"
 log = { version="0.4", optional=true }
 native-tls={ version = "0.2", optional = true }
-postgres={ version = "0.19", features=["with-chrono-0_4"], optional = true}
+tokio-postgres={ version = "0.7", features=["with-chrono-0_4"], optional = true}
 postgres-native-tls={ version = "0.5", optional = true }
 proc-macro2 = "1.0"
 pin-project = "1"
@@ -42,3 +41,6 @@ syn = { version = "1.0", features = ["full", "extra-traits"] }
 thiserror = "1.0"
 chrono = { version = "0.4", features=["serde"], optional = true }
 uuid = {workspace=true, optional=true}
+async-trait = "0.1"
+tokio = { version = "1.24", features=["rt", "sync"] }
+futures-util = "0.3"

butane_core/src/custom.rs

@@ -7,12 +7,13 @@
 
 use serde::{Deserialize, Serialize};
 use std::fmt;
+use tokio_postgres as postgres;
 
 /// For use with [SqlType::Custom](crate::SqlType)
 #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
 pub enum SqlTypeCustom {
     #[cfg(feature = "pg")]
-    Pg(#[serde(with = "pgtypeser")] postgres::types::Type),
+    Pg(#[serde(with = "pgtypeser")] tokio_postgres::types::Type),
 }
 
 /// For use with [SqlVal::Custom](crate::SqlVal)
@@ -137,6 +138,7 @@ impl From<SqlValRefCustom<'_>> for SqlValCustom {
 #[cfg(feature = "pg")]
 mod pgtypeser {
     use serde::{Deserialize, Deserializer, Serialize, Serializer};
+    use tokio_postgres as postgres;
 
     pub fn serialize<S>(ty: &postgres::types::Type, serializer: S) -> Result<S::Ok, S::Error>
     where

butane_core/src/db/connmethods.rs

@@ -5,14 +5,16 @@ use crate::query::{BoolExpr, Expr, Order};
 use crate::{Result, SqlType, SqlVal, SqlValRef};
 use std::ops::{Deref, DerefMut};
 use std::vec::Vec;
+use async_trait::async_trait;
 
 /// Methods available on a database connection. Most users do not need
 /// to call these methods directly and will instead use methods on
 /// [DataObject][crate::DataObject] or the `query!` macro. This trait is
 /// implemented by both database connections and transactions.
-pub trait ConnectionMethods {
-    fn execute(&self, sql: &str) -> Result<()>;
-    fn query<'a, 'b, 'c: 'a>(
+#[async_trait]
+pub trait ConnectionMethods: Sync {
+    async fn execute(&self, sql: &str) -> Result<()>;
+    async fn query<'a, 'b, 'c: 'a>(
         &'c self,
         table: &str,
         columns: &'b [Column],
@@ -21,38 +23,38 @@ pub trait ConnectionMethods {
         offset: Option<i32>,
         sort: Option<&[Order]>,
     ) -> Result<RawQueryResult<'a>>;
-    fn insert_returning_pk(
+    async fn insert_returning_pk(
         &self,
         table: &str,
         columns: &[Column],
         pkcol: &Column,
         values: &[SqlValRef<'_>],
     ) -> Result<SqlVal>;
     /// Like `insert_returning_pk` but with no return value
-    fn insert_only(&self, table: &str, columns: &[Column], values: &[SqlValRef<'_>]) -> Result<()>;
+    async fn insert_only(&self, table: &str, columns: &[Column], values: &[SqlValRef<'_>]) -> Result<()>;
     /// Insert unless there's a conflict on the primary key column, in which case update
-    fn insert_or_replace(
+    async fn insert_or_replace(
         &self,
         table: &str,
         columns: &[Column],
         pkcol: &Column,
         values: &[SqlValRef<'_>],
     ) -> Result<()>;
-    fn update(
+    async fn update<'a>(
         &self,
         table: &str,
         pkcol: Column,
-        pk: SqlValRef,
+        pk: SqlValRef<'a>,
         columns: &[Column],
         values: &[SqlValRef<'_>],
     ) -> Result<()>;
-    fn delete(&self, table: &str, pkcol: &'static str, pk: SqlVal) -> Result<()> {
-        self.delete_where(table, BoolExpr::Eq(pkcol, Expr::Val(pk)))?;
+    async fn delete(&self, table: &str, pkcol: &'static str, pk: SqlVal) -> Result<()> {
+        self.delete_where(table, BoolExpr::Eq(pkcol, Expr::Val(pk))).await?;
         Ok(())
     }
-    fn delete_where(&self, table: &str, expr: BoolExpr) -> Result<usize>;
+    async fn delete_where(&self, table: &str, expr: BoolExpr) -> Result<usize>;
     /// Tests if a table exists in the database.
-    fn has_table(&self, table: &str) -> Result<bool>;
+    async fn has_table(&self, table: &str) -> Result<bool>;
 }
 
 /// Represents a database column. Most users do not need to use this

butane_core/src/db/macros.rs

@@ -1,11 +1,12 @@
 #[macro_export]
 macro_rules! connection_method_wrapper {
     ($ty:path) => {
+        #[async_trait::async_trait]
         impl ConnectionMethods for $ty {
-            fn execute(&self, sql: &str) -> Result<()> {
-                ConnectionMethods::execute(self.wrapped_connection_methods()?, sql)
+            async fn execute(&self, sql: &str) -> Result<()> {
+                ConnectionMethods::execute(self.wrapped_connection_methods()?, sql).await
             }
-            fn query<'a, 'b, 'c: 'a>(
+            async fn query<'a, 'b, 'c: 'a>(
                 &'c self,
                 table: &str,
                 columns: &'b [Column],
@@ -15,9 +16,9 @@ macro_rules! connection_method_wrapper {
                 sort: Option<&[$crate::query::Order]>,
             ) -> Result<RawQueryResult<'a>> {
                 self.wrapped_connection_methods()?
-                    .query(table, columns, expr, limit, offset, sort)
+                    .query(table, columns, expr, limit, offset, sort).await
             }
-            fn insert_returning_pk(
+            async fn insert_returning_pk(
                 &self,
                 table: &str,
                 columns: &[Column],
@@ -26,17 +27,19 @@ macro_rules! connection_method_wrapper {
             ) -> Result<SqlVal> {
                 self.wrapped_connection_methods()?
                     .insert_returning_pk(table, columns, pkcol, values)
+                    .await
             }
-            fn insert_only(
+            async fn insert_only(
                 &self,
                 table: &str,
                 columns: &[Column],
                 values: &[SqlValRef<'_>],
             ) -> Result<()> {
                 self.wrapped_connection_methods()?
                     .insert_only(table, columns, values)
+                    .await
             }
-            fn insert_or_replace(
+            async fn insert_or_replace(
                 &self,
                 table: &str,
                 columns: &[Column],
@@ -45,23 +48,25 @@ macro_rules! connection_method_wrapper {
             ) -> Result<()> {
                 self.wrapped_connection_methods()?
                     .insert_or_replace(table, columns, pkcol, values)
+                    .await
             }
-            fn update(
+            async fn update<'a>(
                 &self,
                 table: &str,
                 pkcol: Column,
-                pk: SqlValRef,
+                pk: SqlValRef<'a>,
                 columns: &[Column],
                 values: &[SqlValRef<'_>],
             ) -> Result<()> {
                 self.wrapped_connection_methods()?
                     .update(table, pkcol, pk, columns, values)
+                    .await
             }
-            fn delete_where(&self, table: &str, expr: BoolExpr) -> Result<usize> {
-                self.wrapped_connection_methods()?.delete_where(table, expr)
+            async fn delete_where(&self, table: &str, expr: BoolExpr) -> Result<usize> {
+                self.wrapped_connection_methods()?.delete_where(table, expr).await
             }
-            fn has_table(&self, table: &str) -> Result<bool> {
-                self.wrapped_connection_methods()?.has_table(table)
+            async fn has_table(&self, table: &str) -> Result<bool> {
+                self.wrapped_connection_methods()?.has_table(table).await
             }
         }
     };