WIP: PostreSQL, MySQL and SQLite using SQLx

Cargo.toml

@@ -24,77 +24,72 @@ features = [ "full", "serde-support", "json-1", "uuid-0_8", "chrono-0_4", "array
 [features]
 default = []
 
-full = ["pooled", "sqlite", "json-1", "postgresql", "uuid-0_8", "chrono-0_4", "mysql", "mssql"]
+full = ["pooled", "json-1", "postgresql", "uuid-0_8", "chrono-0_4", "mysql", "mssql", "sqlite"]
 full-postgresql = ["pooled", "postgresql", "json-1", "uuid-0_8", "chrono-0_4", "array"]
 full-mysql = ["pooled", "mysql", "json-1", "uuid-0_8", "chrono-0_4"]
 full-sqlite = ["pooled", "sqlite", "json-1", "uuid-0_8", "chrono-0_4"]
 full-mssql = ["pooled", "mssql"]
 
-single = ["sqlite", "json-1", "postgresql", "uuid-0_8", "chrono-0_4", "mysql", "mssql"]
+single = [ "json-1", "postgresql", "uuid-0_8", "chrono-0_4", "mysql", "mssql", "sqlite"]
 single-postgresql = ["postgresql", "json-1", "uuid-0_8", "chrono-0_4", "array"]
 single-mysql = ["mysql", "json-1", "uuid-0_8", "chrono-0_4"]
 single-sqlite = ["sqlite", "json-1", "uuid-0_8", "chrono-0_4"]
 single-mssql = ["mssql"]
 
 postgresql = [
-    "rust_decimal/tokio-pg",
-    "native-tls",
-    "tokio-postgres",
-    "postgres-types",
-    "postgres-native-tls",
+    "sqlx/postgres",
     "array",
     "bytes",
-    "tokio",
     "bit-vec",
-    "lru-cache"
+    "ipnetwork"
 ]
 
 pooled = ["mobc"]
-sqlite = ["rusqlite", "libsqlite3-sys", "tokio/sync"]
+sqlite = ["sqlx/sqlite"]
 json-1 = ["serde_json", "base64"]
-uuid-0_8 = ["uuid"]
+uuid-0_8 = ["uuid", "sqlx/uuid"]
 chrono-0_4 = ["chrono"]
-mysql = ["mysql_async", "tokio"]
-mssql = ["tiberius", "uuid-0_8", "chrono-0_4", "tokio-util"]
+mysql = ["sqlx/mysql"]
+mssql = ["tiberius", "uuid-0_8", "chrono-0_4"]
 tracing-log = ["tracing", "tracing-core"]
 array = []
 serde-support = ["serde", "chrono/serde"]
 
+runtime-tokio = ["tokio", "tokio-util", "sqlx/runtime-tokio", "tiberius/sql-browser-tokio", "mobc/tokio"]
+runtime-async-std = ["async-std", "sqlx/runtime-async-std", "tiberius/sql-browser-async-std", "mobc/async-std"]
+
 [dependencies]
 url = "2.1"
 metrics = "0.12"
 percent-encoding = "2"
 once_cell = "1.3"
 num_cpus = "1.12"
-rust_decimal = { git = "https://github.com/pimeys/rust-decimal", branch = "pgbouncer-mode" }
+rust_decimal = "1.7"
 futures = "0.3"
 thiserror = "1.0"
 async-trait = "0.1"
 hex = "0.4"
+bigdecimal = "0.1"
 
 uuid = { version = "0.8", optional = true }
 chrono = { version = "0.4", optional = true }
 serde_json = { version = "1.0.48", optional = true }
 base64 = { version = "0.11.0", optional = true }
 lru-cache = { version = "0.1", optional = true }
 
-rusqlite = { version = "0.21", features = ["chrono", "bundled"], optional = true }
-libsqlite3-sys = { version = "0.17", default-features = false, features = ["bundled"], optional = true }
-
-native-tls = { version = "0.2", optional = true }
-
-mysql_async = { version = "0.23", optional = true }
-
 log = { version = "0.4", features = ["release_max_level_trace"] }
 tracing = { version = "0.1", optional = true }
 tracing-core = { version = "0.1", optional = true }
 
-mobc = { version = "0.5.7", optional = true }
+mobc = { version = "0.5.7", optional = true, default-features = false, features = ["unstable"] }
 bytes = { version = "0.5", optional = true }
 tokio = { version = "0.2", features = ["rt-threaded", "macros", "sync"], optional = true}
 tokio-util = { version = "0.3", features = ["compat"], optional = true }
+async-std = { version = "1.6.2", optional = true }
 serde = { version = "1.0", optional = true }
 bit-vec = { version = "0.6.1", optional = true }
+ipnetwork = { version = "0.16.0", optional = true }
+either = "1.5.3"
 
 [dev-dependencies]
 tokio = { version = "0.2", features = ["rt-threaded", "macros"]}
@@ -106,24 +101,12 @@ test-setup = { path = "test-setup" }
 paste = "1.0"
 
 [dependencies.tiberius]
-git = "https://github.com/prisma/tiberius"
-optional = true
-features = ["rust_decimal", "sql-browser-tokio", "chrono"]
-branch = "pgbouncer-mode-hack"
-
-[dependencies.tokio-postgres]
-git = "https://github.com/pimeys/rust-postgres"
-features = ["with-uuid-0_8", "with-chrono-0_4", "with-serde_json-1", "with-bit-vec-0_6"]
-branch = "pgbouncer-mode"
-optional = true
-
-[dependencies.postgres-types]
-git = "https://github.com/pimeys/rust-postgres"
-features = ["with-uuid-0_8", "with-chrono-0_4", "with-serde_json-1", "with-bit-vec-0_6"]
-branch = "pgbouncer-mode"
+version = "0.4"
 optional = true
+features = ["rust_decimal", "chrono"]
 
-[dependencies.postgres-native-tls]
-git = "https://github.com/pimeys/rust-postgres"
+[dependencies.sqlx]
+path = "../sqlx"
+default_features = false
+features = ["decimal", "json", "chrono", "ipnetwork", "bit-vec"]
 optional = true
-branch = "pgbouncer-mode"

src/ast/values.rs

@@ -332,6 +332,7 @@ impl<'a> Value<'a> {
 
     /// Transforms the `Value` to a `String` if it's text,
     /// otherwise `None`.
+
     pub fn into_string(self) -> Option<String> {
         match self {
             Value::Text(Some(cow)) => Some(cow.into_owned()),
@@ -366,6 +367,15 @@ impl<'a> Value<'a> {
         }
     }
 
+    /// Returns a cloned `Vec<u8>` if the value is text or a byte slice, otherwise `None`.
+    pub fn into_bytes(self) -> Option<Vec<u8>> {
+        match self {
+            Value::Text(Some(cow)) => Some(cow.into_owned().into()),
+            Value::Bytes(Some(cow)) => Some(cow.into_owned()),
+            _ => None,
+        }
+    }
+
     /// `true` if the `Value` is an integer.
     pub fn is_integer(&self) -> bool {
         match self {
@@ -485,6 +495,7 @@ impl<'a> Value<'a> {
     pub fn as_date(&self) -> Option<NaiveDate> {
         match self {
             Value::Date(dt) => dt.clone(),
+            Value::DateTime(dt) => dt.map(|dt| dt.date().naive_utc()),
             _ => None,
         }
     }
@@ -503,6 +514,7 @@ impl<'a> Value<'a> {
     pub fn as_time(&self) -> Option<NaiveTime> {
         match self {
             Value::Time(time) => time.clone(),
+            Value::DateTime(dt) => dt.map(|dt| dt.time()),
             _ => None,
         }
     }

src/connector.rs

@@ -9,10 +9,12 @@
 //! implement the [Queryable](trait.Queryable.html) trait for generalized
 //! querying interface.
 
+mod bind;
 mod connection_info;
 pub(crate) mod metrics;
 mod queryable;
 mod result_set;
+mod timeout;
 mod transaction;
 mod type_identifier;
 
@@ -25,16 +27,15 @@ pub(crate) mod postgres;
 #[cfg(feature = "sqlite")]
 pub(crate) mod sqlite;
 
-#[cfg(feature = "mysql")]
-pub use self::mysql::*;
 #[cfg(feature = "postgresql")]
 pub use self::postgres::*;
 pub use self::result_set::*;
 pub use connection_info::*;
 #[cfg(feature = "mssql")]
 pub use mssql::*;
+#[cfg(feature = "mysql")]
+pub use mysql::*;
 pub use queryable::*;
 #[cfg(feature = "sqlite")]
 pub use sqlite::*;
 pub use transaction::*;
-pub(crate) use type_identifier::*;

src/connector/bind.rs

@@ -0,0 +1,11 @@
+use crate::ast::Value;
+use sqlx::Database;
+
+pub trait Bind<'a, DB>
+where
+    DB: Database,
+{
+    fn bind_value(self, value: Value<'a>, type_info: Option<&DB::TypeInfo>) -> crate::Result<Self>
+    where
+        Self: Sized;
+}

src/connector/metrics.rs

@@ -40,3 +40,45 @@ where
 
     res
 }
+
+pub(crate) async fn query_new<'a, F, T, U>(
+    tag: &'static str,
+    query: &'a str,
+    params: Vec<Value<'a>>,
+    f: F,
+) -> crate::Result<T>
+where
+    F: FnOnce(Vec<Value<'a>>) -> U + 'a,
+    U: Future<Output = crate::Result<T>> + 'a,
+{
+    if *crate::LOG_QUERIES {
+        let start = Instant::now();
+        let res = f(params.clone()).await;
+        let end = Instant::now();
+
+        #[cfg(not(feature = "tracing-log"))]
+        {
+            info!(
+                "query: \"{}\", params: {} (in {}ms)",
+                query,
+                Params(&params),
+                start.elapsed().as_millis(),
+            );
+        }
+        #[cfg(feature = "tracing-log")]
+        {
+            tracing::info!(
+                query,
+                item_type = "query",
+                params = %Params(&params),
+                duration_ms = start.elapsed().as_millis() as u64,
+            )
+        }
+
+        timing!(format!("{}.query.time", tag), start, end);
+
+        res
+    } else {
+        f(params).await
+    }
+}