Remove remaining panics in AnyConnectionBackend implementations

Author: FSMaxB

sqlx-mysql/src/any.rs

@@ -6,7 +6,7 @@ use crate::{
 use either::Either;
 use futures_core::future::BoxFuture;
 use futures_core::stream::BoxStream;
-use futures_util::{StreamExt, TryFutureExt, TryStreamExt};
+use futures_util::{stream, StreamExt, TryFutureExt, TryStreamExt};
 use sqlx_core::any::{
     Any, AnyArguments, AnyColumn, AnyConnectOptions, AnyConnectionBackend, AnyQueryResult, AnyRow,
     AnyStatement, AnyTypeInfo, AnyTypeInfoKind,
@@ -16,6 +16,7 @@ use sqlx_core::database::Database;
 use sqlx_core::describe::Describe;
 use sqlx_core::executor::Executor;
 use sqlx_core::transaction::TransactionManager;
+use std::future;
 
 sqlx_core::declare_driver_with_optional_migrate!(DRIVER = MySql);
 
@@ -77,14 +78,15 @@ impl AnyConnectionBackend for MySqlConnection {
         arguments: Option<AnyArguments<'q>>,
     ) -> BoxStream<'q, sqlx_core::Result<Either<AnyQueryResult, AnyRow>>> {
         let persistent = arguments.is_some();
-        let args = arguments
-            .as_ref()
-            .map(AnyArguments::convert_to)
-            .transpose()
-            .expect("Failed to encode arguments");
+        let arguments = match arguments.as_ref().map(AnyArguments::convert_to).transpose() {
+            Ok(arguments) => arguments,
+            Err(error) => {
+                return stream::once(future::ready(Err(sqlx_core::Error::Encode(error)))).boxed()
+            }
+        };
 
         Box::pin(
-            self.run(query, args, persistent)
+            self.run(query, arguments, persistent)
                 .try_flatten_stream()
                 .map(|res| {
                     Ok(match res? {
@@ -101,14 +103,15 @@ impl AnyConnectionBackend for MySqlConnection {
         arguments: Option<AnyArguments<'q>>,
     ) -> BoxFuture<'q, sqlx_core::Result<Option<AnyRow>>> {
         let persistent = arguments.is_some();
-        let args = arguments
+        let arguments = arguments
             .as_ref()
             .map(AnyArguments::convert_to)
             .transpose()
-            .expect("Failed to encode arguments");
+            .map_err(sqlx_core::Error::Encode);
 
         Box::pin(async move {
-            let stream = self.run(query, args, persistent).await?;
+            let arguments = arguments?;
+            let stream = self.run(query, arguments, persistent).await?;
             futures_util::pin_mut!(stream);
 
             if let Some(Either::Right(row)) = stream.try_next().await? {

sqlx-postgres/src/any.rs

@@ -4,7 +4,8 @@ use crate::{
 };
 use futures_core::future::BoxFuture;
 use futures_core::stream::BoxStream;
-use futures_util::{StreamExt, TryFutureExt, TryStreamExt};
+use futures_util::{stream, FutureExt, StreamExt, TryFutureExt, TryStreamExt};
+use std::future;
 
 pub use sqlx_core::any::*;
 
@@ -76,14 +77,15 @@ impl AnyConnectionBackend for PgConnection {
         arguments: Option<AnyArguments<'q>>,
     ) -> BoxStream<'q, sqlx_core::Result<Either<AnyQueryResult, AnyRow>>> {
         let persistent = arguments.is_some();
-        let args = arguments
-            .as_ref()
-            .map(AnyArguments::convert_to)
-            .transpose()
-            .expect("Failed to encode arguments");
+        let arguments = match arguments.as_ref().map(AnyArguments::convert_to).transpose() {
+            Ok(arguments) => arguments,
+            Err(error) => {
+                return stream::once(future::ready(Err(sqlx_core::Error::Encode(error)))).boxed()
+            }
+        };
 
         Box::pin(
-            self.run(query, args, 0, persistent, None)
+            self.run(query, arguments, 0, persistent, None)
                 .try_flatten_stream()
                 .map(
                     move |res: sqlx_core::Result<Either<PgQueryResult, PgRow>>| match res? {
@@ -100,14 +102,15 @@ impl AnyConnectionBackend for PgConnection {
         arguments: Option<AnyArguments<'q>>,
     ) -> BoxFuture<'q, sqlx_core::Result<Option<AnyRow>>> {
         let persistent = arguments.is_some();
-        let args = arguments
+        let arguments = arguments
             .as_ref()
             .map(AnyArguments::convert_to)
             .transpose()
-            .expect("Failed to encode arguments");
+            .map_err(sqlx_core::Error::Encode);
 
         Box::pin(async move {
-            let stream = self.run(query, args, 1, persistent, None).await?;
+            let arguments = arguments?;
+            let stream = self.run(query, arguments, 1, persistent, None).await?;
             futures_util::pin_mut!(stream);
 
             if let Some(Either::Right(row)) = stream.try_next().await? {