siderite_orm/
m2m.rs

1//! Many-to-many operations on [`ManyToManyManager`]: the join table is
2//! read and written directly, so relations with an auto-created table need
3//! no through model.
4//!
5//! Mutations run in a transaction (a savepoint inside an open one). An
6//! explicit through model with further required columns cannot be filled by
7//! `add`; insert its rows directly instead.
8
9use crate::db::Db;
10use crate::error::OrmError;
11use crate::expr::Expr;
12use crate::model::Model;
13use crate::persist::{pk_field, value_key};
14use crate::plan::QueryPlan;
15use crate::queryset::QuerySet;
16use crate::relations::ManyToManyManager;
17use crate::signals::{M2mAction, SignalKind, SignalName};
18use crate::types::DbType;
19use crate::value::Value;
20use crate::write::{DeletePlan, InsertPlan, WritePlan};
21use std::collections::{HashMap, HashSet};
22
23impl<S: Model, T: Model> ManyToManyManager<S, T> {
24    /// `through.source = <this object>` predicate.
25    fn owner(&self) -> Expr {
26        Expr::col(self.meta().source_column).eq(Expr::Value(self.source_pk().clone()))
27    }
28
29    /// Lazy queryset over the related objects (`WHERE pk IN (SELECT target
30    /// FROM through WHERE source = ..)`); filter or order it further.
31    pub fn queryset(&self) -> QuerySet<T> {
32        let meta = self.meta();
33        let targets = QueryPlan::from_table(meta.through_table)
34            .select(Expr::col(meta.target_column), None)
35            .filter(self.owner());
36        let pk = crate::expr::pk_column::<T>();
37        T::objects(self.db()).filter(Expr::col(pk).in_subquery(targets))
38    }
39
40    /// Every related object.
41    ///
42    /// # Errors
43    /// Capability or backend errors.
44    pub async fn all(&self) -> Result<Vec<T>, OrmError> {
45        self.queryset().all().await
46    }
47
48    /// Number of related objects.
49    ///
50    /// # Errors
51    /// Capability or backend errors.
52    pub async fn count(&self) -> Result<u64, OrmError> {
53        self.queryset().count().await
54    }
55
56    /// Whether `object` is related.
57    ///
58    /// # Errors
59    /// Capability or backend errors.
60    pub async fn contains(&self, object: &T) -> Result<bool, OrmError> {
61        self.queryset().contains(object).await
62    }
63
64    /// Relate `objects` (already related ones are skipped).
65    ///
66    /// # Errors
67    /// Capability or backend errors ([`BackendError::Constraint`](crate::BackendError::Constraint)
68    /// when a target does not exist).
69    pub async fn add<'a>(&self, objects: impl IntoIterator<Item = &'a T>) -> Result<(), OrmError> {
70        self.add_pks(objects.into_iter().map(Model::pk)).await
71    }
72
73    /// [`add`](Self::add) by primary key.
74    ///
75    /// # Errors
76    /// As [`add`](Self::add).
77    pub async fn add_pks(&self, keys: impl IntoIterator<Item = T::Pk>) -> Result<(), OrmError> {
78        let keys = distinct(keys);
79        self.db()
80            .transaction(|tx| async move {
81                let present = self.linked(&tx, Some(&keys)).await?;
82                let missing: Vec<Value> = keys
83                    .into_iter()
84                    .filter(|k| !present.contains_key(&value_key(k)))
85                    .collect();
86                self.link(&tx, missing).await
87            })
88            .await
89    }
90
91    /// Unrelate `objects`; returns the number of links removed.
92    ///
93    /// # Errors
94    /// Capability or backend errors.
95    pub async fn remove<'a>(
96        &self,
97        objects: impl IntoIterator<Item = &'a T>,
98    ) -> Result<u64, OrmError> {
99        self.remove_pks(objects.into_iter().map(Model::pk)).await
100    }
101
102    /// [`remove`](Self::remove) by primary key.
103    ///
104    /// # Errors
105    /// As [`remove`](Self::remove).
106    pub async fn remove_pks(&self, keys: impl IntoIterator<Item = T::Pk>) -> Result<u64, OrmError> {
107        let keys = distinct(keys);
108        self.db()
109            .transaction(|tx| async move { self.unlink(&tx, Some(keys)).await })
110            .await
111    }
112
113    /// Unrelate everything; returns the number of links removed.
114    ///
115    /// # Errors
116    /// Capability or backend errors.
117    pub async fn clear(&self) -> Result<u64, OrmError> {
118        self.db()
119            .transaction(|tx| async move { self.unlink(&tx, None).await })
120            .await
121    }
122
123    /// Make `objects` exactly the related set: add the missing, remove the rest.
124    ///
125    /// # Errors
126    /// Capability or backend errors; nothing changes on failure.
127    pub async fn set<'a>(&self, objects: impl IntoIterator<Item = &'a T>) -> Result<(), OrmError> {
128        self.set_pks(objects.into_iter().map(Model::pk)).await
129    }
130
131    /// [`set`](Self::set) by primary key.
132    ///
133    /// # Errors
134    /// As [`set`](Self::set).
135    pub async fn set_pks(&self, keys: impl IntoIterator<Item = T::Pk>) -> Result<(), OrmError> {
136        let wanted = distinct(keys);
137        self.db()
138            .transaction(|tx| async move {
139                let current = self.linked(&tx, None).await?;
140                let wanted_keys: HashSet<String> = wanted.iter().map(value_key).collect();
141                let stale: Vec<Value> = current
142                    .iter()
143                    .filter(|(key, _)| !wanted_keys.contains(*key))
144                    .map(|(_, value)| value.clone())
145                    .collect();
146                let missing: Vec<Value> = wanted
147                    .into_iter()
148                    .filter(|k| !current.contains_key(&value_key(k)))
149                    .collect();
150                if !stale.is_empty() {
151                    self.unlink(&tx, Some(stale)).await?;
152                }
153                self.link(&tx, missing).await
154            })
155            .await
156    }
157
158    /// Keys (as [`value_key`]s) of the targets already linked; restricted to
159    /// `among` when given.
160    async fn linked(
161        &self,
162        db: &Db,
163        among: Option<&[Value]>,
164    ) -> Result<HashMap<String, Value>, OrmError> {
165        let meta = self.meta();
166        let mut filter = self.owner();
167        if let Some(among) = among {
168            filter = filter.and(Expr::col(meta.target_column).is_in(among.iter().cloned()));
169        }
170        let plan = QueryPlan::from_table(meta.through_table)
171            .select(Expr::col(meta.target_column), None)
172            .filter(filter);
173        let mut linked = HashMap::new();
174        for row in db.fetch(&plan).await?.rows {
175            // Normalise through the key type: storage forms differ by backend.
176            let key = row.decode_at::<T::Pk>(0)?.to_value();
177            linked.insert(value_key(&key), key);
178        }
179        Ok(linked)
180    }
181
182    /// Send `m2m_changed` for `action` to receivers typed on the source model.
183    ///
184    /// The source object is loaded by primary key, and only when a receiver is
185    /// connected, so unobserved relations pay nothing.
186    async fn notify(&self, db: &Db, action: M2mAction, keys: &[Value]) -> Result<(), OrmError> {
187        let signals = db.signals();
188        if !signals.has_receivers::<S>(SignalName::M2mChanged) {
189            return Ok(());
190        }
191        let pk = pk_field::<S>()?;
192        let source = S::objects(db)
193            .get(Expr::col(pk.column).eq(Expr::Value(self.source_pk().clone())))
194            .await?;
195        signals
196            .send(
197                db,
198                &source,
199                SignalKind::M2mChanged { action },
200                Some(self.meta().name),
201                keys,
202            )
203            .await
204    }
205
206    async fn link(&self, db: &Db, keys: Vec<Value>) -> Result<(), OrmError> {
207        if keys.is_empty() {
208            return Ok(());
209        }
210        self.notify(db, M2mAction::PreAdd, &keys).await?;
211        self.insert_links(db, &keys).await?;
212        self.notify(db, M2mAction::PostAdd, &keys).await
213    }
214
215    async fn insert_links(&self, db: &Db, keys: &[Value]) -> Result<(), OrmError> {
216        let meta = self.meta();
217        let per_statement = (db.capabilities().max_params / 2).max(1);
218        for keys in keys.chunks(per_statement) {
219            let plan = WritePlan::Insert(InsertPlan {
220                table: meta.through_table.into(),
221                columns: vec![meta.source_column.into(), meta.target_column.into()],
222                rows: keys
223                    .iter()
224                    .map(|k| vec![self.source_pk().clone(), k.clone()])
225                    .collect(),
226                returning: Vec::new(),
227            });
228            db.execute(&plan).await?;
229        }
230        Ok(())
231    }
232
233    /// Remove the links to `keys`, or every link for `None`.
234    async fn unlink(&self, db: &Db, keys: Option<Vec<Value>>) -> Result<u64, OrmError> {
235        let (pre, post, pk_set) = match &keys {
236            Some(keys) if keys.is_empty() => return Ok(0),
237            Some(keys) => (M2mAction::PreRemove, M2mAction::PostRemove, keys.as_slice()),
238            None => (M2mAction::PreClear, M2mAction::PostClear, [].as_slice()),
239        };
240        self.notify(db, pre, pk_set).await?;
241        let removed = self.delete_links(db, keys.as_deref()).await?;
242        self.notify(db, post, pk_set).await?;
243        Ok(removed)
244    }
245
246    async fn delete_links(&self, db: &Db, keys: Option<&[Value]>) -> Result<u64, OrmError> {
247        let meta = self.meta();
248        let per_statement = db.capabilities().max_params.saturating_sub(1).max(1);
249        let batches: Vec<Option<&[Value]>> = match keys {
250            Some(keys) => keys.chunks(per_statement).map(Some).collect(),
251            None => vec![None],
252        };
253        let mut removed = 0;
254        for batch in batches {
255            let mut filter = self.owner();
256            if let Some(batch) = batch {
257                filter = filter.and(Expr::col(meta.target_column).is_in(batch.iter().cloned()));
258            }
259            let plan = WritePlan::Delete(DeletePlan {
260                table: meta.through_table.into(),
261                filter: Some(filter),
262                returning: Vec::new(),
263            });
264            removed += db.execute(&plan).await?.rows_affected;
265        }
266        Ok(removed)
267    }
268}
269
270/// Key values without duplicates, in first-seen order.
271fn distinct<K: DbType>(keys: impl IntoIterator<Item = K>) -> Vec<Value> {
272    let mut seen = HashSet::new();
273    keys.into_iter()
274        .map(|k| k.to_value())
275        .filter(|v| seen.insert(value_key(v)))
276        .collect()
277}
siderite_orm/m2m.rs

siderite_orm/
m2m.rs
