sqlx_cloudflare_do/
executor.rs

1use std::future;
2
3use futures_core::future::BoxFuture;
4use futures_core::stream::BoxStream;
5use futures_util::{stream, FutureExt, StreamExt, TryFutureExt};
6use sqlx_core::describe::Describe;
7use sqlx_core::error::Error;
8use sqlx_core::executor::{Execute, Executor};
9use sqlx_core::sql_str::SqlStr;
10use sqlx_core::Either;
11
12use crate::error::unsupported;
13use crate::{js, Do, DoArguments, DoConnection, DoQueryResult, DoRow, DoStatement, DoTypeInfo};
14
15// `sql.exec` is synchronous, so each future here finishes on its first poll
16// and holds nothing across an `await` -- which is why none of them needs
17// D1's `SendFuture`: they are `Send` because `DoConnection` is `Sync`. They
18// still run the query when polled, not when built, as sqlx's futures do.
19impl<'c> Executor<'c> for &'c DoConnection {
20    type Database = Do;
21
22    /// Reports `rows_affected` and `last_insert_rowid` but no rows.
23    fn execute<'e, 'q: 'e, E>(self, query: E) -> BoxFuture<'e, Result<DoQueryResult, Error>>
24    where
25        'c: 'e,
26        E: 'q + Execute<'q, Do>,
27    {
28        async move {
29            let (sql, arguments) = take(query)?;
30            js::run(&self.sql, sql.as_str(), arguments_of(arguments.as_ref()))
31        }
32        .boxed()
33    }
34
35    fn execute_many<'e, 'q: 'e, E>(self, query: E) -> BoxStream<'e, Result<DoQueryResult, Error>>
36    where
37        'c: 'e,
38        E: 'q + Execute<'q, Do>,
39    {
40        stream::once(self.execute(query)).boxed()
41    }
42
43    /// Yields the rows only, never a `DoQueryResult`. Use
44    /// [`execute`](Executor::execute) for what a statement changed.
45    fn fetch_many<'e, 'q: 'e, E>(
46        self,
47        query: E,
48    ) -> BoxStream<'e, Result<Either<DoQueryResult, DoRow>, Error>>
49    where
50        'c: 'e,
51        E: 'q + Execute<'q, Do>,
52    {
53        async move {
54            let (sql, arguments) = take(query)?;
55            js::fetch(
56                &self.sql,
57                sql.as_str(),
58                arguments_of(arguments.as_ref()),
59                None,
60            )
61            .map(|(rows, _)| rows)
62        }
63        .map_ok(|rows| stream::iter(rows.into_iter().map(|row| Ok(Either::Right(row)))))
64        .try_flatten_stream()
65        .boxed()
66    }
67
68    fn fetch_optional<'e, 'q: 'e, E>(self, query: E) -> BoxFuture<'e, Result<Option<DoRow>, Error>>
69    where
70        'c: 'e,
71        E: 'q + Execute<'q, Do>,
72    {
73        async move {
74            let (sql, arguments) = take(query)?;
75            let (rows, _) = js::fetch(
76                &self.sql,
77                sql.as_str(),
78                arguments_of(arguments.as_ref()),
79                Some(1),
80            )?;
81            Ok(rows.into_iter().next())
82        }
83        .boxed()
84    }
85
86    fn prepare_with<'e>(
87        self,
88        sql: SqlStr,
89        _parameters: &'e [DoTypeInfo],
90    ) -> BoxFuture<'e, Result<DoStatement, Error>>
91    where
92        'c: 'e,
93    {
94        future::ready(Ok(DoStatement { sql })).boxed()
95    }
96
97    /// Always fails: `sql.exec` cannot describe a statement without running
98    /// it. Only sqlx's `query!` macros need this, and they cannot target a
99    /// third-party driver anyway.
100    fn describe<'e>(self, _sql: SqlStr) -> BoxFuture<'e, Result<Describe<Do>, Error>>
101    where
102        'c: 'e,
103    {
104        future::ready(Err(unsupported("statements cannot be described"))).boxed()
105    }
106}
107
108/// `query`'s SQL and its arguments, bound.
109fn take<'q, E>(mut query: E) -> Result<(SqlStr, Option<DoArguments>), Error>
110where
111    E: Execute<'q, Do>,
112{
113    let arguments = query.take_arguments().map_err(Error::Encode)?;
114    Ok((query.sql(), arguments))
115}
116
117fn arguments_of(arguments: Option<&DoArguments>) -> &[crate::DoArgumentValue] {
118    arguments.map_or(&[][..], DoArguments::values)
119}
120
121// What sqlx code actually passes -- `&mut conn` -- running on the shared
122// implementation above. `&'c mut` gives up its uniqueness to become `&'c`,
123// which is all a query needs.
124impl<'c> Executor<'c> for &'c mut DoConnection {
125    type Database = Do;
126
127    fn execute<'e, 'q: 'e, E>(self, query: E) -> BoxFuture<'e, Result<DoQueryResult, Error>>
128    where
129        'c: 'e,
130        E: 'q + Execute<'q, Do>,
131    {
132        (&*self).execute(query)
133    }
134
135    fn execute_many<'e, 'q: 'e, E>(self, query: E) -> BoxStream<'e, Result<DoQueryResult, Error>>
136    where
137        'c: 'e,
138        E: 'q + Execute<'q, Do>,
139    {
140        (&*self).execute_many(query)
141    }
142
143    fn fetch_many<'e, 'q: 'e, E>(
144        self,
145        query: E,
146    ) -> BoxStream<'e, Result<Either<DoQueryResult, DoRow>, Error>>
147    where
148        'c: 'e,
149        E: 'q + Execute<'q, Do>,
150    {
151        (&*self).fetch_many(query)
152    }
153
154    fn fetch_optional<'e, 'q: 'e, E>(self, query: E) -> BoxFuture<'e, Result<Option<DoRow>, Error>>
155    where
156        'c: 'e,
157        E: 'q + Execute<'q, Do>,
158    {
159        (&*self).fetch_optional(query)
160    }
161
162    fn prepare_with<'e>(
163        self,
164        sql: SqlStr,
165        parameters: &'e [DoTypeInfo],
166    ) -> BoxFuture<'e, Result<DoStatement, Error>>
167    where
168        'c: 'e,
169    {
170        (&*self).prepare_with(sql, parameters)
171    }
172
173    fn describe<'e>(self, sql: SqlStr) -> BoxFuture<'e, Result<Describe<Do>, Error>>
174    where
175        'c: 'e,
176    {
177        (&*self).describe(sql)
178    }
179}
sqlx_cloudflare_do/executor.rs

sqlx_cloudflare_do/
executor.rs
