make macros work again

sqlx-core/src/arguments.rs

@@ -18,30 +18,3 @@ pub trait Arguments: Send + Sized + Default + 'static {
         T: Type<Self::Database>,
         T: Encode<Self::Database>;
 }
-
-pub trait IntoArguments<DB>
-where
-    DB: Database,
-{
-    fn into_arguments(self) -> DB::Arguments;
-}
-
-impl<A> IntoArguments<A::Database> for A
-where
-    A: Arguments,
-    A::Database: Database<Arguments = Self> + Sized,
-{
-    #[inline]
-    fn into_arguments(self) -> Self {
-        self
-    }
-}
-
-#[doc(hidden)]
-pub struct ImmutableArguments<DB: Database>(pub DB::Arguments);
-
-impl<DB: Database> IntoArguments<DB> for ImmutableArguments<DB> {
-    fn into_arguments(self) -> <DB as Database>::Arguments {
-        self.0
-    }
-}

sqlx-core/src/cursor.rs

@@ -49,17 +49,4 @@ where
 
     /// Fetch the next row in the result. Returns `None` if there are no more rows.
     fn next(&mut self) -> BoxFuture<crate::Result<Option<<DB as HasRow>::Row>>>;
-
-    /// Map the `Row`s in this result to a different type, returning a [`Stream`] of the results.
-    fn map<'a, T, F>(self, f: F) -> BoxStream<'a, crate::Result<T>>
-    where
-        Self: Sized,
-        F: Send + Sync + 'static,
-        T: Send + Unpin + 'static,
-        F: for<'r> Fn(<DB as HasRow<'r>>::Row) -> T,
-    {
-        unimplemented!(
-            "Cursor::map is currently supplied by inherent methods to work around not having GATs"
-        )
-    }
 }

sqlx-core/src/executor.rs

@@ -61,3 +61,18 @@ where
         (self, None)
     }
 }
+
+macro_rules! impl_execute_for_query {
+    ($db:ty) => {
+        impl<'q> $crate::executor::Execute<'q, $db> for $crate::query::Query<'q, $db> {
+            fn into_parts(
+                self,
+            ) -> (
+                &'q str,
+                Option<<$db as $crate::database::Database>::Arguments>,
+            ) {
+                (self.query, Some(self.arguments))
+            }
+        }
+    };
+}

sqlx-core/src/lib.rs

@@ -15,8 +15,10 @@ mod cache;
 mod connection;
 mod cursor;
 mod database;
+
+#[macro_use]
 mod executor;
-mod query;
+
 mod transaction;
 mod url;
 
@@ -32,6 +34,7 @@ pub mod decode;
 pub mod describe;
 pub mod encode;
 pub mod pool;
+pub mod query;
 pub mod types;
 
 #[macro_use]
@@ -53,7 +56,6 @@ pub use error::{Error, Result};
 pub use connection::{Connect, Connection};
 pub use cursor::Cursor;
 pub use executor::{Execute, Executor};
-pub use query::{query, Query};
 pub use transaction::Transaction;
 
 #[doc(inline)]

sqlx-core/src/mysql/executor.rs

@@ -338,3 +338,5 @@ impl Executor for MySqlConnection {
         Box::pin(self.describe(query))
     }
 }
+
+impl_execute_for_query!(MySql);

sqlx-core/src/postgres/cursor.rs

@@ -32,24 +32,6 @@ pub struct PgCursor<'c, 'q> {
     state: State<'c, 'q>,
 }
 
-impl<'c, 'q> PgCursor<'c, 'q> {
-    #[doc(hidden)]
-    pub fn map<'a, T, F>(mut self, f: F) -> impl Stream<Item = crate::Result<T>> + 'a + 'c
-    where
-        F: Send + Sync + 'static,
-        T: Send + Unpin + 'static,
-        F: for<'r> Fn(PgRow<'r>) -> T,
-        'q: 'c,
-        'c: 'a,
-    {
-        try_stream! {
-            while let Some(row) = next(&mut self).await? {
-                yield f(row);
-            }
-        }
-    }
-}
-
 impl<'c, 'q> Cursor<'c, 'q, Postgres> for PgCursor<'c, 'q> {
     #[doc(hidden)]
     fn from_pool<E>(pool: &Pool<PgConnection>, query: E) -> Self

sqlx-core/src/postgres/executor.rs

@@ -74,3 +74,5 @@ impl<'e> Executor<'e> for &'e mut super::PgConnection {
         self.execute(query)
     }
 }
+
+impl_execute_for_query!(Postgres);

sqlx-core/src/postgres/row.rs

@@ -56,3 +56,14 @@ impl<'c> RowIndex<'c, PgRow<'c>> for &'_ str {
 }
 
 // TODO: impl_from_row_for_row!(PgRow);
