renox_core/db/model.rs
1use std::future::Future;
2
3use super::{
4 DateTime, Db, DbValue, Executor, ModelColumn, ModelIndex, ModelKey, Query, ToDbValue, now,
5 quote, sql,
6};
7use crate::{Error, Result};
8use anyhow::anyhow;
9
10/// A struct stored as a row in a table. Derive it with `#[derive(Model)]`.
11///
12/// ```
13/// # use renox::prelude::*;
14/// # use serde::Serialize;
15/// #[derive(Model, Serialize, Default)]
16/// #[model(table = "products", soft_deletes)]
17/// struct Product {
18/// id: i64,
19/// name: String,
20/// price: i64,
21/// created_at: Option<DateTime>,
22/// updated_at: Option<DateTime>,
23/// deleted_at: Option<DateTime>,
24/// }
25///
26/// # async fn demo(db: Db) -> Result {
27/// let mut coffee = Product { name: "Coffee".into(), price: 18_000, ..Default::default() };
28/// coffee.save(&db).await?; // INSERT, sets id and timestamps
29/// let cheap = Product::query().where_op("price", "<", 20_000).get(&db).await?;
30/// # let _ = cheap; Ok(()) }
31/// ```
32///
33/// Implement it with `#[derive(Model)]`: the hidden items it writes may
34/// change in a minor release.
35///
36/// The primary key is the `id` column. Its type is the `id` field's: `i64`
37/// (numbered by the database; `0` means "not saved yet"), or a
38/// [`Ulid`](super::Ulid), a UUID or a `String` (see [`ModelKey`]).
39pub trait Model: super::FromRow + Sized + Send + Sync + Unpin + 'static {
40 /// The table name (the snake_case struct name unless `#[model(table = …)]`).
41 const TABLE: &'static str;
42 /// Every column, including `id`.
43 const COLUMNS: &'static [&'static str];
44 /// `delete()` sets `deleted_at` instead of removing the row, and queries
45 /// skip deleted rows unless asked with `with_trashed()` / `only_trashed()`.
46 const SOFT_DELETES: bool = false;
47 /// Select every column (`*`) instead of `COLUMNS`, so `from_row` also
48 /// sees columns the struct doesn't list, and queries may filter on them.
49 /// The built-in `User` does this to keep the app's own columns.
50 const SELECT_ALL: bool = false;
51 /// The text columns full-text search looks in, most important first
52 /// (they weigh more in the ranking); empty: the model can't be
53 /// searched. Set it with `#[model(search = "title, body")]`; see
54 /// [`renox::db::search`](super::search).
55 const SEARCHABLE: &'static [&'static str] = &[];
56 /// The language full-text search stems words for: `english` (the
57 /// default) or a PostgreSQL text search configuration such as `simple`
58 /// (no stemming) or `spanish`. Set it with
59 /// `#[model(search_language = "simple")]`.
60 const SEARCH_LANGUAGE: &'static str = "english";
61
62 /// The type of the `id` field.
63 type Key: ModelKey;
64
65 /// The primary key.
66 fn id(&self) -> Self::Key;
67 /// Sets the primary key (after an insert, for keys the database numbers).
68 #[doc(hidden)]
69 fn set_id(&mut self, id: Self::Key);
70 /// Values of every column except `id`, in `COLUMNS` order.
71 #[doc(hidden)]
72 fn values(&self) -> Vec<DbValue>;
73 /// Updates `created_at` / `updated_at` if the model has them.
74 #[doc(hidden)]
75 fn touch(&mut self, _now: DateTime, _creating: bool) {}
76 /// Updates `deleted_at` if the model has it.
77 #[doc(hidden)]
78 fn set_deleted_at(&mut self, _at: Option<DateTime>) {}
79 /// Empties `created_at` / `updated_at` when they are `Option`s.
80 #[doc(hidden)]
81 fn forget_timestamps(&mut self) {}
82
83 /// The model's columns with the kind of each field's type. The derive
84 /// fills this in; a hand-written `impl Model` returns none.
85 fn column_info() -> Vec<ModelColumn> {
86 Vec::new()
87 }
88
89 /// The indexes declared with `#[model(index(…))]` and `#[model(unique(…))]`.
90 fn indexes() -> Vec<ModelIndex> {
91 Vec::new()
92 }
93
94 /// A copy of the model that isn't saved yet (Laravel's `replicate`):
95 /// the same values, with an unsaved id, no `deleted_at` and, when they
96 /// are `Option`s, no timestamps, so `save` inserts a new row. Change
97 /// what must differ (a unique SKU, a name) before saving it, or show it
98 /// in the "new" form for someone to finish ("Duplicate").
99 ///
100 /// ```
101 /// # use renox::prelude::*;
102 /// # #[derive(Model, serde::Serialize, Default, Clone)]
103 /// # #[model(table = "products")]
104 /// # struct Product { id: i64, sku: String, name: String, created_at: Option<DateTime> }
105 /// # async fn demo(db: Db) -> Result {
106 /// let original = Product::find_or_404(&db, 1).await?;
107 /// let mut copy = original.replicate();
108 /// copy.sku = format!("{}-COPY", original.sku);
109 /// copy.save(&db).await?; // a new row, with its own id
110 /// # Ok(()) }
111 /// ```
112 fn replicate(&self) -> Self
113 where
114 Self: Clone,
115 {
116 let mut copy = self.clone();
117 copy.set_id(Self::Key::default());
118 copy.set_deleted_at(None);
119 copy.forget_timestamps();
120 copy
121 }
122
123 /// Conditions every query of this model starts with, e.g. the current
124 /// tenant, read from [`renox::context`](mod@crate::context). `query()`,
125 /// `find`, `all`, `where_eq` and the relation loaders apply it;
126 /// [`Model::unscoped`] doesn't. Saving, deleting and restoring a loaded
127 /// model work by its id. Set it with `#[model(default_scope = "…")]`:
128 ///
129 /// ```
130 /// # use renox::prelude::*;
131 /// #[derive(Clone)]
132 /// struct CurrentTeam(i64);
133 ///
134 /// #[derive(Model, serde::Serialize, Default)]
135 /// #[model(table = "projects", default_scope = "team_only")]
136 /// struct Project { id: i64, team_id: i64, name: String }
137 ///
138 /// fn team_only(query: renox::db::Query<Project>) -> renox::db::Query<Project> {
139 /// match renox::context::get::<CurrentTeam>() {
140 /// Some(team) => query.where_eq("team_id", team.0),
141 /// None => query.none(), // no team, no rows: fail closed
142 /// }
143 /// }
144 /// ```
145 fn default_scope(query: Query<Self>) -> Query<Self> {
146 query
147 }
148
149 /// Runs before the row is written; an error stops the save. Implement
150 /// [`ModelHooks`] and add `#[model(hooks)]` rather than overriding it.
151 fn saving(&mut self, _creating: bool) -> Result {
152 Ok(())
153 }
154
155 /// Runs after the row is written (inside the caller's transaction, if
156 /// any); an error is returned by `save`. See [`ModelHooks`].
157 fn saved(&self, _created: bool) -> impl Future<Output = Result> + Send {
158 async { Ok(()) }
159 }
160
161 /// Runs before `delete`/`force_delete`; an error stops it. See [`ModelHooks`].
162 fn deleting(&self) -> Result {
163 Ok(())
164 }
165
166 /// Runs after `delete`/`force_delete`. See [`ModelHooks`].
167 fn deleted(&self) -> impl Future<Output = Result> + Send {
168 async { Ok(()) }
169 }
170
171 /// A query with the default scope applied (see [`Model::default_scope`]).
172 fn query() -> Query<Self> {
173 Self::default_scope(Query::new())
174 }
175
176 /// A query without the default scope, e.g. for an admin who sees every
177 /// tenant. Soft-deleted rows stay hidden unless asked for.
178 fn unscoped() -> Query<Self> {
179 Query::new()
180 }
181
182 /// Reloads the model's row (e.g. after an `increment` or another request
183 /// changed it); a deleted row is a 404.
184 fn refresh(&mut self, db: &Db) -> impl Future<Output = Result<()>> + Send {
185 async move {
186 let id = self.id();
187 *self = Self::unscoped()
188 .with_trashed()
189 .where_eq("id", id)
190 .first_or_404(db)
191 .await?;
192 Ok(())
193 }
194 }
195
196 /// The rows matching a full-text search, best matches first:
197 /// shorthand for `query().search(words)`. The model needs
198 /// `#[model(search = "…")]` and its index (see
199 /// [`renox::db::search`](super::search)).
200 ///
201 /// ```
202 /// # use renox::prelude::*;
203 /// #[derive(Model, serde::Serialize, Default)]
204 /// #[model(table = "posts", search = "title, body")]
205 /// struct Post { id: i64, title: String, body: String }
206 ///
207 /// # async fn demo(db: Db, q: String) -> Result {
208 /// let posts = Post::search(&q).limit(20).get(&db).await?;
209 /// # let _ = posts; Ok(()) }
210 /// ```
211 fn search(words: &str) -> Query<Self> {
212 Self::query().search(words)
213 }
214
215 /// Shorthand for `query().where_eq(column, value)`.
216 fn where_eq(column: &str, value: impl ToDbValue) -> Query<Self> {
217 Self::query().where_eq(column, value)
218 }
219
220 /// Every row, in id order.
221 fn all<'c, E: Executor<'c>>(db: E) -> impl Future<Output = Result<Vec<Self>>> + Send {
222 Self::query().order_by("id").get(db)
223 }
224
225 /// The row with this id, or `None`.
226 fn find<'c, E: Executor<'c>>(
227 db: E,
228 id: Self::Key,
229 ) -> impl Future<Output = Result<Option<Self>>> + Send {
230 Self::query().where_eq("id", id).first(db)
231 }
232
233 /// The rows with these ids, in id order (missing ids are skipped).
234 fn find_many<'c, E: Executor<'c>>(
235 db: E,
236 ids: impl IntoIterator<Item = Self::Key>,
237 ) -> impl Future<Output = Result<Vec<Self>>> + Send {
238 let ids: Vec<Self::Key> = ids.into_iter().collect();
239 Self::query().where_in("id", ids).order_by("id").get(db)
240 }
241
242 /// Inserts many new models with a few statements (ids aren't returned;
243 /// use `create` when you need them). Timestamps are set; unsaved ULID
244 /// and UUID keys are made, `String` keys must be set, `i64` keys come
245 /// from the database. Returns the number of rows inserted.
246 fn insert_many<'c, E: Executor<'c>>(
247 db: E,
248 models: Vec<Self>,
249 ) -> impl Future<Output = Result<u64>> + Send {
250 async move { write_many::<Self>(db.into_conn(), models, None).await }
251 }
252
253 /// Inserts `models`, or updates the rows they clash with on the
254 /// `unique_by` columns (which need a unique index), setting `update`
255 /// columns (and `updated_at` when the model has it). Returns the rows
256 /// written.
257 ///
258 /// ```
259 /// # use renox::prelude::*;
260 /// # #[derive(Model, serde::Serialize, Default)] struct Stock { id: i64, sku: String, qty: i64 }
261 /// # async fn demo(db: Db) -> Result {
262 /// let feed = vec![Stock { sku: "COFFEE-1".into(), qty: 12, ..Default::default() }];
263 /// Stock::upsert(&db, feed, &["sku"], &["qty"]).await?;
264 /// # Ok(()) }
265 /// ```
266 fn upsert<'c, E: Executor<'c>>(
267 db: E,
268 models: Vec<Self>,
269 unique_by: &[&str],
270 update: &[&str],
271 ) -> impl Future<Output = Result<u64>> + Send {
272 let unique_by: Vec<String> = unique_by.iter().map(|c| (*c).to_owned()).collect();
273 let update: Vec<String> = update.iter().map(|c| (*c).to_owned()).collect();
274 async move {
275 // Keys the app writes (ULID, UUID, String) may be the conflict
276 // target; an `i64` id isn't written, and no key is ever updated.
277 let key_target = !<Self::Key as ModelKey>::AUTO_INCREMENT;
278 for column in unique_by.iter().chain(&update) {
279 let id_allowed =
280 key_target && unique_by.contains(column) && !update.contains(column);
281 if !Self::COLUMNS.contains(&column.as_str()) || (column == "id" && !id_allowed) {
282 return Err(
283 anyhow!("`{}` has no column `{column}` to upsert", Self::TABLE).into(),
284 );
285 }
286 }
287 if unique_by.is_empty() {
288 return Err(anyhow!("upsert needs at least one `unique_by` column").into());
289 }
290 write_many::<Self>(db.into_conn(), models, Some((unique_by, update))).await
291 }
292 }
293
294 /// Like `find`, but a missing row becomes a 404 response.
295 fn find_or_404<'c, E: Executor<'c>>(
296 db: E,
297 id: Self::Key,
298 ) -> impl Future<Output = Result<Self>> + Send {
299 let found = Self::find(db, id);
300 async move { found.await?.ok_or(Error::NotFound) }
301 }
302
303 /// Inserts a new model and returns it with its id and timestamps.
304 fn create<'c, E: Executor<'c>>(
305 db: E,
306 mut model: Self,
307 ) -> impl Future<Output = Result<Self>> + Send {
308 async move {
309 model.insert(db).await?;
310 Ok(model)
311 }
312 }
313
314 /// Inserts the model as a new row, whatever its id: an unsaved id gets a
315 /// new key (from the database for `i64`, a new ULID or UUID v7), a set
316 /// one is written as it is (a `String` key must be set).
317 fn insert<'c, E: Executor<'c>>(&mut self, db: E) -> impl Future<Output = Result> + Send {
318 async move {
319 self.saving(true)?;
320 self.touch(now(), true);
321 if self.id().is_unsaved()
322 && let Some(key) = Self::Key::generate()
323 {
324 self.set_id(key);
325 }
326 let key = self.id();
327 let mut columns: Vec<String> = Self::COLUMNS
328 .iter()
329 .filter(|c| **c != "id")
330 .map(|c| quote(c))
331 .collect();
332 let mut values = self.values();
333 let table = quote(Self::TABLE);
334 if key.is_unsaved() {
335 if !<Self::Key as ModelKey>::AUTO_INCREMENT {
336 return Err(anyhow!(
337 "set the `id` of a new {} row before inserting it",
338 Self::TABLE
339 )
340 .into());
341 }
342 let sql_text = if columns.is_empty() {
343 format!("INSERT INTO {table} DEFAULT VALUES RETURNING id")
344 } else {
345 let marks = vec!["?"; columns.len()].join(", ");
346 format!(
347 "INSERT INTO {table} ({}) VALUES ({marks}) RETURNING id",
348 columns.join(", ")
349 )
350 };
351 let id: Self::Key = sql(sql_text).bind_all(values).scalar(db).await?;
352 self.set_id(id);
353 } else {
354 columns.insert(0, quote("id"));
355 values.insert(0, key.to_db_value());
356 let marks = vec!["?"; columns.len()].join(", ");
357 sql(format!(
358 "INSERT INTO {table} ({}) VALUES ({marks})",
359 columns.join(", ")
360 ))
361 .bind_all(values)
362 .execute(db)
363 .await?;
364 }
365 self.saved(true).await
366 }
367 }
368
369 /// Inserts the model if its id is unsaved (`0`, an empty ULID…),
370 /// otherwise updates its row.
371 fn save<'c, E: Executor<'c>>(&mut self, db: E) -> impl Future<Output = Result> + Send {
372 async move {
373 if self.id().is_unsaved() {
374 return self.insert(db).await;
375 }
376 self.saving(false)?;
377 self.touch(now(), false);
378 let columns: Vec<String> = Self::COLUMNS
379 .iter()
380 .filter(|c| **c != "id")
381 .map(|c| quote(c))
382 .collect();
383 let values = self.values();
384 let table = quote(Self::TABLE);
385 if columns.is_empty() {
386 return self.saved(false).await;
387 }
388 let sets: Vec<String> = columns.iter().map(|c| format!("{c} = ?")).collect();
389 let changed = sql(format!(
390 "UPDATE {table} SET {} WHERE id = ?",
391 sets.join(", ")
392 ))
393 .bind_all(values)
394 .bind(self.id())
395 .execute(db)
396 .await?;
397 if changed == 0 {
398 return Err(Error::NotFound);
399 }
400 self.saved(false).await
401 }
402 }
403
404 /// Updates only `columns` (and `updated_at`, if the model has it) of a
405 /// saved model, so a concurrent change to another column isn't
406 /// overwritten. A column the `saving` hook changes is saved only if
407 /// it's listed.
408 ///
409 /// ```
410 /// # use renox::prelude::*;
411 /// # #[derive(Model, serde::Serialize, Default)]
412 /// # #[model(table = "posts")]
413 /// # struct Post { id: i64, title: String, views: i64 }
414 /// # async fn demo(db: Db, mut post: Post) -> Result {
415 /// post.title = "New title".into();
416 /// post.save_only(&db, &["title"]).await?; // leaves `views` alone
417 /// # Ok(()) }
418 /// ```
419 fn save_only<'c, E: Executor<'c>>(
420 &mut self,
421 db: E,
422 columns: &[&str],
423 ) -> impl Future<Output = Result> + Send {
424 let columns: Vec<String> = columns.iter().map(|c| (*c).to_owned()).collect();
425 async move {
426 self.saving(false)?;
427 update_columns(self, db, columns).await
428 }
429 }
430
431 /// Saves the columns that differ from `original` (the model as it was
432 /// loaded), including those the `saving` hook changes, and returns
433 /// whether anything was written. When nothing changed there's no query
434 /// and no `saved` hook.
435 ///
436 /// ```
437 /// # use renox::prelude::*;
438 /// # #[derive(Model, serde::Serialize, Default, Clone)]
439 /// # #[model(table = "posts")]
440 /// # struct Post { id: i64, title: String, views: i64 }
441 /// # async fn demo(db: Db) -> Result {
442 /// let original = Post::find_or_404(&db, 1).await?;
443 /// let mut post = original.clone();
444 /// post.title = "New title".into();
445 /// post.save_changes(&db, &original).await?; // UPDATE posts SET title = ?
446 /// # Ok(()) }
447 /// ```
448 fn save_changes<'c, E: Executor<'c>>(
449 &mut self,
450 db: E,
451 original: &Self,
452 ) -> impl Future<Output = Result<bool>> + Send {
453 let hooked = self.saving(false);
454 let changed: Vec<String> = Self::COLUMNS
455 .iter()
456 .filter(|c| **c != "id")
457 .zip(self.values().into_iter().zip(original.values()))
458 .filter(|(_, (now, before))| now != before)
459 .map(|(column, _)| (*column).to_owned())
460 .collect();
461 async move {
462 hooked?;
463 if changed.is_empty() {
464 return Ok(false);
465 }
466 update_columns(self, db, changed).await?;
467 Ok(true)
468 }
469 }
470
471 /// Deletes the row, or marks it deleted for models with soft deletes.
472 fn delete<'c, E: Executor<'c>>(&mut self, db: E) -> impl Future<Output = Result> + Send {
473 async move {
474 if !Self::SOFT_DELETES {
475 return self.force_delete(db).await;
476 }
477 self.deleting()?;
478 let at = now();
479 sql(format!(
480 "UPDATE {} SET deleted_at = ? WHERE id = ?",
481 quote(Self::TABLE)
482 ))
483 .bind(at)
484 .bind(self.id())
485 .execute(db)
486 .await?;
487 self.set_deleted_at(Some(at));
488 self.deleted().await
489 }
490 }
491
492 /// Removes the row, even for models with soft deletes.
493 fn force_delete<'c, E: Executor<'c>>(&self, db: E) -> impl Future<Output = Result> + Send {
494 async move {
495 self.deleting()?;
496 sql(format!("DELETE FROM {} WHERE id = ?", quote(Self::TABLE)))
497 .bind(self.id())
498 .execute(db)
499 .await?;
500 self.deleted().await
501 }
502 }
503
504 /// Brings back a soft-deleted row.
505 fn restore<'c, E: Executor<'c>>(&mut self, db: E) -> impl Future<Output = Result> + Send {
506 async move {
507 if !Self::SOFT_DELETES {
508 return Err(anyhow!("{} does not use soft deletes", Self::TABLE).into());
509 }
510 sql(format!(
511 "UPDATE {} SET deleted_at = NULL WHERE id = ?",
512 quote(Self::TABLE)
513 ))
514 .bind(self.id())
515 .execute(db)
516 .await?;
517 self.set_deleted_at(None);
518 Ok(())
519 }
520 }
521}
522
523/// `save_only` after the `saving` hook: writes `columns` and `updated_at`,
524/// then runs `saved`.
525async fn update_columns<'c, M: Model, E: Executor<'c>>(
526 model: &mut M,
527 db: E,
528 columns: Vec<String>,
529) -> Result {
530 if model.id().is_unsaved() {
531 return Err(anyhow!("save_only on an unsaved {} row", M::TABLE).into());
532 }
533 for column in &columns {
534 if column == "id" || !M::COLUMNS.contains(&column.as_str()) {
535 return Err(anyhow!("{} has no column `{column}` to save", M::TABLE).into());
536 }
537 }
538 model.touch(now(), false);
539 let mut sets = Vec::new();
540 let mut binds = Vec::new();
541 for (column, value) in M::COLUMNS
542 .iter()
543 .filter(|c| **c != "id")
544 .zip(model.values())
545 {
546 if columns.iter().any(|c| c == column) || *column == "updated_at" {
547 sets.push(format!("{} = ?", quote(column)));
548 binds.push(value);
549 }
550 }
551 if !sets.is_empty() {
552 let changed = sql(format!(
553 "UPDATE {} SET {} WHERE id = ?",
554 quote(M::TABLE),
555 sets.join(", ")
556 ))
557 .bind_all(binds)
558 .bind(model.id())
559 .execute(db)
560 .await?;
561 if changed == 0 {
562 return Err(Error::NotFound);
563 }
564 }
565 model.saved(false).await
566}
567
568/// Code that runs around a model's writes, like Laravel's model events:
569/// fill a slug, check an invariant, forget a cache key, emit an event. Add
570/// `#[model(hooks)]` and implement the ones you need:
571///
572/// ```
573/// # use renox::prelude::*;
574/// use renox::db::ModelHooks;
575///
576/// #[derive(Model, serde::Serialize, Default)]
577/// #[model(table = "posts", hooks)]
578/// struct Post { id: i64, title: String, slug: String }
579///
580/// impl ModelHooks for Post {
581/// fn saving(&mut self, _creating: bool) -> Result {
582/// self.slug = self.title.to_lowercase().replace(' ', "-");
583/// Ok(())
584/// }
585///
586/// async fn saved(&self, _created: bool) -> Result {
587/// if let Some(state) = renox::context::app() { // the request's or job's app
588/// state.cache.forget("posts.latest").await?;
589/// }
590/// Ok(())
591/// }
592/// }
593/// ```
594///
595/// They run for `save`, `save_only`, `save_changes`, `create`, `insert`, `delete` and
596/// `force_delete`, not for `restore` or bulk `Query::update`/`delete` and
597/// `insert_many`, which write many rows in one statement.
598pub trait ModelHooks {
599 /// Runs before the row is written (`creating` is true for an insert); an error cancels it.
600 fn saving(&mut self, _creating: bool) -> Result {
601 Ok(())
602 }
603
604 /// Runs after the row is written (`created` is true for an insert).
605 fn saved(&self, _created: bool) -> impl Future<Output = Result> + Send {
606 async { Ok(()) }
607 }
608
609 /// Runs before the row is deleted; an error cancels the delete.
610 fn deleting(&self) -> Result {
611 Ok(())
612 }
613
614 /// Runs after the row is deleted.
615 fn deleted(&self) -> impl Future<Output = Result> + Send {
616 async { Ok(()) }
617 }
618}
619
620/// Most parameters one statement binds; both databases allow more
621/// (SQLite 32,766, PostgreSQL 65,535).
622const MAX_BINDS: usize = 30_000;
623
624/// `insert_many` / `upsert`: multi-row INSERTs in chunks under the bind limit.
625async fn write_many<M: Model>(
626 mut conn: super::Conn<'_>,
627 mut models: Vec<M>,
628 upsert: Option<(Vec<String>, Vec<String>)>,
629) -> Result<u64> {
630 let at = now();
631 // `i64` keys come from the database; other keys are written, made here
632 // (ULID, UUID) when unsaved.
633 let with_keys = !<M::Key as super::ModelKey>::AUTO_INCREMENT;
634 for model in &mut models {
635 model.touch(at, true);
636 if with_keys && model.id().is_unsaved() {
637 match <M::Key as super::ModelKey>::generate() {
638 Some(key) => model.set_id(key),
639 None => {
640 return Err(anyhow!(
641 "set the `id` of every new {} row before inserting them",
642 M::TABLE
643 )
644 .into());
645 }
646 }
647 }
648 }
649 let mut columns: Vec<&str> = M::COLUMNS.iter().copied().filter(|c| *c != "id").collect();
650 if with_keys {
651 columns.insert(0, "id");
652 }
653 if columns.is_empty() || models.is_empty() {
654 return Ok(0);
655 }
656 let quoted: Vec<String> = columns.iter().map(|c| quote(c)).collect();
657 let row_marks = format!("({})", vec!["?"; columns.len()].join(", "));
658 let conflict = upsert.map(|(unique_by, update)| {
659 let targets: Vec<String> = unique_by.iter().map(|c| quote(c)).collect();
660 let mut sets: Vec<String> = update
661 .iter()
662 .map(|c| format!("{0} = excluded.{0}", quote(c)))
663 .collect();
664 if columns.contains(&"updated_at") && !update.iter().any(|c| c == "updated_at") {
665 sets.push(format!("{0} = excluded.{0}", quote("updated_at")));
666 }
667 if sets.is_empty() {
668 format!(" ON CONFLICT ({}) DO NOTHING", targets.join(", "))
669 } else {
670 format!(
671 " ON CONFLICT ({}) DO UPDATE SET {}",
672 targets.join(", "),
673 sets.join(", ")
674 )
675 }
676 });
677 let per_statement = (MAX_BINDS / columns.len()).max(1);
678 let mut written = 0;
679 for chunk in models.chunks(per_statement) {
680 let marks = vec![row_marks.as_str(); chunk.len()].join(", ");
681 let statement = format!(
682 "INSERT INTO {} ({}) VALUES {marks}{}",
683 quote(M::TABLE),
684 quoted.join(", "),
685 conflict.as_deref().unwrap_or_default()
686 );
687 let values = chunk.iter().flat_map(|model| {
688 let key = with_keys.then(|| model.id().to_db_value());
689 key.into_iter().chain(model.values())
690 });
691 written += sql(statement)
692 .bind_all(values)
693 .execute(conn.reborrow())
694 .await?;
695 }
696 Ok(written)
697}