rorm_db/
database.rs

1//! [`Database`] struct and several common operations
2
3use std::sync::Arc;
4
5use rorm_declaration::config::DatabaseDriver;
6use rorm_sql::conditional;
7use rorm_sql::delete::Delete;
8use rorm_sql::insert::Insert;
9use rorm_sql::join_table::JoinTableData;
10use rorm_sql::ordering::OrderByEntry;
11#[cfg(feature = "postgres-only")]
12use rorm_sql::select::LockingClause;
13use rorm_sql::select_column::SelectColumnData;
14use rorm_sql::update::Update;
15use rorm_sql::value::Value;
16use tracing::warn;
17
18use crate::error::Error;
19use crate::executor::AffectedRows;
20use crate::executor::All;
21use crate::executor::Executor;
22use crate::executor::Nothing;
23use crate::executor::One;
24use crate::executor::QueryStrategy;
25use crate::internal::any::AnyPool;
26use crate::query_type::GetLimitClause;
27use crate::row::Row;
28use crate::transaction::Transaction;
29use crate::transaction::TransactionError;
30
31/**
32Type alias for [`SelectColumnData`]..
33
34As all databases use currently the same fields, a type alias is sufficient.
35*/
36pub type ColumnSelector<'a> = SelectColumnData<'a>;
37
38/**
39Type alias for [`JoinTableData`].
40
41As all databases use currently the same fields, a type alias is sufficient.
42*/
43pub type JoinTable<'until_build, 'post_build> = JoinTableData<'until_build, 'post_build>;
44
45/// Configuration use in [`Database::connect`].
46#[derive(Debug)]
47#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
48pub struct DatabaseConfiguration {
49    /// The driver and its corresponding settings
50    pub driver: DatabaseDriver,
51
52    /// Minimal connections to initialize upfront.
53    ///
54    /// Must be greater than `0` and can't be larger than `max_connections`.
55    pub min_connections: u32,
56
57    /// Maximum connections that allowed to be created.
58    ///
59    /// Must be greater than `0`.
60    pub max_connections: u32,
61}
62
63impl DatabaseConfiguration {
64    /**
65    Create a new database configuration with some defaults set.
66
67    **Defaults**:
68    - `min_connections`: 1
69    - `max_connections`: 10
70
71    **Parameter**:
72    - `driver`: [`DatabaseDriver`]: Configuration of the database driver.
73    */
74    pub fn new(driver: DatabaseDriver) -> Self {
75        DatabaseConfiguration {
76            driver,
77            min_connections: 1,
78            max_connections: 10,
79        }
80    }
81}
82
83/// Handle to a pool of database connections
84///
85/// Executing sql statements is done through the [`Executor`] trait
86/// which is implemented on `&Database`.
87///
88/// Common operations are implemented as functions in the [`database`](self) module.
89///
90/// Cloning is cheap i.e. two `Arc`s.
91#[derive(Clone)]
92pub struct Database(pub(crate) AnyPool, Arc<()>);
93
94impl Database {
95    /// Connects to the database using `configuration`
96    pub async fn connect(configuration: DatabaseConfiguration) -> Result<Self, Error> {
97        Ok(Self(AnyPool::connect(configuration).await?, Arc::new(())))
98    }
99
100    /// Starts a new transaction
101    ///
102    /// `&mut Transaction` implements [`Executor`] like `&Database` does
103    /// but its database operations can be reverted using [`Transaction::rollback`]
104    /// or simply dropping the transaction without calling [`Transaction::commit`].
105    pub async fn start_transaction(&self) -> Result<Transaction, Error> {
106        Ok(Transaction::new(self.0.begin().await?))
107    }
108
109    /// Closes the database connection
110    ///
111    /// While calling this method is not strictly necessary,
112    /// terminating your program without it
113    /// might result in some final queries not being flushed properly.
114    ///
115    /// This method consumes the database handle,
116    /// but actually all handles created using `clone` will become invalid after this call.
117    /// This means any further operation would result in an `Err`
118    pub async fn close(self) {
119        self.0.close().await;
120    }
121}
122
123#[cfg(feature = "postgres-only")]
124impl Database {
125    /// Accesses the raw underlying connection pool
126    ///
127    /// This can be used to access some postgres specific features
128    /// which sqlx implements but rorm doesn't (yet).
129    pub fn as_pool(&self) -> &sqlx::postgres::PgPool {
130        match &self.0 {
131            AnyPool::Postgres(x) => x,
132        }
133    }
134}
135
136impl Drop for Database {
137    /// Checks whether [`Database::close`] has been called before the last instance is dropped
138    fn drop(&mut self) {
139        // The use of strong_count should be correct:
140        // - the arc is private and we don't create WeakRefs
141        // => when observing a strong_count of 1, there can't be any remaining refs
142        if Arc::strong_count(&self.1) == 1 && !self.0.is_closed() {
143            warn!("Database has been dropped without calling close. This might cause the last queries to not be flushed properly");
144        }
145    }
146}
147
148/// Executes a simple `SELECT` query.
149///
150/// It is generic over a [`QueryStrategy`] which specifies how and how many rows to query.
151///
152/// **Parameter**:
153/// - `model`: Model to query.
154/// - `columns`: Columns to retrieve values from.
155/// - `joins`: Join tables expressions.
156/// - `conditions`: Optional conditions to apply.
157/// - `order_by_clause`: Columns to order the rows by.
158/// - `limit`: Optional limit / offset to apply to the query.
159///   Depending on the query strategy, this is either [`LimitClause`](rorm_sql::limit_clause::LimitClause)
160///   (for [`All`] and [`Stream`](crate::executor::Stream))
161///   or a simple [`u64`] (for [`One`] and [`Optional`](crate::executor::Optional)).
162#[allow(clippy::too_many_arguments)] // TODO: refactor this API, clippy is right
163pub fn query<'exe, 'data, Q: QueryStrategy + GetLimitClause>(
164    executor: impl Executor<'exe>,
165    model: &str,
166    columns: &[ColumnSelector<'data>],
167    joins: &[JoinTable<'data, 'data>],
168    conditions: Option<&conditional::Condition<'data>>,
169    order_by_clause: &[OrderByEntry<'_>],
170    limit: Option<Q::LimitOrOffset>,
171    distinct: bool,
172    #[cfg(feature = "postgres-only")] locking_clause: Option<LockingClause>,
173) -> Q::Result<'exe> {
174    let columns: Vec<_> = columns
175        .iter()
176        .map(|c| {
177            executor.dialect().select_column(
178                c.table_name,
179                c.column_name,
180                c.select_alias,
181                c.aggregation,
182            )
183        })
184        .collect();
185    let joins: Vec<_> = joins
186        .iter()
187        .map(|j| {
188            executor.dialect().join_table(
189                j.join_type,
190                j.table_name,
191                j.join_alias,
192                j.join_condition.clone(),
193            )
194        })
195        .collect();
196    let mut q = executor
197        .dialect()
198        .select(&columns, model, &joins, order_by_clause);
199
200    if let Some(condition) = conditions {
201        q = q.where_clause(condition);
202    }
203
204    if let Some(limit) = Q::get_limit_clause(limit) {
205        q = q.limit_clause(limit);
206    }
207
208    if distinct {
209        q = q.distinct();
210    }
211
212    #[cfg(feature = "postgres-only")]
213    if let Some(x) = locking_clause {
214        q = q.locking_clause(x);
215    }
216
217    let (query_string, bind_params) = q.build();
218
219    executor.execute::<Q>(query_string, bind_params)
220}
221
222/// Inserts a single row and returns columns from it.
223///
224/// **Parameter**:
225/// - `model`: Table to insert to
226/// - `columns`: Columns to set `values` for.
227/// - `values`: Values to bind to the corresponding columns.
228/// - `returning`: Columns to query from the inserted row.
229pub async fn insert_returning(
230    executor: impl Executor<'_>,
231    model: &str,
232    columns: &[&str],
233    values: &[Value<'_>],
234    returning: &[&str],
235) -> Result<Row, Error> {
236    generic_insert::<One>(executor, model, columns, values, Some(returning)).await
237}
238
239/// Inserts a single row.
240///
241/// **Parameter**:
242/// - `model`: Table to insert to
243/// - `columns`: Columns to set `values` for.
244/// - `values`: Values to bind to the corresponding columns.
245pub async fn insert(
246    executor: impl Executor<'_>,
247    model: &str,
248    columns: &[&str],
249    values: &[Value<'_>],
250) -> Result<(), Error> {
251    generic_insert::<Nothing>(executor, model, columns, values, None).await
252}
253
254/// Generic implementation of:
255/// - [`Database::insert`]
256/// - [`Database::insert_returning`]
257pub(crate) fn generic_insert<'exe, Q: QueryStrategy>(
258    executor: impl Executor<'exe>,
259    model: &str,
260    columns: &[&str],
261    values: &[Value<'_>],
262    returning: Option<&[&str]>,
263) -> Q::Result<'exe> {
264    let values = &[values];
265    let q = executor.dialect().insert(model, columns, values, returning);
266
267    let (query_string, bind_params) = q.build();
268
269    executor.execute::<Q>(query_string, bind_params)
270}
271
272/// This method is used to bulk insert rows.
273///
274/// If one insert statement fails, the complete operation will be rolled back.
275///
276/// **Parameter**:
277/// - `model`: Table to insert to
278/// - `columns`: Columns to set `rows` for.
279/// - `rows`: List of values to bind to the corresponding columns.
280/// - `transaction`: Optional transaction to execute the query on.
281pub async fn insert_bulk(
282    executor: impl Executor<'_>,
283    model: &str,
284    columns: &[&str],
285    rows: &[&[Value<'_>]],
286) -> Result<(), Error> {
287    let mut tr = executor.ensure_transaction().await?;
288
289    for chunk in rows.chunks(25) {
290        let mut insert = tr.dialect().insert(model, columns, chunk, None);
291        insert = insert.rollback_transaction();
292        let (insert_query, insert_params) = insert.build();
293
294        tr.execute::<Nothing>(insert_query, insert_params).await?;
295    }
296
297    tr.commit_if_owned().await.map_err(|x| match x {
298        TransactionError::Database(x) => x,
299        TransactionError::Hook(_) => {
300            unreachable!("Potentially create transaction does not use hooks")
301        }
302    })?;
303
304    Ok(())
305}
306
307/// This method is used to bulk insert rows.
308///
309/// If one insert statement fails, the complete operation will be rolled back.
310///
311/// **Parameter**:
312/// - `model`: Table to insert to
313/// - `columns`: Columns to set `rows` for.
314/// - `rows`: List of values to bind to the corresponding columns.
315/// - `transaction`: Optional transaction to execute the query on.
316pub async fn insert_bulk_returning(
317    executor: impl Executor<'_>,
318    model: &str,
319    columns: &[&str],
320    rows: &[&[Value<'_>]],
321    returning: &[&str],
322) -> Result<Vec<Row>, Error> {
323    let mut tr = executor.ensure_transaction().await?;
324
325    let mut inserted = Vec::with_capacity(rows.len());
326    for chunk in rows.chunks(25) {
327        let mut insert = tr.dialect().insert(model, columns, chunk, Some(returning));
328        insert = insert.rollback_transaction();
329        let (insert_query, insert_params) = insert.build();
330
331        inserted.extend(tr.execute::<All>(insert_query, insert_params).await?);
332    }
333
334    tr.commit_if_owned().await.map_err(|x| match x {
335        TransactionError::Database(x) => x,
336        TransactionError::Hook(_) => {
337            unreachable!("Potentially create transaction does not use hooks")
338        }
339    })?;
340
341    Ok(inserted)
342}
343
344/// This method is used to delete rows from a table.
345///
346/// **Parameter**:
347/// - `model`: Name of the model to delete rows from
348/// - `condition`: Optional condition to apply.
349/// - `transaction`: Optional transaction to execute the query on.
350///
351/// **Returns** the rows affected of the delete statement. Note that this also includes
352/// relations, etc.
353pub async fn delete<'post_build>(
354    executor: impl Executor<'_>,
355    model: &str,
356    condition: Option<&conditional::Condition<'post_build>>,
357) -> Result<u64, Error> {
358    let mut q = executor.dialect().delete(model);
359    if let Some(condition) = condition {
360        q = q.where_clause(condition);
361    }
362
363    let (query_string, bind_params) = q.build();
364
365    executor
366        .execute::<AffectedRows>(query_string, bind_params)
367        .await
368}
369
370/// This method is used to update rows in a table.
371///
372/// **Parameter**:
373/// - `model`: Name of the model to update rows from
374/// - `updates`: A list of updates. An update is a tuple that consists of a list of columns to
375///   update as well as the value to set to the columns.
376/// - `condition`: Optional condition to apply.
377/// - `transaction`: Optional transaction to execute the query on.
378///
379/// **Returns** the rows affected from the update statement. Note that this also includes
380/// relations, etc.
381pub async fn update<'post_build>(
382    executor: impl Executor<'_>,
383    model: &str,
384    updates: &[(&str, Value<'post_build>)],
385    condition: Option<&conditional::Condition<'post_build>>,
386) -> Result<u64, Error> {
387    let mut stmt = executor.dialect().update(model);
388
389    for (column, value) in updates {
390        stmt = stmt.add_update(column, *value);
391    }
392
393    if let Some(cond) = condition {
394        stmt = stmt.where_clause(cond);
395    }
396
397    let (query_string, bind_params) = stmt.build()?;
398
399    executor
400        .execute::<AffectedRows>(query_string, bind_params)
401        .await
402}
rorm_db/database.rs

rorm_db/
database.rs
