ocre/runtime/query.rs
1//! The terminal methods of [`Query`]: they send the built statement to D1.
2
3use serde::{Deserialize, de::DeserializeOwned};
4
5use super::Db;
6use crate::{Batches, Error, Page, Paginated, Param, Query, Result};
7
8#[derive(Deserialize)]
9struct Count {
10 count: i64,
11}
12
13#[derive(Deserialize)]
14struct Value<V> {
15 value: V,
16}
17
18#[derive(Deserialize)]
19struct Plan {
20 detail: String,
21}
22
23impl<T: DeserializeOwned> Query<T> {
24 /// Runs the query and returns every matching row.
25 ///
26 /// All rows are buffered in memory: bound the query with
27 /// [`limit`](Self::limit) or [`page`](Self::page).
28 ///
29 /// # Errors
30 ///
31 /// [`Error::Internal`](crate::Error::Internal) (500, logged with the SQL)
32 /// when D1 rejects the statement or a row does not deserialize into `T`.
33 ///
34 /// # Free plan
35 ///
36 /// Counts every row D1 scans as a row read.
37 ///
38 /// # Examples
39 ///
40 /// ```no_run
41 /// # use ocre::{Ctx, Query, Result};
42 /// # #[derive(serde::Deserialize)] struct Post { id: i64 }
43 /// async fn drafts(ctx: &Ctx) -> Result<Vec<Post>> {
44 /// Query::table("posts").eq("published", false).order_desc("id").limit(50).all(&ctx.db()?).await
45 /// }
46 /// ```
47 pub async fn all(&self, db: &Db) -> Result<Vec<T>> {
48 let stmt = self.to_statement();
49 db.all(&stmt.sql, stmt.params).await
50 }
51
52 /// Runs the query with `LIMIT 1` and returns the first row, if any
53 /// (Rails' `first`, `take` and `find_by`).
54 ///
55 /// Rows come in the query's order; add one ([`order_asc`](Self::order_asc))
56 /// for a predictable row.
57 ///
58 /// # Errors
59 ///
60 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the
61 /// statement or the row does not deserialize into `T`.
62 ///
63 /// # Free plan
64 ///
65 /// Reads the rows D1 scans before the first match: one with an index on
66 /// the filtered column.
67 ///
68 /// # Examples
69 ///
70 /// ```no_run
71 /// # use ocre::{Ctx, OptionExt, Query, Result};
72 /// # #[derive(serde::Deserialize)] struct Post { id: i64 }
73 /// async fn by_slug(ctx: &Ctx, slug: &str) -> Result<Post> {
74 /// Query::table("posts").eq("slug", slug).first(&ctx.db()?).await?.or_404()
75 /// }
76 /// ```
77 pub async fn first(&self, db: &Db) -> Result<Option<T>> {
78 let stmt = self.clone().limit(1).to_statement();
79 db.first(&stmt.sql, stmt.params).await
80 }
81
82 /// Runs the query and its [`count_statement`](Self::count_statement) for
83 /// `page`: the rows plus the total for pagination links.
84 ///
85 /// # Errors
86 ///
87 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects
88 /// either statement or a row does not deserialize.
89 ///
90 /// # Free plan
91 ///
92 /// Two queries: the count reads every matching row (see [`Paginated`]).
93 ///
94 /// # Examples
95 ///
96 /// ```no_run
97 /// use axum::extract::State;
98 /// use ocre::{ApiResult, Ctx, Json, Page, Paginated, Query};
99 ///
100 /// #[derive(serde::Deserialize, serde::Serialize)]
101 /// struct Post {
102 /// id: i64,
103 /// title: String,
104 /// }
105 ///
106 /// async fn index(State(ctx): State<Ctx>, page: Page) -> ApiResult<Json<Paginated<Post>>> {
107 /// let posts = Query::table("posts").order_desc("id").paginate(&ctx.db()?, page).await?;
108 /// Ok(Json(posts))
109 /// }
110 /// # let _ = index;
111 /// ```
112 pub async fn paginate(&self, db: &Db, page: Page) -> Result<Paginated<T>> {
113 let total = self.count(db).await?;
114 let items = self.clone().page(page).all(db).await?;
115 Ok(Paginated { items, total, limit: page.limit, offset: page.offset })
116 }
117
118 /// Returns the first matching row, or runs `create` when there is none
119 /// (Rails' `find_or_create_by`; with a `New...` value built in memory
120 /// instead of saved, `find_or_initialize_by`).
121 ///
122 /// Two requests can both miss and both create: back the lookup with a
123 /// `UNIQUE` index and use [`create_or_first`](Self::create_or_first)
124 /// when duplicates must not happen.
125 ///
126 /// # Errors
127 ///
128 /// The lookup's errors, or those of `create` (e.g. [`Error::Invalid`]).
129 ///
130 /// # Free plan
131 ///
132 /// One query when the row exists, plus `create`'s otherwise.
133 ///
134 /// # Examples
135 ///
136 /// ```no_run
137 /// use ocre::{Ctx, Error, Query, Result, params};
138 ///
139 /// #[derive(serde::Deserialize)]
140 /// struct Tag {
141 /// id: i64,
142 /// name: String,
143 /// }
144 ///
145 /// async fn tag_named(ctx: &Ctx, name: &str) -> Result<Tag> {
146 /// let db = ctx.db()?;
147 /// Query::table("tags")
148 /// .eq("name", name)
149 /// .first_or_create(&db, || async {
150 /// let sql = "INSERT INTO tags (name) VALUES (?1) RETURNING *";
151 /// db.first(sql, params![name]).await?.ok_or_else(|| Error::internal("no row returned"))
152 /// })
153 /// .await
154 /// }
155 /// ```
156 pub async fn first_or_create<F, Fut>(&self, db: &Db, create: F) -> Result<T>
157 where
158 F: FnOnce() -> Fut,
159 Fut: Future<Output = Result<T>>,
160 {
161 match self.first(db).await? {
162 Some(row) => Ok(row),
163 None => create().await,
164 }
165 }
166
167 /// Runs `create` first and, when it fails because the value is already
168 /// taken, returns the existing row instead (Rails' `create_or_find_by`).
169 ///
170 /// Safe against two requests racing: the table's `UNIQUE` index decides
171 /// and the loser reads the winner's row. "Taken" means
172 /// [`Error::is_taken`]: a generated model's "has already been taken"
173 /// validation, or D1's `UNIQUE constraint failed`. The query must match
174 /// the conflicting row (usually `eq` on the unique column).
175 ///
176 /// # Errors
177 ///
178 /// Any other error of `create`; [`Error::NotFound`] when the conflicting
179 /// row cannot be found by this query.
180 ///
181 /// # Free plan
182 ///
183 /// `create`'s queries, plus one lookup after a conflict.
184 ///
185 /// # Examples
186 ///
187 /// ```no_run
188 /// use ocre::{Ctx, Error, Query, Result, params};
189 ///
190 /// #[derive(serde::Deserialize)]
191 /// struct Subscriber {
192 /// id: i64,
193 /// email: String,
194 /// }
195 ///
196 /// // `email` has a UNIQUE index (`email:string^`).
197 /// async fn subscribe(ctx: &Ctx, email: &str) -> Result<Subscriber> {
198 /// let db = ctx.db()?;
199 /// Query::table("subscribers")
200 /// .eq("email", email)
201 /// .create_or_first(&db, || async {
202 /// let sql = "INSERT INTO subscribers (email) VALUES (?1) RETURNING *";
203 /// db.first(sql, params![email]).await?.ok_or_else(|| Error::internal("no row returned"))
204 /// })
205 /// .await
206 /// }
207 /// ```
208 pub async fn create_or_first<F, Fut>(&self, db: &Db, create: F) -> Result<T>
209 where
210 F: FnOnce() -> Fut,
211 Fut: Future<Output = Result<T>>,
212 {
213 match create().await {
214 Err(err) if err.is_taken() => self.first(db).await?.ok_or(Error::NotFound),
215 other => other,
216 }
217 }
218}
219
220impl<T: DeserializeOwned> Batches<T> {
221 /// Runs the next batch: up to the batch size of rows after the last id
222 /// read, or `None` once every row was read.
223 ///
224 /// See [`Query::batches`].
225 ///
226 /// # Errors
227 ///
228 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the
229 /// statement or a row does not deserialize into `T`.
230 ///
231 /// # Free plan
232 ///
233 /// One query reading at most the batch size of rows (plus the rows the
234 /// conditions skip without an index).
235 ///
236 /// # Examples
237 ///
238 /// ```no_run
239 /// use ocre::{Ctx, Query, Result};
240 ///
241 /// #[derive(serde::Deserialize)]
242 /// struct Post {
243 /// id: i64,
244 /// }
245 ///
246 /// async fn count_by_hand(ctx: &Ctx) -> Result<usize> {
247 /// let db = ctx.db()?;
248 /// let mut batches = Query::<Post>::table("posts").batches(500, |post| post.id);
249 /// let mut total = 0;
250 /// while let Some(posts) = batches.next(&db).await? {
251 /// total += posts.len();
252 /// }
253 /// Ok(total)
254 /// }
255 /// ```
256 pub async fn next(&mut self, db: &Db) -> Result<Option<Vec<T>>> {
257 if self.is_done() {
258 return Ok(None);
259 }
260 let stmt = self.statement();
261 let rows: Vec<T> = db.all(&stmt.sql, stmt.params).await?;
262 self.advance(&rows);
263 Ok(if rows.is_empty() { None } else { Some(rows) })
264 }
265}
266
267impl<T> Query<T> {
268 /// The query plan, one line per step (Rails' `explain`): `SEARCH` means
269 /// an index is used, `SCAN` a full table read.
270 ///
271 /// Runs [`explain_statement`](Self::explain_statement). Check a query
272 /// while developing (log it, or return it from a debug route), not on
273 /// every request.
274 ///
275 /// # Errors
276 ///
277 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the statement.
278 ///
279 /// # Free plan
280 ///
281 /// One query that reads no table row.
282 ///
283 /// # Examples
284 ///
285 /// ```no_run
286 /// # use ocre::{Ctx, Query, Result};
287 /// async fn plan(ctx: &Ctx) -> Result<String> {
288 /// let steps = Query::<()>::table("posts").eq("author_id", 3).explain(&ctx.db()?).await?;
289 /// Ok(steps.join("\n")) // "SEARCH posts USING INDEX index_posts_on_author_id (author_id=?)"
290 /// }
291 /// ```
292 pub async fn explain(&self, db: &Db) -> Result<Vec<String>> {
293 let stmt = self.explain_statement();
294 let rows: Vec<Plan> = db.all(&stmt.sql, stmt.params).await?;
295 Ok(rows.into_iter().map(|row| row.detail).collect())
296 }
297
298 /// Number of matching rows (`COUNT(*)`), ignoring order and limits.
299 ///
300 /// # Errors
301 ///
302 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the statement.
303 ///
304 /// # Free plan
305 ///
306 /// Reads every matching row (an index on the filtered columns keeps it to those).
307 ///
308 /// # Examples
309 ///
310 /// ```no_run
311 /// # use ocre::{Ctx, Query, Result};
312 /// async fn published_count(ctx: &Ctx) -> Result<i64> {
313 /// Query::<()>::table("posts").eq("published", true).count(&ctx.db()?).await
314 /// }
315 /// ```
316 pub async fn count(&self, db: &Db) -> Result<i64> {
317 let stmt = self.count_statement();
318 let row: Option<Count> = db.first(&stmt.sql, stmt.params).await?;
319 Ok(row.map_or(0, |row| row.count))
320 }
321
322 /// Whether any row matches (`SELECT 1 ... LIMIT 1`; Rails' `exists?`/`any?`).
323 ///
324 /// # Errors
325 ///
326 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the statement.
327 ///
328 /// # Free plan
329 ///
330 /// Stops at the first match: one row read with an index.
331 ///
332 /// # Examples
333 ///
334 /// ```no_run
335 /// # use ocre::{Ctx, Query, Result};
336 /// async fn email_taken(ctx: &Ctx, email: &str) -> Result<bool> {
337 /// Query::<()>::table("users").eq("email", email).exists(&ctx.db()?).await
338 /// }
339 /// ```
340 pub async fn exists(&self, db: &Db) -> Result<bool> {
341 let stmt = self.exists_statement();
342 db.exists(&stmt.sql, stmt.params).await
343 }
344
345 /// Values of one column (or expression) of the matching rows (Rails' `pluck`/`ids`).
346 ///
347 /// Keeps the query's order and limits; with [`first`](Self::first)-like
348 /// use, add `.limit(1)` (Rails' `pick`).
349 ///
350 /// # Errors
351 ///
352 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the
353 /// statement or a value does not deserialize into `V`.
354 ///
355 /// # Free plan
356 ///
357 /// Same rows read as [`all`](Self::all); less memory and CPU than full rows.
358 ///
359 /// # Examples
360 ///
361 /// ```no_run
362 /// # use ocre::{Ctx, Query, Result};
363 /// async fn published_ids(ctx: &Ctx) -> Result<Vec<i64>> {
364 /// Query::<()>::table("posts").eq("published", true).limit(100).pluck(&ctx.db()?, "id").await
365 /// }
366 /// ```
367 pub async fn pluck<V: DeserializeOwned>(&self, db: &Db, expression: &'static str) -> Result<Vec<V>> {
368 let stmt = self.value_statement(expression);
369 let rows: Vec<Value<V>> = db.all(&stmt.sql, stmt.params).await?;
370 Ok(rows.into_iter().map(|row| row.value).collect())
371 }
372
373 /// An aggregate over the matching rows: `SUM(price)`, `AVG(rating)`,
374 /// `MIN(created_at)`, `MAX(views)`, `COUNT(DISTINCT author_id)`...
375 ///
376 /// `None` when SQL returns `NULL` (e.g. `SUM` or `MAX` of no row), so
377 /// read it as `Option<V>`.
378 ///
379 /// # Errors
380 ///
381 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the
382 /// statement or the value does not deserialize into `V`.
383 ///
384 /// # Free plan
385 ///
386 /// Reads every matching row.
387 ///
388 /// # Examples
389 ///
390 /// ```no_run
391 /// # use ocre::{Ctx, Query, Result};
392 /// async fn average_rating(ctx: &Ctx, book_id: i64) -> Result<Option<f64>> {
393 /// Query::<()>::table("reviews").eq("book_id", book_id).aggregate(&ctx.db()?, "AVG(stars)").await
394 /// }
395 /// ```
396 pub async fn aggregate<V: DeserializeOwned>(&self, db: &Db, expression: &'static str) -> Result<Option<V>> {
397 let stmt = self.aggregate_statement(expression);
398 let row: Option<Value<Option<V>>> = db.first(&stmt.sql, stmt.params).await?;
399 Ok(row.and_then(|row| row.value))
400 }
401
402 /// Updates every matching row without validation (Rails' `update_all`)
403 /// and returns how many changed.
404 ///
405 /// # Errors
406 ///
407 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the
408 /// statement (e.g. a constraint).
409 ///
410 /// # Free plan
411 ///
412 /// Counts the rows written, plus the rows scanned to find them.
413 ///
414 /// # Examples
415 ///
416 /// ```no_run
417 /// # use ocre::{Ctx, IntoParam, Query, Result};
418 /// async fn unpublish_author(ctx: &Ctx, author_id: i64) -> Result<usize> {
419 /// Query::<()>::table("posts")
420 /// .eq("author_id", author_id)
421 /// .update_all(&ctx.db()?, vec![("published", false.into_param())])
422 /// .await
423 /// }
424 /// ```
425 pub async fn update_all(&self, db: &Db, sets: Vec<(&'static str, Param)>) -> Result<usize> {
426 let stmt = self.update_statement(sets);
427 db.execute(&stmt.sql, stmt.params).await
428 }
429
430 /// Deletes every matching row (Rails' `delete_all`) and returns how many were deleted.
431 ///
432 /// Runs no model code: attachments in R2 stay (see `storage::delete_attachments`).
433 ///
434 /// # Errors
435 ///
436 /// [`Error::Internal`](crate::Error::Internal) (500) when D1 rejects the
437 /// statement (e.g. a foreign key).
438 ///
439 /// # Free plan
440 ///
441 /// Counts the rows written (deleted, plus cascaded ones), plus the rows scanned.
442 ///
443 /// # Examples
444 ///
445 /// ```no_run
446 /// # use ocre::{Ctx, Query, Result};
447 /// async fn purge_expired(ctx: &Ctx) -> Result<usize> {
448 /// Query::<()>::table("sessions").lt("expires_at", ocre::now()).delete_all(&ctx.db()?).await
449 /// }
450 /// ```
451 pub async fn delete_all(&self, db: &Db) -> Result<usize> {
452 let stmt = self.delete_statement();
453 db.execute(&stmt.sql, stmt.params).await
454 }
455}