+
+impl<O: Unpin, F> crate::query::MapRow<Postgres> for F
+where
+    F: for<'c> FnMut(PgRow<'c>) -> crate::Result<O>,
+{
+    type Mapped = O;
+
+    fn map_row(&mut self, row: PgRow) -> crate::Result<O> {
+        (self)(row)
+    }
+}

sqlx-core/src/query.rs

@@ -1,42 +1,118 @@
+use std::future::Future;
+use std::marker::PhantomData;
+use std::mem;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+
+use async_stream::try_stream;
+use futures_core::stream::BoxStream;
+use futures_core::Stream;
+use futures_util::future::ready;
+use futures_util::ready;
+use futures_util::stream::try_unfold;
+use futures_util::TryFutureExt;
+use futures_util::TryStreamExt;
+
 use crate::arguments::Arguments;
-use crate::arguments::IntoArguments;
 use crate::cursor::Cursor;
 use crate::database::{Database, HasCursor, HasRow};
 use crate::encode::Encode;
 use crate::executor::{Execute, Executor};
 use crate::types::Type;
-use futures_core::stream::BoxStream;
-use futures_util::future::ready;
-use futures_util::TryFutureExt;
-use futures_util::TryStreamExt;
-use std::future::Future;
-use std::marker::PhantomData;
-use std::mem;
+use crate::{Error, FromRow};
+use futures_core::future::BoxFuture;
 
-/// Raw SQL query with bind parameters. Returned by [`query`].
-pub struct Query<'q, DB, T = <DB as Database>::Arguments>
+/// Raw SQL query with bind parameters. Returned by [`query`][crate::query::query].
+pub struct Query<'q, DB, A = <DB as Database>::Arguments>
 where
     DB: Database,
 {
-    query: &'q str,
-    arguments: T,
+    pub(crate) query: &'q str,
+    pub(crate) arguments: A,
     database: PhantomData<DB>,
 }
 
-impl<'q, DB, P> Execute<'q, DB> for Query<'q, DB, P>
+/// SQL query that will map its results to owned Rust types.
+///
+/// Returned by [Query::map], `query!()`, etc. Has most of the same methods as [Query] but
+/// the return types are changed to reflect the mapping. However, there is no equivalent of
+/// [Query::execute] as it doesn't make much sense to map the result type and then ignore it.
+pub struct Map<'q, DB, F, A = <DB as Database>::Arguments>
+where
+    DB: Database,
+{
+    query: Query<'q, DB, A>,
+    mapper: F,
+}
+
+#[doc(hidden)]
+pub struct ImmutableArguments<DB: Database>(DB::Arguments);
+
+// necessary because we can't have a blanket impl for `Query<'q, DB>`
+// the compiler thinks that `ImmutableArguments<DB>` could be `DB::Arguments` even though
+// that would be an infinitely recursive type
+impl<'q, DB> Execute<'q, DB> for Query<'q, DB, ImmutableArguments<DB>>
 where
     DB: Database,
-    P: IntoArguments<DB> + Send,
 {
     fn into_parts(self) -> (&'q str, Option<<DB as Database>::Arguments>) {
-        (self.query, Some(self.arguments.into_arguments()))
+        (self.query, Some(self.arguments.0))
     }
 }
 
-impl<'q, DB, P> Query<'q, DB, P>
+impl<'q, DB> Query<'q, DB>
 where
     DB: Database,
-    P: IntoArguments<DB> + Send,
+{
+    /// Bind a value for use with this SQL query.
+    pub fn bind<T>(mut self, value: T) -> Self
+    where
+        T: Type<DB>,
+        T: Encode<DB>,
+    {
+        self.arguments.add(value);
+        self
+    }
+
+    #[doc(hidden)]
+    pub fn bind_all(self, arguments: DB::Arguments) -> Query<'q, DB, ImmutableArguments<DB>> {
+        Query {
+            query: self.query,
+            arguments: ImmutableArguments(arguments),
+            database: PhantomData,
+        }
+    }
+}
+
+impl<'q, DB, A> Query<'q, DB, A>
+where
+    DB: Database,
+{
+    /// Map each row in the result to another type.
+    ///
+    /// The returned type has most of the same methods but does not have [`.execute()`][Query::execute].
+    ///
+    /// The mapping function returns [`crate::Result`] so [`Row::try_get`] can be used.
+    ///
+    /// Stylistically, we recommend placing this call after any [`.bind()`][Query::bind]
+    /// calls, just before [`.fetch()`][Query::fetch], etc.
+    ///
+    /// See also: [query_as].
+    pub fn map<F>(self, mapper: F) -> Map<'q, DB, F, A>
+    where
+        F: MapRow<DB>,
+    {
+        Map {
+            query: self,
+            mapper,
+        }
+    }
+}
+
+impl<'q, DB, A> Query<'q, DB, A>
+where
+    DB: Database,
+    Self: Execute<'q, DB>,
 {
     pub async fn execute<'e, E>(self, executor: E) -> crate::Result<u64>
     where
@@ -77,7 +153,7 @@ where
     }
 }
 
