sqlx_core_oldapi/odbc/connection/
executor.rs

1use crate::describe::Describe;
2use crate::error::Error;
3use crate::executor::{Execute, Executor};
4use crate::odbc::{Odbc, OdbcConnection, OdbcQueryResult, OdbcRow, OdbcStatement, OdbcTypeInfo};
5use either::Either;
6use futures_core::future::BoxFuture;
7use futures_core::stream::BoxStream;
8use futures_util::{TryFutureExt, TryStreamExt};
9
10impl<'c> Executor<'c> for &'c mut OdbcConnection {
11    type Database = Odbc;
12
13    fn fetch_many<'e, 'q: 'e, E>(
14        self,
15        mut query: E,
16    ) -> BoxStream<'e, Result<Either<OdbcQueryResult, OdbcRow>, Error>>
17    where
18        'c: 'e,
19        E: Execute<'q, Self::Database> + 'q,
20    {
21        let sql = query.sql();
22        let args = query.take_arguments();
23        let persistent = query.persistent();
24
25        Box::pin(
26            self.execute_stream(sql, args, persistent)
27                .map_ok(flume::Receiver::into_stream)
28                .try_flatten_stream(),
29        )
30    }
31
32    fn fetch_optional<'e, 'q: 'e, E>(
33        self,
34        query: E,
35    ) -> BoxFuture<'e, Result<Option<OdbcRow>, Error>>
36    where
37        'c: 'e,
38        E: Execute<'q, Self::Database> + 'q,
39    {
40        let mut stream = self.fetch_many(query);
41
42        Box::pin(async move {
43            while let Some(step) = stream.try_next().await? {
44                if let Either::Right(row) = step {
45                    return Ok(Some(row));
46                }
47            }
48
49            Ok(None)
50        })
51    }
52
53    fn prepare_with<'e, 'q: 'e>(
54        self,
55        sql: &'q str,
56        _parameters: &'e [OdbcTypeInfo],
57    ) -> BoxFuture<'e, Result<OdbcStatement<'q>, Error>>
58    where
59        'c: 'e,
60    {
61        Box::pin(async move { self.prepare(sql).await })
62    }
63
64    #[doc(hidden)]
65    fn describe<'e, 'q: 'e>(self, sql: &'q str) -> BoxFuture<'e, Result<Describe<Odbc>, Error>>
66    where
67        'c: 'e,
68    {
69        Box::pin(async move {
70            let statement = self.describe_statement(sql).await?;
71            let nullable = vec![None; statement.metadata.columns.len()];
72
73            Ok(Describe {
74                columns: statement.metadata.columns,
75                parameters: Some(Either::Right(statement.metadata.parameters)),
76                nullable,
77            })
78        })
79    }
80}
sqlx_core_oldapi/odbc/connection/executor.rs

sqlx_core_oldapi/odbc/connection/
executor.rs
