sqlx_cloudflare_d1/
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;
11use worker::send::SendFuture;
12use worker::worker_sys::types::D1PreparedStatement as D1PreparedStatementSys;
13
14use crate::error::unsupported;
15use crate::{js, D1Arguments, D1Connection, D1QueryResult, D1Row, D1Statement, D1TypeInfo, D1};
16
17// Every future here runs JavaScript, so none of them is `Send`, and sqlx
18// requires all of them to be. `SendFuture` asserts it -- soundly, because a
19// Worker is single-threaded -- and it is applied here, once per method, so
20// that `js` never has to know about sqlx's bounds.
21impl<'c> Executor<'c> for &'c D1Connection {
22    type Database = D1;
23
24    /// Runs the query with `run()`, which reports `rows_affected` and
25    /// `last_insert_rowid` but no rows.
26    fn execute<'e, 'q: 'e, E>(self, query: E) -> BoxFuture<'e, Result<D1QueryResult, Error>>
27    where
28        'c: 'e,
29        E: 'q + Execute<'q, D1>,
30    {
31        SendFuture::new(async move {
32            let statement = prepare(self, query)?;
33            js::run(&statement).await
34        })
35        .boxed()
36    }
37
38    fn execute_many<'e, 'q: 'e, E>(self, query: E) -> BoxStream<'e, Result<D1QueryResult, Error>>
39    where
40        'c: 'e,
41        E: 'q + Execute<'q, D1>,
42    {
43        stream::once(self.execute(query)).boxed()
44    }
45
46    /// Yields the rows only, never a `D1QueryResult`: rows come from
47    /// `raw({ columnNames: true })`, which reports nothing else. Use
48    /// [`execute`](Executor::execute) for what a statement changed.
49    fn fetch_many<'e, 'q: 'e, E>(
50        self,
51        query: E,
52    ) -> BoxStream<'e, Result<Either<D1QueryResult, D1Row>, Error>>
53    where
54        'c: 'e,
55        E: 'q + Execute<'q, D1>,
56    {
57        SendFuture::new(async move {
58            let statement = prepare(self, query)?;
59            js::fetch(&statement, None).await
60        })
61        .map_ok(|rows| stream::iter(rows.into_iter().map(|row| Ok(Either::Right(row)))))
62        .try_flatten_stream()
63        .boxed()
64    }
65
66    fn fetch_optional<'e, 'q: 'e, E>(self, query: E) -> BoxFuture<'e, Result<Option<D1Row>, Error>>
67    where
68        'c: 'e,
69        E: 'q + Execute<'q, D1>,
70    {
71        SendFuture::new(async move {
72            let statement = prepare(self, query)?;
73            Ok(js::fetch(&statement, Some(1)).await?.into_iter().next())
74        })
75        .boxed()
76    }
77
78    fn prepare_with<'e>(
79        self,
80        sql: SqlStr,
81        _parameters: &'e [D1TypeInfo],
82    ) -> BoxFuture<'e, Result<D1Statement, Error>>
83    where
84        'c: 'e,
85    {
86        future::ready(Ok(D1Statement { sql })).boxed()
87    }
88
89    /// Always fails: D1 cannot describe a statement without running it. Only
90    /// sqlx's `query!` macros need this, and they cannot target a third-party
91    /// driver anyway.
92    fn describe<'e>(self, _sql: SqlStr) -> BoxFuture<'e, Result<Describe<D1>, Error>>
93    where
94        'c: 'e,
95    {
96        future::ready(Err(unsupported("statements cannot be described"))).boxed()
97    }
98}
99
100/// `query` prepared on `conn`'s database, with its arguments bound.
101fn prepare<'q, E>(conn: &D1Connection, mut query: E) -> Result<D1PreparedStatementSys, Error>
102where
103    E: Execute<'q, D1>,
104{
105    let arguments = query.take_arguments().map_err(Error::Encode)?;
106    let arguments = arguments.as_ref().map_or(&[][..], D1Arguments::values);
107
108    js::prepare(&conn.db, query.sql().as_str(), arguments)
109}
110
111// What sqlx code actually passes -- `&mut conn` -- running on the shared
112// implementation above. `&'c mut` gives up its uniqueness to become `&'c`,
113// which is all a query needs.
114impl<'c> Executor<'c> for &'c mut D1Connection {
115    type Database = D1;
116
117    fn execute<'e, 'q: 'e, E>(self, query: E) -> BoxFuture<'e, Result<D1QueryResult, Error>>
118    where
119        'c: 'e,
120        E: 'q + Execute<'q, D1>,
121    {
122        (&*self).execute(query)
123    }
124
125    fn execute_many<'e, 'q: 'e, E>(self, query: E) -> BoxStream<'e, Result<D1QueryResult, Error>>
126    where
127        'c: 'e,
128        E: 'q + Execute<'q, D1>,
129    {
130        (&*self).execute_many(query)
131    }
132
133    fn fetch_many<'e, 'q: 'e, E>(
134        self,
135        query: E,
136    ) -> BoxStream<'e, Result<Either<D1QueryResult, D1Row>, Error>>
137    where
138        'c: 'e,
139        E: 'q + Execute<'q, D1>,
140    {
141        (&*self).fetch_many(query)
142    }
143
144    fn fetch_optional<'e, 'q: 'e, E>(self, query: E) -> BoxFuture<'e, Result<Option<D1Row>, Error>>
145    where
146        'c: 'e,
147        E: 'q + Execute<'q, D1>,
148    {
149        (&*self).fetch_optional(query)
150    }
151
152    fn prepare_with<'e>(
153        self,
154        sql: SqlStr,
155        parameters: &'e [D1TypeInfo],
156    ) -> BoxFuture<'e, Result<D1Statement, Error>>
157    where
158        'c: 'e,
159    {
160        (&*self).prepare_with(sql, parameters)
161    }
162
163    fn describe<'e>(self, sql: SqlStr) -> BoxFuture<'e, Result<Describe<D1>, Error>>
164    where
165        'c: 'e,
166    {
167        (&*self).describe(sql)
168    }
169}
sqlx_cloudflare_d1/executor.rs

sqlx_cloudflare_d1/
executor.rs