-impl<'q, DB> Query<'q, DB>
+impl<'q, DB, F> Map<'q, DB, F>
 where
     DB: Database,
 {
@@ -87,9 +163,107 @@ where
         T: Type<DB>,
         T: Encode<DB>,
     {
-        self.arguments.add(value);
+        self.query.arguments.add(value);
         self
     }
+
+    #[doc(hidden)]
+    pub fn bind_all(self, arguments: DB::Arguments) -> Map<'q, DB, F, ImmutableArguments<DB>> {
+        Map {
+            query: self.query.bind_all(arguments),
+            mapper: self.mapper,
+        }
+    }
+}
+
+impl<'q, DB, F, A> Map<'q, DB, F, A>
+where
+    DB: Database,
+    Query<'q, DB, A>: Execute<'q, DB>,
+    F: MapRow<DB>,
+{
+    /// Execute the query and get a [Stream] of the results, returning our mapped type.
+    pub fn fetch<'e: 'q, E>(
+        mut self,
+        executor: E,
+    ) -> impl Stream<Item = crate::Result<F::Mapped>> + 'e
+    where
+        'q: 'e,
+        E: Executor<'e, Database = DB> + 'e,
+        F: 'e,
+        F::Mapped: 'e,
+        A: 'e,
+    {
+        try_stream! {
+            let mut cursor = executor.execute(self.query);
+            while let Some(next) = cursor.next().await? {
+                let mapped = self.mapper.map_row(next)?;
+                yield mapped;
+            }
+        }
+    }
+
+    /// Get the first row in the result
+    pub async fn fetch_optional<'e, E>(mut self, executor: E) -> crate::Result<Option<F::Mapped>>
+    where
+        E: Executor<'e, Database = DB>,
+        'q: 'e,
+    {
+        // could be implemented in terms of `fetch()` but this avoids overhead from `try_stream!`
+        let mut cursor = executor.execute(self.query);
+        let mut mapper = self.mapper;
+        let val = cursor.next().await?;
+        val.map(|row| mapper.map_row(row)).transpose()
+    }
+
+    pub async fn fetch_one<'e, E>(self, executor: E) -> crate::Result<F::Mapped>
+    where
+        E: Executor<'e, Database = DB>,
+        'q: 'e,
+    {
+        self.fetch_optional(executor)
+            .and_then(|row| match row {
+                Some(row) => ready(Ok(row)),
+                None => ready(Err(crate::Error::NotFound)),
+            })
+            .await
+    }
+
+    pub async fn fetch_all<'e, E>(mut self, executor: E) -> crate::Result<Vec<F::Mapped>>
+    where
+        E: Executor<'e, Database = DB>,
+        'q: 'e,
+    {
+        let mut cursor = executor.execute(self.query);
+        let mut out = vec![];
+
+        while let Some(row) = cursor.next().await? {
+            out.push(self.mapper.map_row(row)?);
+        }
+
+        Ok(out)
+    }
+}
+
+/// A (hopefully) temporary workaround for an internal compiler error (ICE) involving higher-ranked
+/// trait bounds (HRTBs), associated types and closures.
+///
+/// See https://github.com/rust-lang/rust/issues/62529
+pub trait MapRow<DB: Database> {
+    type Mapped: Unpin;
+
+    fn map_row(&mut self, row: <DB as HasRow>::Row) -> crate::Result<Self::Mapped>;
+}
+
+impl<O: Unpin, DB> MapRow<DB> for for<'c> fn(<DB as HasRow<'c>>::Row) -> crate::Result<O>
+where
+    DB: Database,
+{
+    type Mapped = O;
+
+    fn map_row(&mut self, row: <DB as HasRow>::Row) -> crate::Result<O> {
+        (self)(row)
+    }
 }
 
 /// Construct a raw SQL query that can be chained to bind parameters and executed.
@@ -103,3 +277,13 @@ where
         query: sql,
     }
 }
+
+pub fn query_as<T, DB>(
+    sql: &str,
+) -> Map<DB, for<'c> fn(<DB as HasRow<'c>>::Row) -> crate::Result<T>>
+where
+    DB: Database,
+    T: Unpin + for<'c> FromRow<'c, <DB as HasRow<'c>>::Row>,
+{
+    query(sql).map(|row| Ok(T::from_row(row)))
+}