toolu-orm-query 0.2.0

Type-safe query builders for toolu-orm
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121

//! Tokio-postgres [`Executor`] implementation and transaction wrapper.
//!
//! # Public API
//!
//! - [`Executor`] — async execute/query for `tokio_postgres::Client`
//! - [`PgTransaction`] — transaction wrapper implementing [`Executor`]

use toolu_orm_core::error::DbCoreError;
use toolu_orm_core::row::FromRow;
use toolu_orm_core::value::Value;

use crate::QueryError;

#[async_trait::async_trait]
pub trait Executor {
  async fn execute_sql(&self, sql: &str, params: Vec<Value>) -> Result<u64, QueryError>;

  async fn query_map<T: FromRow + Send>(
    &self,
    sql: &str,
    params: Vec<Value>,
  ) -> Result<Vec<T>, QueryError>;
}

pub(crate) fn pg_param_slice(
  pg_params: &[Box<dyn tokio_postgres::types::ToSql + Sync + Send>],
) -> Vec<&(dyn tokio_postgres::types::ToSql + Sync)> {
  pg_params
    .iter()
    .map(|p| p.as_ref() as &(dyn tokio_postgres::types::ToSql + Sync))
    .collect()
}

async fn pg_execute(
  client: &(impl tokio_postgres::GenericClient + Sync),
  sql: &str,
  params: Vec<Value>,
) -> Result<u64, QueryError> {
  let pg_params = toolu_orm_core::value::to_pg_params(&params);
  let refs = pg_param_slice(&pg_params);
  let affected = client.execute(sql, &refs[..]).await?;
  Ok(affected)
}

async fn pg_query_map<T: FromRow + Send>(
  client: &(impl tokio_postgres::GenericClient + Sync),
  sql: &str,
  params: Vec<Value>,
) -> Result<Vec<T>, QueryError> {
  let pg_params = toolu_orm_core::value::to_pg_params(&params);
  let refs = pg_param_slice(&pg_params);
  let rows = client.query(sql, &refs[..]).await?;
  let mut results = Vec::new();
  for row in rows.iter() {
    results.push(toolu_orm_core::row::from_postgres_row(row)?);
  }
  Ok(results)
}

#[async_trait::async_trait]
impl Executor for tokio_postgres::Client {
  async fn execute_sql(&self, sql: &str, params: Vec<Value>) -> Result<u64, QueryError> {
    pg_execute(self, sql, params).await
  }

  async fn query_map<T: FromRow + Send>(
    &self,
    sql: &str,
    params: Vec<Value>,
  ) -> Result<Vec<T>, QueryError> {
    pg_query_map(self, sql, params).await
  }
}

/// Wrapper around `tokio_postgres::Transaction` that implements [`Executor`].
pub struct PgTransaction<'a> {
  inner: tokio_postgres::Transaction<'a>,
}

impl<'a> PgTransaction<'a> {
  pub fn new(txn: tokio_postgres::Transaction<'a>) -> Self {
    Self { inner: txn }
  }

  /// # Errors
  ///
  /// Returns [`DbCoreError::Connection`] if the commit RPC fails.
  pub async fn commit(self) -> Result<(), DbCoreError> {
    self
      .inner
      .commit()
      .await
      .map_err(|e| DbCoreError::Connection(e.to_string()))
  }

  /// # Errors
  ///
  /// Returns [`DbCoreError::Connection`] if the rollback RPC fails.
  pub async fn rollback(self) -> Result<(), DbCoreError> {
    self
      .inner
      .rollback()
      .await
      .map_err(|e| DbCoreError::Connection(e.to_string()))
  }
}

#[async_trait::async_trait]
impl<'a> Executor for PgTransaction<'a> {
  async fn execute_sql(&self, sql: &str, params: Vec<Value>) -> Result<u64, QueryError> {
    pg_execute(&self.inner, sql, params).await
  }

  async fn query_map<T: FromRow + Send>(
    &self,
    sql: &str,
    params: Vec<Value>,
  ) -> Result<Vec<T>, QueryError> {
    pg_query_map(&self.inner, sql, params).await
  }
}