1use crate::{
4 connection::OdbcExecution, DataTypeExt, Odbc, OdbcArgumentValue, OdbcArguments, OdbcColumn,
5 OdbcConnectOptions, OdbcConnection, OdbcQueryResult, OdbcTransactionManager, OdbcTypeInfo,
6};
7use futures_core::future::BoxFuture;
8use futures_core::stream::BoxStream;
9use futures_util::{future, stream, FutureExt, StreamExt};
10use sqlx_core::any::driver::AnyDriver;
11use sqlx_core::any::{
12 AnyArguments, AnyColumn, AnyConnectOptions, AnyConnectionBackend, AnyQueryResult, AnyRow,
13 AnyStatement, AnyTypeInfo, AnyTypeInfoKind, AnyValueKind,
14};
15use sqlx_core::column::Column;
16use sqlx_core::connection::{ConnectOptions, Connection};
17use sqlx_core::database::Database;
18use sqlx_core::ext::ustr::UStr;
19use sqlx_core::row::Row;
20use sqlx_core::sql_str::SqlStr;
21use sqlx_core::statement::Statement;
22use sqlx_core::transaction::TransactionManager;
23use sqlx_core::{Either, HashMap};
24use std::sync::Arc;
25
26pub const DRIVER: AnyDriver = AnyDriver::without_migrate::<Odbc>();
28
29impl AnyConnectionBackend for OdbcConnection {
30 fn name(&self) -> &str {
31 <Odbc as Database>::NAME
32 }
33
34 fn close(self: Box<Self>) -> BoxFuture<'static, sqlx_core::Result<()>> {
35 Connection::close(*self).boxed()
36 }
37
38 fn close_hard(self: Box<Self>) -> BoxFuture<'static, sqlx_core::Result<()>> {
39 Connection::close_hard(*self).boxed()
40 }
41
42 fn ping(&mut self) -> BoxFuture<'_, sqlx_core::Result<()>> {
43 Connection::ping(self).boxed()
44 }
45
46 fn begin(&mut self, statement: Option<SqlStr>) -> BoxFuture<'_, sqlx_core::Result<()>> {
47 OdbcTransactionManager::begin(self, statement).boxed()
48 }
49
50 fn commit(&mut self) -> BoxFuture<'_, sqlx_core::Result<()>> {
51 OdbcTransactionManager::commit(self).boxed()
52 }
53
54 fn rollback(&mut self) -> BoxFuture<'_, sqlx_core::Result<()>> {
55 OdbcTransactionManager::rollback(self).boxed()
56 }
57
58 fn start_rollback(&mut self) {
59 OdbcTransactionManager::start_rollback(self);
60 }
61
62 fn get_transaction_depth(&self) -> usize {
63 OdbcTransactionManager::get_transaction_depth(self)
64 }
65
66 fn shrink_buffers(&mut self) {
67 Connection::shrink_buffers(self);
68 }
69
70 fn flush(&mut self) -> BoxFuture<'_, sqlx_core::Result<()>> {
71 Connection::flush(self).boxed()
72 }
73
74 fn should_flush(&self) -> bool {
75 Connection::should_flush(self)
76 }
77
78 fn fetch_many(
79 &mut self,
80 query: SqlStr,
81 _persistent: bool,
82 arguments: Option<AnyArguments>,
83 ) -> BoxStream<'_, sqlx_core::Result<Either<AnyQueryResult, AnyRow>>> {
84 let arguments = arguments.map(map_arguments);
85
86 stream::once(async move { self.run_blocking_sql(query.as_str(), arguments.as_ref()) })
87 .map(|result| match result {
88 Ok(OdbcExecution::Done(result)) => {
89 stream::once(future::ready(Ok(Either::Left(map_result(result))))).boxed()
90 }
91 Ok(OdbcExecution::Rows(rows)) => stream::iter(
92 rows.into_iter()
93 .map(|row| {
94 let column_names = column_names(row.columns());
95 AnyRow::map_from(&row, column_names).map(Either::Right)
96 })
97 .chain(std::iter::once(Ok(Either::Left(map_result(
98 OdbcQueryResult::new(0),
99 ))))),
100 )
101 .boxed(),
102 Err(error) => stream::once(future::ready(Err(error))).boxed(),
103 })
104 .flatten()
105 .boxed()
106 }
107
108 fn fetch_optional(
109 &mut self,
110 query: SqlStr,
111 _persistent: bool,
112 arguments: Option<AnyArguments>,
113 ) -> BoxFuture<'_, sqlx_core::Result<Option<AnyRow>>> {
114 let arguments = arguments.map(map_arguments);
115
116 Box::pin(async move {
117 match self.run_blocking_sql(query.as_str(), arguments.as_ref())? {
118 OdbcExecution::Done(_) => Ok(None),
119 OdbcExecution::Rows(rows) => rows
120 .into_iter()
121 .next()
122 .map(|row| {
123 let column_names = column_names(row.columns());
124 AnyRow::map_from(&row, column_names)
125 })
126 .transpose(),
127 }
128 })
129 }
130
131 fn prepare_with<'c, 'q: 'c>(
132 &'c mut self,
133 sql: SqlStr,
134 _parameters: &[AnyTypeInfo],
135 ) -> BoxFuture<'c, sqlx_core::Result<AnyStatement>> {
136 Box::pin(async move {
137 let statement = self.prepare_blocking(sql)?;
138 let column_names = column_names(statement.columns());
139 AnyStatement::try_from_statement(statement, column_names)
140 })
141 }
142}
143
144impl<'a> TryFrom<&'a AnyConnectOptions> for OdbcConnectOptions {
145 type Error = sqlx_core::Error;
146
147 fn try_from(options: &'a AnyConnectOptions) -> Result<Self, Self::Error> {
148 let mut options_out = OdbcConnectOptions::from_url(&options.database_url)?;
149 options_out.log_statements = options.log_settings.statements_level;
150 options_out.log_slow_statements = options.log_settings.slow_statements_level;
151 options_out.log_slow_statement_duration = options.log_settings.slow_statements_duration;
152 Ok(options_out)
153 }
154}
155
156impl<'a> TryFrom<&'a OdbcTypeInfo> for AnyTypeInfo {
157 type Error = sqlx_core::Error;
158
159 fn try_from(type_info: &'a OdbcTypeInfo) -> Result<Self, Self::Error> {
160 let kind = match type_info.data_type() {
161 odbc_api::DataType::Unknown => AnyTypeInfoKind::Null,
162 odbc_api::DataType::Bit => AnyTypeInfoKind::Bool,
163 odbc_api::DataType::TinyInt | odbc_api::DataType::SmallInt => AnyTypeInfoKind::SmallInt,
164 odbc_api::DataType::Integer => AnyTypeInfoKind::Integer,
165 odbc_api::DataType::BigInt => AnyTypeInfoKind::BigInt,
166 odbc_api::DataType::Real => AnyTypeInfoKind::Real,
167 odbc_api::DataType::Float { .. } | odbc_api::DataType::Double => {
168 AnyTypeInfoKind::Double
169 }
170 data_type if data_type.accepts_character_data() => AnyTypeInfoKind::Text,
171 data_type if data_type.accepts_binary_data() => AnyTypeInfoKind::Blob,
172 data_type => {
173 return Err(sqlx_core::Error::AnyDriverError(
174 format!("Any driver does not support the ODBC type {data_type:?}").into(),
175 ));
176 }
177 };
178
179 Ok(AnyTypeInfo { kind })
180 }
181}
182
183impl<'a> TryFrom<&'a OdbcColumn> for AnyColumn {
184 type Error = sqlx_core::Error;
185
186 fn try_from(column: &'a OdbcColumn) -> Result<Self, Self::Error> {
187 let type_info = AnyTypeInfo::try_from(column.type_info()).map_err(|error| {
188 sqlx_core::Error::ColumnDecode {
189 index: column.name().to_owned(),
190 source: error.into(),
191 }
192 })?;
193
194 Ok(Self {
195 ordinal: column.ordinal(),
196 name: UStr::new(column.name()),
197 type_info,
198 })
199 }
200}
201
202fn map_arguments(arguments: AnyArguments) -> OdbcArguments {
203 let mut out = OdbcArguments::default();
204
205 for value in arguments.values.0 {
206 out.add_value(match value {
207 AnyValueKind::Null(kind) => OdbcArgumentValue::Null(any_type_to_odbc(kind)),
208 AnyValueKind::Bool(value) => OdbcArgumentValue::Bit(value),
209 AnyValueKind::SmallInt(value) => OdbcArgumentValue::Int(i64::from(value)),
210 AnyValueKind::Integer(value) => OdbcArgumentValue::Int(i64::from(value)),
211 AnyValueKind::BigInt(value) => OdbcArgumentValue::Int(value),
212 AnyValueKind::Real(value) => OdbcArgumentValue::Float(f64::from(value)),
213 AnyValueKind::Double(value) => OdbcArgumentValue::Float(value),
214 AnyValueKind::Text(value) => OdbcArgumentValue::Text(value.to_string()),
215 AnyValueKind::TextSlice(value) => OdbcArgumentValue::Text(value.to_string()),
216 AnyValueKind::Blob(value) => OdbcArgumentValue::Bytes(value.to_vec()),
217 _ => unreachable!("unhandled Any argument value"),
218 });
219 }
220
221 out
222}
223
224fn any_type_to_odbc(kind: AnyTypeInfoKind) -> OdbcTypeInfo {
225 OdbcTypeInfo::new(match kind {
226 AnyTypeInfoKind::Null => odbc_api::DataType::Unknown,
227 AnyTypeInfoKind::Bool => odbc_api::DataType::Bit,
228 AnyTypeInfoKind::SmallInt => odbc_api::DataType::SmallInt,
229 AnyTypeInfoKind::Integer => odbc_api::DataType::Integer,
230 AnyTypeInfoKind::BigInt => odbc_api::DataType::BigInt,
231 AnyTypeInfoKind::Real => odbc_api::DataType::Real,
232 AnyTypeInfoKind::Double => odbc_api::DataType::Double,
233 AnyTypeInfoKind::Text => odbc_api::DataType::WVarchar { length: None },
234 AnyTypeInfoKind::Blob => odbc_api::DataType::Varbinary { length: None },
235 })
236}
237
238fn map_result(result: OdbcQueryResult) -> AnyQueryResult {
239 AnyQueryResult {
240 rows_affected: result.rows_affected(),
241 last_insert_id: None,
242 }
243}
244
245fn column_names(columns: &[OdbcColumn]) -> Arc<HashMap<UStr, usize>> {
246 Arc::new(
247 columns
248 .iter()
249 .map(|column| (UStr::new(column.name()), column.ordinal()))
250 .collect(),
251 )
252}
253
254#[cfg(test)]
255mod tests {
256 use super::*;
257
258 #[test]
259 fn maps_stable_odbc_types_to_any_types() {
260 assert_eq!(
261 AnyTypeInfo::try_from(&OdbcTypeInfo::new(odbc_api::DataType::Bit))
262 .unwrap()
263 .kind(),
264 AnyTypeInfoKind::Bool
265 );
266 assert_eq!(
267 AnyTypeInfo::try_from(&OdbcTypeInfo::new(odbc_api::DataType::Integer))
268 .unwrap()
269 .kind(),
270 AnyTypeInfoKind::Integer
271 );
272 assert_eq!(
273 AnyTypeInfo::try_from(&OdbcTypeInfo::new(odbc_api::DataType::WVarchar {
274 length: None
275 }))
276 .unwrap()
277 .kind(),
278 AnyTypeInfoKind::Text
279 );
280 assert_eq!(
281 AnyTypeInfo::try_from(&OdbcTypeInfo::new(odbc_api::DataType::Varbinary {
282 length: None
283 }))
284 .unwrap()
285 .kind(),
286 AnyTypeInfoKind::Blob
287 );
288 }
289
290 #[test]
291 fn rejects_unstable_odbc_types_for_any_mapping() {
292 assert!(matches!(
293 AnyTypeInfo::try_from(&OdbcTypeInfo::new(odbc_api::DataType::Timestamp {
294 precision: 6
295 })),
296 Err(sqlx_core::Error::AnyDriverError(_))
297 ));
298 }
299}