use std::ops::DerefMut;
use tokio_postgres::{Client, Statement, Transaction as PgTransaction};
use crate::adapters::params::convert_params;
use crate::middleware::{ConversionMode, CustomDbRow, ResultSet, RowValues, SqlMiddlewareDbError};
use crate::tx_outcome::TxOutcome;
use super::{Params, build_result_set};
use crate::postgres::query::{build_result_set_from_rows, convert_affected_rows};
pub struct Tx<'a> {
tx: PgTransaction<'a>,
}
pub struct Prepared {
stmt: Statement,
}
pub async fn begin_transaction<C>(conn: &mut C) -> Result<Tx<'_>, SqlMiddlewareDbError>
where
C: DerefMut<Target = Client>,
{
let tx = conn.deref_mut().transaction().await?;
Ok(Tx { tx })
}
impl<'conn> Tx<'conn> {
pub async fn prepare(&self, sql: &str) -> Result<Prepared, SqlMiddlewareDbError> {
let stmt = self.tx.prepare(sql).await?;
Ok(Prepared { stmt })
}
#[must_use]
pub fn select<'tx, 'prepared>(
&'tx self,
prepared: &'prepared Prepared,
) -> PreparedSelect<'tx, 'prepared, 'static, 'conn> {
PreparedSelect {
tx: self,
prepared,
params: &[],
}
}
#[must_use]
pub fn execute<'tx, 'prepared>(
&'tx self,
prepared: &'prepared Prepared,
) -> PreparedExecute<'tx, 'prepared, 'static, 'conn> {
PreparedExecute {
tx: self,
prepared,
params: &[],
}
}
pub(crate) async fn execute_prepared(
&self,
prepared: &Prepared,
params: &[RowValues],
) -> Result<usize, SqlMiddlewareDbError> {
let converted = convert_params::<Params>(params, ConversionMode::Execute)?;
let rows = self.tx.execute(&prepared.stmt, converted.as_refs()).await?;
convert_affected_rows(rows, "Invalid rows affected count")
}
pub async fn execute_dml(
&self,
query: &str,
params: &[RowValues],
) -> Result<usize, SqlMiddlewareDbError> {
let converted = convert_params::<Params>(params, ConversionMode::Execute)?;
let rows = self.tx.execute(query, converted.as_refs()).await?;
convert_affected_rows(rows, "Invalid rows affected count")
}
pub(crate) async fn query_prepared(
&self,
prepared: &Prepared,
params: &[RowValues],
) -> Result<ResultSet, SqlMiddlewareDbError> {
let converted = convert_params::<Params>(params, ConversionMode::Query)?;
build_result_set(&prepared.stmt, converted.as_refs(), &self.tx).await
}
pub(crate) async fn query_prepared_optional(
&self,
prepared: &Prepared,
params: &[RowValues],
) -> Result<Option<CustomDbRow>, SqlMiddlewareDbError> {
self.query_prepared(prepared, params)
.await
.map(ResultSet::into_optional)
}
pub(crate) async fn query_prepared_one(
&self,
prepared: &Prepared,
params: &[RowValues],
) -> Result<CustomDbRow, SqlMiddlewareDbError> {
self.query_prepared(prepared, params).await?.into_one()
}
pub(crate) async fn query_prepared_map_one<T, F>(
&self,
prepared: &Prepared,
params: &[RowValues],
mapper: F,
) -> Result<T, SqlMiddlewareDbError>
where
F: FnOnce(&tokio_postgres::Row) -> Result<T, SqlMiddlewareDbError>,
{
self.query_prepared_map_optional(prepared, params, mapper)
.await?
.ok_or_else(|| SqlMiddlewareDbError::ExecutionError("query returned no rows".into()))
}
pub(crate) async fn query_prepared_map_optional<T, F>(
&self,
prepared: &Prepared,
params: &[RowValues],
mapper: F,
) -> Result<Option<T>, SqlMiddlewareDbError>
where
F: FnOnce(&tokio_postgres::Row) -> Result<T, SqlMiddlewareDbError>,
{
let converted = convert_params::<Params>(params, ConversionMode::Query)?;
let row = self
.tx
.query_opt(&prepared.stmt, converted.as_refs())
.await?;
row.as_ref().map(mapper).transpose()
}
pub async fn query(
&self,
query: &str,
params: &[RowValues],
) -> Result<ResultSet, SqlMiddlewareDbError> {
let converted = convert_params::<Params>(params, ConversionMode::Query)?;
let rows = self.tx.query(query, converted.as_refs()).await?;
build_result_set_from_rows(&rows)
}
pub async fn execute_batch(&self, sql: &str) -> Result<(), SqlMiddlewareDbError> {
self.tx.batch_execute(sql).await?;
Ok(())
}
pub async fn commit(self) -> Result<TxOutcome, SqlMiddlewareDbError> {
self.tx.commit().await?;
Ok(TxOutcome::without_restored_connection())
}
pub async fn rollback(self) -> Result<TxOutcome, SqlMiddlewareDbError> {
self.tx.rollback().await?;
Ok(TxOutcome::without_restored_connection())
}
}
pub struct PreparedExecute<'tx, 'prepared, 'params, 'conn> {
tx: &'tx Tx<'conn>,
prepared: &'prepared Prepared,
params: &'params [RowValues],
}
impl<'tx, 'prepared, 'params, 'conn> PreparedExecute<'tx, 'prepared, 'params, 'conn> {
#[must_use]
pub fn params<'next>(
self,
params: &'next [RowValues],
) -> PreparedExecute<'tx, 'prepared, 'next, 'conn> {
PreparedExecute {
tx: self.tx,
prepared: self.prepared,
params,
}
}
pub async fn run(self) -> Result<usize, SqlMiddlewareDbError> {
self.tx.execute_prepared(self.prepared, self.params).await
}
}
pub struct PreparedSelect<'tx, 'prepared, 'params, 'conn> {
tx: &'tx Tx<'conn>,
prepared: &'prepared Prepared,
params: &'params [RowValues],
}
impl<'tx, 'prepared, 'params, 'conn> PreparedSelect<'tx, 'prepared, 'params, 'conn> {
#[must_use]
pub fn params<'next>(
self,
params: &'next [RowValues],
) -> PreparedSelect<'tx, 'prepared, 'next, 'conn> {
PreparedSelect {
tx: self.tx,
prepared: self.prepared,
params,
}
}
pub async fn all(self) -> Result<ResultSet, SqlMiddlewareDbError> {
self.tx.query_prepared(self.prepared, self.params).await
}
pub async fn optional(self) -> Result<Option<CustomDbRow>, SqlMiddlewareDbError> {
self.tx
.query_prepared_optional(self.prepared, self.params)
.await
}
pub async fn one(self) -> Result<CustomDbRow, SqlMiddlewareDbError> {
self.tx.query_prepared_one(self.prepared, self.params).await
}
pub async fn map_one<T, F>(self, mapper: F) -> Result<T, SqlMiddlewareDbError>
where
F: FnOnce(&tokio_postgres::Row) -> Result<T, SqlMiddlewareDbError>,
{
self.tx
.query_prepared_map_one(self.prepared, self.params, mapper)
.await
}
pub async fn map_optional<T, F>(self, mapper: F) -> Result<Option<T>, SqlMiddlewareDbError>
where
F: FnOnce(&tokio_postgres::Row) -> Result<T, SqlMiddlewareDbError>,
{
self.tx
.query_prepared_map_optional(self.prepared, self.params, mapper)
.await
}
}