use crate::abstract_layer::DbType;
use crate::abstract_layer::common_helpers;
use crate::model::{DbBackendTypeMapper, Model, Row, Value};
use crate::query::builder::{
FourTableSelect, InnerJoinedSelect, LeftJoinedSelect, MultiTableSelect, RelatedSelect,
RightJoinedSelect, Select, WhereExpr,
};
use crate::query::filter::FilterExpr;
use std::collections::HashMap;
use std::marker::PhantomData;
use tokio_postgres::NoTls;
pub struct PostgreSQLTypeMapper;
impl DbBackendTypeMapper for PostgreSQLTypeMapper {
fn sql_type(
rust_type: &str,
is_primary: bool,
is_auto_increment: bool,
is_nullable: bool,
) -> String {
if is_primary {
if is_auto_increment {
let serial_type = match rust_type {
"i8" | "i16" | "i32" => "SERIAL",
"i64" | "u16" | "u32" | "u64" => "BIGSERIAL",
"u8" => "SMALLSERIAL", _ => "SERIAL", };
return format!("{serial_type} PRIMARY KEY");
} else {
let int_type = match rust_type {
"i8" | "i16" | "u8" => "SMALLINT",
"i32" | "u16" | "u32" => "INTEGER",
"i64" | "u64" => "BIGINT",
_ => "INTEGER",
};
return format!("{int_type} PRIMARY KEY");
}
}
let base_type = match rust_type {
"i8" => "SMALLINT",
"i16" => "SMALLINT",
"i32" => "INTEGER",
"i64" => "BIGINT",
"u8" => "SMALLINT",
"u16" => "INTEGER",
"u32" => "BIGINT",
"u64" => "BIGINT",
"f32" => "REAL",
"f64" => "DOUBLE PRECISION",
"String" => "VARCHAR",
"bool" => "BOOLEAN",
"Vec<u8>" | "&[u8]" => "BYTEA",
"Uuid" | "uuid::Uuid" => "UUID",
"DateTime" | "chrono::DateTime" | "NaiveDateTime" | "chrono::NaiveDateTime" => {
"TIMESTAMP"
}
"NaiveDate" | "chrono::NaiveDate" => "DATE",
"NaiveTime" | "chrono::NaiveTime" => "TIME",
"JsonValue" | "serde_json::Value" => "JSONB",
_ => "TEXT",
};
let mut sql_type = base_type.to_string();
if !is_nullable {
sql_type.push_str(" NOT NULL");
}
sql_type
}
}
#[allow(dead_code)]
pub struct Database {
client: tokio_postgres::Client,
connection_handle: tokio::task::JoinHandle<Result<(), tokio_postgres::Error>>,
}
impl Database {
pub async fn connect(
_db_type: super::DbType,
connection_string: &str,
) -> Result<Self, crate::Error> {
let (client, connection) = tokio_postgres::connect(connection_string, NoTls)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let connection_handle = tokio::spawn(async move {
if let Err(e) = connection.await {
eprintln!("PostgreSQL connection error: {e}");
}
Ok(())
});
Ok(Self {
client,
connection_handle,
})
}
pub async fn create_table<T: Model>(&self) -> Result<(), crate::Error> {
let table_exists = self.check_table_exists::<T>().await?;
if table_exists {
self.validate_table_schema::<T>().await?;
return Ok(());
}
let create_sql =
crate::generate_create_table_sql::<T>(crate::abstract_layer::DbType::PostgreSQL);
self.client
.execute(&create_sql, &[])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(())
}
async fn check_table_exists<T: Model>(&self) -> Result<bool, crate::Error> {
let sql = "SELECT COUNT(*) FROM information_schema.tables WHERE table_type='BASE TABLE' AND table_name=$1";
let row = self
.client
.query_one(sql, &[&T::TABLE_NAME])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let count: i64 = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(count > 0)
}
async fn validate_table_schema<T: Model>(&self) -> Result<(), crate::Error> {
let sql = r#"
SELECT column_name, data_type, is_nullable
FROM information_schema.columns
WHERE table_name = $1
ORDER BY ordinal_position
"#;
let rows = self
.client
.query(sql, &[&T::TABLE_NAME])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut actual_columns: Vec<(String, String, bool)> = Vec::new();
for row in rows {
let name: String = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
let col_type: String = row
.try_get(1)
.map_err(|e| crate::Error::Database(e.to_string()))?;
let is_nullable: String = row
.try_get(2)
.map_err(|e| crate::Error::Database(e.to_string()))?;
actual_columns.push((name, col_type, is_nullable == "YES"));
}
if actual_columns.len() != T::COLUMNS.len() {
return Err(crate::Error::SchemaMismatch {
table: T::TABLE_NAME.to_string(),
reason: format!(
"Column count mismatch: expected {}, but actual is {}",
T::COLUMNS.len(),
actual_columns.len()
),
});
}
for (i, expected_col) in T::COLUMN_SCHEMA.iter().enumerate() {
if i >= actual_columns.len() {
return Err(crate::Error::SchemaMismatch {
table: T::TABLE_NAME.to_string(),
reason: format!("Missing column: {}", expected_col.name),
});
}
let (actual_name, actual_type, actual_nullable) = &actual_columns[i];
if actual_name != expected_col.name {
return Err(crate::Error::SchemaMismatch {
table: T::TABLE_NAME.to_string(),
reason: format!(
"Column name mismatch at position {i}: expected '{}', but actual is '{actual_name}'",
expected_col.name
),
});
}
let expected_type = crate::abstract_layer::DbType::PostgreSQL.sql_type(
expected_col.rust_type,
expected_col.is_primary,
expected_col.is_auto_increment,
expected_col.is_nullable,
);
let type_to_compare = if expected_col.is_primary {
match expected_col.rust_type {
"i8" | "i16" | "i32" => "SMALLINT".to_string(),
"i64" | "u16" | "u32" | "u64" => "BIGINT".to_string(),
"u8" => "SMALLINT".to_string(),
_ => "INTEGER".to_string(),
}
} else {
let full_type = crate::abstract_layer::DbType::PostgreSQL.sql_type(
expected_col.rust_type,
false,
expected_col.is_auto_increment,
expected_col.is_nullable,
);
full_type.replace(" NOT NULL", "")
};
if !self.types_compatible(actual_type, &type_to_compare) {
return Err(crate::Error::SchemaMismatch {
table: T::TABLE_NAME.to_string(),
reason: format!(
"Column type mismatch for '{}': expected '{expected_type}', but actual is '{actual_type}'",
expected_col.name
),
});
}
if !expected_col.is_primary {
let expected_nullable = expected_col.is_nullable;
if *actual_nullable != expected_nullable {
return Err(crate::Error::SchemaMismatch {
table: T::TABLE_NAME.to_string(),
reason: format!(
"Column nullability mismatch for '{}': expected {}NULL, but actual is {}NULL",
expected_col.name,
if expected_nullable { "" } else { "NOT " },
if *actual_nullable { "" } else { "NOT " }
),
});
}
}
}
Ok(())
}
fn types_compatible(&self, actual: &str, expected: &str) -> bool {
fn normalize(s: &str) -> String {
let upper = s.to_uppercase();
match upper.as_str() {
"SMALLINT" | "INT2" => "SMALLINT".to_string(),
"INTEGER" | "INT" | "INT4" | "SERIAL" => "INTEGER".to_string(),
"BIGINT" | "INT8" | "BIGSERIAL" => "BIGINT".to_string(),
"CHARACTER VARYING" | "VARCHAR" | "TEXT" | "CHAR" | "CHARACTER" | "BPCHAR" => {
"VARCHAR".to_string()
}
"BOOLEAN" | "BOOL" => "BOOLEAN".to_string(),
"REAL" | "FLOAT4" => "REAL".to_string(),
"DOUBLE PRECISION" | "FLOAT8" | "FLOAT" => "DOUBLE PRECISION".to_string(),
"BYTEA" | "BLOB" => "BYTEA".to_string(),
_ => upper,
}
}
normalize(actual) == normalize(expected)
}
pub async fn insert<T: Model>(&self, model: &T) -> Result<(), crate::Error> {
self.insert_batch::<T>(&[model]).await
}
pub async fn insert_or_update<T: Model>(&self, model: &T) -> Result<(), crate::Error> {
self.insert_or_update_batch::<T>(&[model]).await
}
pub async fn insert_batch<T: Model>(&self, models: &[&T]) -> Result<(), crate::Error> {
if models.is_empty() {
return Ok(());
}
let columns = T::COLUMNS.join(", ");
let col_count = T::COLUMNS.len();
let mut sql = format!("INSERT INTO {} ({columns}) VALUES ", T::TABLE_NAME);
let mut all_values = Vec::new();
let mut param_idx = 1;
for (idx, model) in models.iter().enumerate() {
if idx > 0 {
sql.push_str(", ");
}
let placeholders: Vec<String> = (1..=col_count)
.map(|i| format!("${}", param_idx + i - 1))
.collect();
sql.push_str(&format!("({})", placeholders.join(", ")));
param_idx += col_count;
let values = model.field_values();
all_values.extend(values);
}
let params = values_to_params(&all_values)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> =
params.iter().map(|p| p.as_ref()).collect();
self.client
.execute(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(())
}
pub async fn insert_or_update_batch<T: Model>(
&self,
models: &[&T],
) -> Result<(), crate::Error> {
if models.is_empty() {
return Ok(());
}
let columns = T::COLUMNS.join(", ");
let col_count = T::COLUMNS.len();
let primary_key = T::primary_key_column();
let mut sql = format!("INSERT INTO {} ({columns}) VALUES ", T::TABLE_NAME);
let mut all_values = Vec::new();
let mut param_idx = 1;
for (idx, model) in models.iter().enumerate() {
if idx > 0 {
sql.push_str(", ");
}
let placeholders: Vec<String> = (1..=col_count)
.map(|i| format!("${}", param_idx + i - 1))
.collect();
sql.push_str(&format!("({})", placeholders.join(", ")));
param_idx += col_count;
let values = model.field_values();
all_values.extend(values);
}
sql.push_str(&format!(" ON CONFLICT ({primary_key}) DO UPDATE SET "));
let mut first = true;
for col_name in T::COLUMNS.iter() {
if col_name == &primary_key {
continue; }
if !first {
sql.push_str(", ");
}
sql.push_str(&format!("{col_name} = EXCLUDED.{col_name}"));
first = false;
}
let params = values_to_params(&all_values)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> =
params.iter().map(|p| p.as_ref()).collect();
self.client
.execute(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(())
}
pub fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
SelectExecutor {
select: Select::<T>::new(),
client: &self.client,
_marker: PhantomData,
}
}
pub fn delete<T: Model>(&self) -> DeleteExecutor<'_, T> {
DeleteExecutor {
filters: Vec::new(),
client: &self.client,
_marker: PhantomData,
}
}
pub fn update<T: Model>(&self) -> UpdateExecutor<'_, T> {
UpdateExecutor {
sets: Vec::new(),
filters: Vec::new(),
client: &self.client,
_marker: PhantomData,
}
}
pub fn related<T: Model + 'static, R: Model>(&self) -> RelatedSelectExecutor<'_, T, R> {
RelatedSelectExecutor {
select: Select::<T>::new().from::<T, R>(),
client: &self.client,
_marker: PhantomData,
}
}
pub async fn begin(&self) -> Result<Transaction<'_>, crate::Error> {
self.client
.execute("BEGIN", &[])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(Transaction {
client: &self.client,
committed: false,
rolled_back: false,
})
}
pub async fn drop_table<T: Model>(&self) -> Result<(), crate::Error> {
let sql = format!("DROP TABLE IF EXISTS {}", T::TABLE_NAME);
self.client
.execute(&sql, &[])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(())
}
pub async fn exec_table<T: Model>(&self, sql: &str) -> Result<Vec<T>, crate::Error> {
let rows = self
.client
.query(sql, &[])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut results = Vec::new();
for row in rows {
let mut data = HashMap::new();
for (i, col_name) in T::COLUMNS.iter().enumerate() {
let ormer_value = convert_postgres_value(&row, i)?;
data.insert(col_name.to_string(), ormer_value);
}
let ormer_row = Row::new(data);
let model = T::from_row(&ormer_row)?;
results.push(model);
}
Ok(results)
}
pub async fn exec_non_query(&self, sql: &str) -> Result<u64, crate::Error> {
let result = self
.client
.execute(sql, &[])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(result)
}
pub async fn is_valid(&self) -> bool {
self.client.execute("SELECT 1", &[]).await.is_ok()
}
}
pub struct Transaction<'a> {
client: &'a tokio_postgres::Client,
committed: bool,
rolled_back: bool,
}
impl<'a> Transaction<'a> {
pub async fn commit(mut self) -> Result<(), crate::Error> {
if self.committed || self.rolled_back {
return Err(crate::Error::Database(
"Transaction already committed or rolled back".to_string(),
));
}
self.client
.execute("COMMIT", &[])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
self.committed = true;
Ok(())
}
pub async fn rollback(mut self) -> Result<(), crate::Error> {
if self.committed || self.rolled_back {
return Err(crate::Error::Database(
"Transaction already committed or rolled back".to_string(),
));
}
self.client
.execute("ROLLBACK", &[])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
self.rolled_back = true;
Ok(())
}
pub fn select<T: Model>(&self) -> SelectExecutor<'_, T> {
SelectExecutor {
select: Select::<T>::new(),
client: self.client,
_marker: PhantomData,
}
}
pub fn delete<T: Model>(&self) -> DeleteExecutor<'_, T> {
DeleteExecutor {
filters: Vec::new(),
client: self.client,
_marker: PhantomData,
}
}
pub fn update<T: Model>(&self) -> UpdateExecutor<'_, T> {
UpdateExecutor {
sets: Vec::new(),
filters: Vec::new(),
client: self.client,
_marker: PhantomData,
}
}
pub async fn insert<T: Model>(&self, model: &T) -> Result<(), crate::Error> {
self.insert_batch::<T>(&[model]).await
}
pub async fn insert_or_update<T: Model>(&self, model: &T) -> Result<(), crate::Error> {
self.insert_or_update_batch::<T>(&[model]).await
}
pub async fn insert_batch<T: Model>(&self, models: &[&T]) -> Result<(), crate::Error> {
if models.is_empty() {
return Ok(());
}
let (sql, _) = crate::abstract_layer::common_helpers::build_batch_insert_sql_postgresql::<T>(
models.len(),
);
let all_values =
crate::abstract_layer::common_helpers::collect_batch_insert_values::<T>(models);
let params = values_to_params(&all_values)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> =
params.iter().map(|p| p.as_ref()).collect();
self.client
.execute(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(())
}
pub async fn insert_or_update_batch<T: Model>(
&self,
models: &[&T],
) -> Result<(), crate::Error> {
if models.is_empty() {
return Ok(());
}
let columns = T::COLUMNS.join(", ");
let col_count = T::COLUMNS.len();
let primary_key = T::primary_key_column();
let mut sql = format!("INSERT INTO {} ({columns}) VALUES ", T::TABLE_NAME);
let mut all_values = Vec::new();
let mut param_idx = 1;
for (idx, model) in models.iter().enumerate() {
if idx > 0 {
sql.push_str(", ");
}
let placeholders: Vec<String> = (1..=col_count)
.map(|i| format!("${}", param_idx + i - 1))
.collect();
sql.push_str(&format!("({})", placeholders.join(", ")));
param_idx += col_count;
let values = model.field_values();
all_values.extend(values);
}
sql.push_str(&format!(" ON CONFLICT ({primary_key}) DO UPDATE SET "));
let mut first = true;
for col_name in T::COLUMNS.iter() {
if col_name == &primary_key {
continue; }
if !first {
sql.push_str(", ");
}
sql.push_str(&format!("{col_name} = EXCLUDED.{col_name}"));
first = false;
}
let params = values_to_params(&all_values)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> =
params.iter().map(|p| p.as_ref()).collect();
self.client
.execute(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(())
}
}
pub struct LeftJoinedSelectExecutor<'a, T: Model, J: Model> {
select: LeftJoinedSelect<T, J>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<(T, J)>,
}
pub struct InnerJoinedSelectExecutor<'a, T: Model, J: Model> {
select: InnerJoinedSelect<T, J>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<(T, J)>,
}
pub struct RightJoinedSelectExecutor<'a, T: Model, J: Model> {
select: RightJoinedSelect<T, J>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<(T, J)>,
}
pub struct SelectExecutor<'a, T: Model> {
select: Select<T>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<T>,
}
impl<'a, T: Model> SelectExecutor<'a, T> {
pub fn filter<F>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> WhereExpr,
{
Self {
select: self.select.filter(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn order_by<F, O>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> O,
O: Into<crate::OrderBy>,
{
Self {
select: self.select.order_by(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn order_by_desc<F, O>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> O,
O: Into<crate::OrderBy>,
{
Self {
select: self.select.order_by_desc(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn range<RR: Into<crate::query::builder::RangeBounds>>(self, range: RR) -> Self {
Self {
select: self.select.range(range),
client: self.client,
_marker: PhantomData,
}
}
pub fn left_join<J: Model>(
self,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> LeftJoinedSelectExecutor<'a, T, J> {
LeftJoinedSelectExecutor {
select: self.select.left_join::<J>(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn inner_join<J: Model>(
self,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> InnerJoinedSelectExecutor<'a, T, J> {
InnerJoinedSelectExecutor {
select: self.select.inner_join::<J>(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn right_join<J: Model>(
self,
f: impl FnOnce(T::Where, J::Where) -> WhereExpr,
) -> RightJoinedSelectExecutor<'a, T, J> {
RightJoinedSelectExecutor {
select: self.select.right_join::<J>(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn collect<C: FromIterator<T> + 'static>(self) -> CollectFuture<'a, T, C> {
CollectFuture {
executor: self,
_marker: PhantomData,
}
}
pub fn count<F, C>(self, f: F) -> AggregateFuture<'a, T, usize>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C>,
{
let aggregate_select = self.select.count(f);
AggregateFuture {
aggregate_select,
client: self.client,
_marker: PhantomData,
}
}
pub fn sum<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C>,
C: crate::query::builder::AggregateResultType + 'static,
{
let aggregate_select = self.select.sum(f);
AggregateFuture {
aggregate_select,
client: self.client,
_marker: PhantomData,
}
}
pub fn avg<F, C>(self, f: F) -> AggregateFuture<'a, T, Option<f64>>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C>,
C: crate::query::builder::AggregateResultType + 'static,
{
let aggregate_select = self.select.avg(f);
AggregateFuture {
aggregate_select,
client: self.client,
_marker: PhantomData,
}
}
pub fn max<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C>,
C: crate::query::builder::AggregateResultType + 'static,
{
let aggregate_select = self.select.max(f);
AggregateFuture {
aggregate_select,
client: self.client,
_marker: PhantomData,
}
}
pub fn min<F, C>(self, f: F) -> AggregateFuture<'a, T, C::Output>
where
F: FnOnce(<T as Model>::Where) -> crate::query::builder::TypedColumn<C>,
C: crate::query::builder::AggregateResultType + 'static,
{
let aggregate_select = self.select.min(f);
AggregateFuture {
aggregate_select,
client: self.client,
_marker: PhantomData,
}
}
pub fn from<T2, R: Model>(self) -> RelatedSelectExecutor<'a, T, R>
where
T2: Model + 'static,
{
RelatedSelectExecutor {
select: self.select.from::<T2, R>(),
client: self.client,
_marker: PhantomData,
}
}
pub fn from3<T2, R1: Model, R2: Model>(self) -> MultiTableSelectExecutor<'a, T, R1, R2>
where
T2: Model + 'static,
{
MultiTableSelectExecutor {
select: self.select.from3::<T2, R1, R2>(),
client: self.client,
_marker: PhantomData,
}
}
pub fn from4<T2, R1: Model, R2: Model, R3: Model>(
self,
) -> FourTableSelectExecutor<'a, T, R1, R2, R3>
where
T2: Model + 'static,
{
FourTableSelectExecutor {
select: self.select.from4::<T2, R1, R2, R3>(),
client: self.client,
_marker: PhantomData,
}
}
}
pub struct CollectFuture<'a, T: Model, C: FromIterator<T>> {
executor: SelectExecutor<'a, T>,
_marker: PhantomData<C>,
}
pub struct AggregateFuture<'a, T: Model, R> {
aggregate_select: crate::query::builder::AggregateSelect<T, R>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<(T, R)>,
}
impl<'a, T: Model + 'static, R: crate::model::FromValue + 'static> std::future::IntoFuture
for AggregateFuture<'a, T, R>
{
type Output = Result<R, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move {
let (sql, params) = self.aggregate_select.to_sql_with_params(DbType::PostgreSQL);
let pg_params: Vec<Box<dyn postgres_types::ToSql + Sync>> = params
.into_iter()
.map(|v| match v {
crate::model::Value::Integer(i) => {
Box::new(i) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Text(t) => {
Box::new(t) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Real(r) => {
Box::new(r) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Null => {
Box::new(None::<i32>) as Box<dyn postgres_types::ToSql + Sync>
}
})
.collect();
let params_ref: Vec<&(dyn postgres_types::ToSql + Sync)> =
pg_params.iter().map(|p| p.as_ref()).collect();
let row = self
.client
.query_one(&sql, ¶ms_ref)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
use tokio_postgres::types::Type;
let column_type = row.columns()[0].type_();
let ormer_value = match *column_type {
Type::INT2 => {
let val: Option<i16> = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
val.map(|v| crate::model::Value::Integer(v as i64))
.unwrap_or(crate::model::Value::Null)
}
Type::INT4 => {
let val: Option<i32> = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
val.map(|v| crate::model::Value::Integer(v as i64))
.unwrap_or(crate::model::Value::Null)
}
Type::INT8 => {
let val: Option<i64> = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
val.map(|v| crate::model::Value::Integer(v))
.unwrap_or(crate::model::Value::Null)
}
Type::FLOAT4 => {
let val: Option<f32> = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
val.map(|v| crate::model::Value::Real(v as f64))
.unwrap_or(crate::model::Value::Null)
}
Type::FLOAT8 => {
let val: Option<f64> = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
val.map(|v| crate::model::Value::Real(v))
.unwrap_or(crate::model::Value::Null)
}
Type::NUMERIC => {
let val: Option<f64> = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
val.map(|v| crate::model::Value::Real(v))
.unwrap_or(crate::model::Value::Null)
}
Type::TEXT | Type::VARCHAR => {
let val: Option<String> = row
.try_get(0)
.map_err(|e| crate::Error::Database(e.to_string()))?;
val.map(|v| crate::model::Value::Text(v))
.unwrap_or(crate::model::Value::Null)
}
_ => crate::model::Value::Null,
};
R::from_value(&ormer_value)
})
}
}
impl<'a, T: Model + 'static, C: FromIterator<T> + 'static> std::future::IntoFuture
for CollectFuture<'a, T, C>
{
type Output = Result<C, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.executor.collect_inner().await })
}
}
impl<'a, T: Model> SelectExecutor<'a, T> {
async fn collect_inner<C: FromIterator<T>>(self) -> Result<C, crate::Error> {
let (sql, params) = self.select.to_sql_with_params(DbType::PostgreSQL);
let pg_params = values_to_params(¶ms)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = pg_params
.iter()
.map(|p| &**p as &(dyn tokio_postgres::types::ToSql + Sync))
.collect();
let rows = self
.client
.query(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut results = Vec::new();
for row in rows {
let mut data = HashMap::new();
for (i, col_name) in T::COLUMNS.iter().enumerate() {
let column_info = &T::COLUMN_SCHEMA[i];
let rust_type = column_info.rust_type;
let is_nullable = column_info.is_nullable;
let ormer_value = if is_nullable {
match rust_type {
"i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
let v: Option<i64> = row.get(i);
match v {
Some(val) => crate::model::Value::Integer(val),
None => crate::model::Value::Null,
}
}
"String" => {
let v: Option<String> = row.get(i);
match v {
Some(val) => crate::model::Value::Text(val),
None => crate::model::Value::Null,
}
}
"f32" | "f64" => {
let v: Option<f64> = row.get(i);
match v {
Some(val) => crate::model::Value::Real(val),
None => crate::model::Value::Null,
}
}
"bool" => {
let v: Option<bool> = row.get(i);
match v {
Some(true) => crate::model::Value::Integer(1),
Some(false) => crate::model::Value::Integer(0),
None => crate::model::Value::Null,
}
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported nullable column type: {rust_type}"
)));
}
}
} else {
match rust_type {
"i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
let v: i64 = row.get(i);
crate::model::Value::Integer(v)
}
"String" => {
let v: String = row.get(i);
crate::model::Value::Text(v)
}
"f32" | "f64" => {
let v: f64 = row.get(i);
crate::model::Value::Real(v)
}
"bool" => {
let v: bool = row.get(i);
if v {
crate::model::Value::Integer(1)
} else {
crate::model::Value::Integer(0)
}
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
}
};
data.insert(col_name.to_string(), ormer_value);
}
let ormer_row = Row::new(data);
let model = T::from_row(&ormer_row)?;
results.push(model);
}
Ok(results.into_iter().collect())
}
}
pub struct DeleteExecutor<'a, T: Model> {
filters: Vec<FilterExpr>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<T>,
}
impl<'a, T: Model> DeleteExecutor<'a, T> {
pub fn filter<F>(mut self, f: F) -> Self
where
F: FnOnce(T::Where) -> WhereExpr,
{
let where_obj = T::Where::default();
let expr = f(where_obj);
self.filters.push(expr.into());
self
}
pub async fn execute(self) -> Result<u64, crate::Error> {
let sql = self.build_sql();
let result = self
.client
.execute(&sql, &[])
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(result)
}
fn build_sql(&self) -> String {
let mut sql = format!("DELETE FROM {}", T::TABLE_NAME);
if !self.filters.is_empty() {
sql.push_str(" WHERE ");
let mut param_idx = 1;
for (i, filter) in self.filters.iter().enumerate() {
if i > 0 {
sql.push_str(" AND ");
}
common_helpers::format_filter(filter, &mut sql, &mut param_idx, DbType::PostgreSQL);
}
}
sql
}
}
impl<'a, T: Model + 'static> std::future::IntoFuture for DeleteExecutor<'a, T> {
type Output = Result<u64, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.execute().await })
}
}
pub struct UpdateExecutor<'a, T: Model> {
sets: Vec<(String, Value)>,
filters: Vec<FilterExpr>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<T>,
}
impl<'a, T: Model> UpdateExecutor<'a, T> {
pub fn filter<F>(mut self, f: F) -> Self
where
F: FnOnce(T::Where) -> WhereExpr,
{
let where_obj = T::Where::default();
let expr = f(where_obj);
self.filters.push(expr.into());
self
}
pub fn set<F, V, C>(mut self, field_fn: F, value: V) -> Self
where
F: FnOnce(T::Where) -> crate::query::builder::TypedColumn<C>,
V: Into<Value>,
{
let where_obj = T::Where::default();
let column = field_fn(where_obj);
let column_name = column.column_name().to_string();
self.sets.push((column_name, value.into()));
self
}
pub async fn execute(self) -> Result<u64, crate::Error> {
let (sql, params) = self.build_sql()?;
let pg_params = values_to_params(¶ms)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = pg_params
.iter()
.map(|p| p.as_ref() as &(dyn tokio_postgres::types::ToSql + Sync))
.collect();
let result = self
.client
.execute(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
Ok(result)
}
fn build_sql(&self) -> Result<(String, Vec<crate::model::Value>), crate::Error> {
let mut sql = format!("UPDATE {} SET ", T::TABLE_NAME);
let mut params = Vec::new();
let mut first = true;
for (col_name, value) in &self.sets {
if !first {
sql.push_str(", ");
}
sql.push_str(&format!("{col_name} = ${}", params.len() + 1));
params.push(value.clone());
first = false;
}
if !self.filters.is_empty() {
sql.push_str(" WHERE ");
let mut param_idx = params.len() + 1;
for (i, filter) in self.filters.iter().enumerate() {
if i > 0 {
sql.push_str(" AND ");
}
common_helpers::format_filter_with_params(
filter,
&mut sql,
&mut param_idx,
&mut params,
DbType::PostgreSQL,
);
}
}
Ok((sql, params))
}
}
impl<'a, T: Model + 'static> std::future::IntoFuture for UpdateExecutor<'a, T> {
type Output = Result<u64, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.execute().await })
}
}
fn values_to_params(
values: &[crate::model::Value],
) -> Result<Vec<Box<dyn tokio_postgres::types::ToSql + Sync>>, crate::Error> {
#[allow(unused_imports)]
use tokio_postgres::types::ToSql;
let mut params: Vec<Box<dyn tokio_postgres::types::ToSql + Sync>> = Vec::new();
for value in values {
let param: Box<dyn tokio_postgres::types::ToSql + Sync> = match value {
crate::model::Value::Integer(v) => Box::new(*v),
crate::model::Value::Text(v) => Box::new(v.clone()),
crate::model::Value::Real(v) => Box::new(*v),
crate::model::Value::Null => {
let null_val: Option<i64> = None;
Box::new(null_val)
}
};
params.push(param);
}
Ok(params)
}
pub struct RelatedSelectExecutor<'a, T: Model, R: Model> {
select: RelatedSelect<T, R>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<(T, R)>,
}
impl<'a, T: Model, R: Model> RelatedSelectExecutor<'a, T, R> {
pub fn filter<F>(self, f: F) -> Self
where
F: FnOnce(T::Where, R::Where) -> WhereExpr,
{
Self {
select: self.select.filter(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn range<RR: Into<crate::query::builder::RangeBounds>>(self, range: RR) -> Self {
Self {
select: self.select.range(range),
client: self.client,
_marker: PhantomData,
}
}
pub fn exec(self) -> RelatedCollectFuture<'a, T, R>
where
T: 'static,
R: 'static,
{
RelatedCollectFuture { executor: self }
}
pub async fn collect<C: FromIterator<T>>(self) -> Result<C, crate::Error> {
let results = self.collect_inner().await?;
Ok(results.into_iter().collect())
}
async fn collect_inner(self) -> Result<Vec<T>, crate::Error> {
let (sql, params) = self.select.to_sql_with_params(DbType::PostgreSQL);
let pg_params = values_to_params(¶ms)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = pg_params
.iter()
.map(|p| &**p as &(dyn tokio_postgres::types::ToSql + Sync))
.collect();
let rows = self
.client
.query(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut results = Vec::new();
for row in rows {
let mut data = HashMap::new();
for (i, col_name) in T::COLUMNS.iter().enumerate() {
let column_info = &T::COLUMN_SCHEMA[i];
let rust_type = column_info.rust_type;
let is_nullable = column_info.is_nullable;
let ormer_value = if is_nullable {
match rust_type {
"i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
let v: Option<i64> = row.get(i);
match v {
Some(val) => crate::model::Value::Integer(val),
None => crate::model::Value::Null,
}
}
"String" => {
let v: Option<String> = row.get(i);
match v {
Some(val) => crate::model::Value::Text(val),
None => crate::model::Value::Null,
}
}
"f32" | "f64" => {
let v: Option<f64> = row.get(i);
match v {
Some(val) => crate::model::Value::Real(val),
None => crate::model::Value::Null,
}
}
"bool" => {
let v: Option<bool> = row.get(i);
match v {
Some(true) => crate::model::Value::Integer(1),
Some(false) => crate::model::Value::Integer(0),
None => crate::model::Value::Null,
}
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported nullable column type: {rust_type}"
)));
}
}
} else {
match rust_type {
"i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
let v: i64 = row.get(i);
crate::model::Value::Integer(v)
}
"String" => {
let v: String = row.get(i);
crate::model::Value::Text(v)
}
"f32" | "f64" => {
let v: f64 = row.get(i);
crate::model::Value::Real(v)
}
"bool" => {
let v: bool = row.get(i);
if v {
crate::model::Value::Integer(1)
} else {
crate::model::Value::Integer(0)
}
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
}
};
data.insert(col_name.to_string(), ormer_value);
}
let ormer_row = Row::new(data);
let model = T::from_row(&ormer_row)?;
results.push(model);
}
Ok(results)
}
}
pub struct RelatedCollectFuture<'a, T: Model, R: Model> {
executor: RelatedSelectExecutor<'a, T, R>,
}
impl<'a, T: Model + 'static, R: Model + 'static> std::future::IntoFuture
for RelatedCollectFuture<'a, T, R>
{
type Output = Result<Vec<T>, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.executor.collect_inner().await })
}
}
pub struct MultiTableSelectExecutor<'a, T: Model, R1: Model, R2: Model> {
select: MultiTableSelect<T, R1, R2>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<(T, R1, R2)>,
}
impl<'a, T: Model, R1: Model, R2: Model> MultiTableSelectExecutor<'a, T, R1, R2> {
pub fn filter<F>(self, f: F) -> Self
where
F: FnOnce(T::Where, R1::Where, R2::Where) -> WhereExpr,
{
Self {
select: self.select.filter(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn range<RR: Into<crate::query::builder::RangeBounds>>(self, range: RR) -> Self {
Self {
select: self.select.range(range),
client: self.client,
_marker: PhantomData,
}
}
pub fn exec(self) -> MultiTableCollectFuture<'a, T, R1, R2>
where
T: 'static,
R1: 'static,
R2: 'static,
{
MultiTableCollectFuture { executor: self }
}
async fn collect_inner(self) -> Result<Vec<T>, crate::Error> {
let (sql, params) = self.select.to_sql_with_params(DbType::PostgreSQL);
let pg_params = values_to_params(¶ms)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = pg_params
.iter()
.map(|p| &**p as &(dyn tokio_postgres::types::ToSql + Sync))
.collect();
let rows = self
.client
.query(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut results = Vec::new();
for row in rows {
let mut data = HashMap::new();
for (i, col_name) in T::COLUMNS.iter().enumerate() {
let column_info = &T::COLUMN_SCHEMA[i];
let rust_type = column_info.rust_type;
let is_nullable = column_info.is_nullable;
let ormer_value = if is_nullable {
match rust_type {
"i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
let v: Option<i64> = row.get(i);
match v {
Some(val) => crate::model::Value::Integer(val),
None => crate::model::Value::Null,
}
}
"String" => {
let v: Option<String> = row.get(i);
match v {
Some(val) => crate::model::Value::Text(val),
None => crate::model::Value::Null,
}
}
"f32" | "f64" => {
let v: Option<f64> = row.get(i);
match v {
Some(val) => crate::model::Value::Real(val),
None => crate::model::Value::Null,
}
}
"bool" => {
let v: Option<bool> = row.get(i);
match v {
Some(true) => crate::model::Value::Integer(1),
Some(false) => crate::model::Value::Integer(0),
None => crate::model::Value::Null,
}
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported nullable column type: {rust_type}"
)));
}
}
} else {
match rust_type {
"i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
let v: i64 = row.get(i);
crate::model::Value::Integer(v)
}
"String" => {
let v: String = row.get(i);
crate::model::Value::Text(v)
}
"f32" | "f64" => {
let v: f64 = row.get(i);
crate::model::Value::Real(v)
}
"bool" => {
let v: bool = row.get(i);
if v {
crate::model::Value::Integer(1)
} else {
crate::model::Value::Integer(0)
}
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
}
};
data.insert(col_name.to_string(), ormer_value);
}
let ormer_row = Row::new(data);
let model = T::from_row(&ormer_row)?;
results.push(model);
}
Ok(results)
}
}
pub struct MultiTableCollectFuture<'a, T: Model, R1: Model, R2: Model> {
executor: MultiTableSelectExecutor<'a, T, R1, R2>,
}
impl<'a, T: Model + 'static, R1: Model + 'static, R2: Model + 'static> std::future::IntoFuture
for MultiTableCollectFuture<'a, T, R1, R2>
{
type Output = Result<Vec<T>, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.executor.collect_inner().await })
}
}
pub struct FourTableSelectExecutor<'a, T: Model, R1: Model, R2: Model, R3: Model> {
select: FourTableSelect<T, R1, R2, R3>,
client: &'a tokio_postgres::Client,
_marker: PhantomData<(T, R1, R2, R3)>,
}
impl<'a, T: Model, R1: Model, R2: Model, R3: Model> FourTableSelectExecutor<'a, T, R1, R2, R3> {
pub fn filter<F>(self, f: F) -> Self
where
F: FnOnce(T::Where, R1::Where, R2::Where, R3::Where) -> WhereExpr,
{
Self {
select: self.select.filter(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn range<RR: Into<crate::query::builder::RangeBounds>>(self, range: RR) -> Self {
Self {
select: self.select.range(range),
client: self.client,
_marker: PhantomData,
}
}
pub fn exec(self) -> FourTableCollectFuture<'a, T, R1, R2, R3>
where
T: 'static,
R1: 'static,
R2: 'static,
R3: 'static,
{
FourTableCollectFuture { executor: self }
}
async fn collect_inner(self) -> Result<Vec<T>, crate::Error> {
let (sql, params) = self.select.to_sql_with_params(DbType::PostgreSQL);
let pg_params = values_to_params(¶ms)?;
let param_refs: Vec<&(dyn tokio_postgres::types::ToSql + Sync)> = pg_params
.iter()
.map(|p| &**p as &(dyn tokio_postgres::types::ToSql + Sync))
.collect();
let rows = self
.client
.query(&sql, ¶m_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut results = Vec::new();
for row in rows {
let mut data = HashMap::new();
for (i, col_name) in T::COLUMNS.iter().enumerate() {
let column_info = &T::COLUMN_SCHEMA[i];
let rust_type = column_info.rust_type;
let is_nullable = column_info.is_nullable;
let ormer_value = if is_nullable {
match rust_type {
"i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
let v: Option<i64> = row.get(i);
match v {
Some(val) => crate::model::Value::Integer(val),
None => crate::model::Value::Null,
}
}
"String" => {
let v: Option<String> = row.get(i);
match v {
Some(val) => crate::model::Value::Text(val),
None => crate::model::Value::Null,
}
}
"f32" | "f64" => {
let v: Option<f64> = row.get(i);
match v {
Some(val) => crate::model::Value::Real(val),
None => crate::model::Value::Null,
}
}
"bool" => {
let v: Option<bool> = row.get(i);
match v {
Some(true) => crate::model::Value::Integer(1),
Some(false) => crate::model::Value::Integer(0),
None => crate::model::Value::Null,
}
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported nullable column type: {rust_type}"
)));
}
}
} else {
match rust_type {
"i8" | "i16" | "i32" | "i64" | "u8" | "u16" | "u32" | "u64" => {
let v: i64 = row.get(i);
crate::model::Value::Integer(v)
}
"String" => {
let v: String = row.get(i);
crate::model::Value::Text(v)
}
"f32" | "f64" => {
let v: f64 = row.get(i);
crate::model::Value::Real(v)
}
"bool" => {
let v: bool = row.get(i);
if v {
crate::model::Value::Integer(1)
} else {
crate::model::Value::Integer(0)
}
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
}
};
data.insert(col_name.to_string(), ormer_value);
}
let ormer_row = Row::new(data);
let model = T::from_row(&ormer_row)?;
results.push(model);
}
Ok(results)
}
}
pub struct FourTableCollectFuture<'a, T: Model, R1: Model, R2: Model, R3: Model> {
executor: FourTableSelectExecutor<'a, T, R1, R2, R3>,
}
impl<'a, T: Model + 'static, R1: Model + 'static, R2: Model + 'static, R3: Model + 'static>
std::future::IntoFuture for FourTableCollectFuture<'a, T, R1, R2, R3>
{
type Output = Result<Vec<T>, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.executor.collect_inner().await })
}
}
impl<'a, T: Model, J: Model> LeftJoinedSelectExecutor<'a, T, J> {
pub fn filter<F>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> WhereExpr,
{
Self {
select: self.select.filter(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn range<RR: Into<crate::query::builder::RangeBounds>>(self, range: RR) -> Self {
Self {
select: self.select.range(range),
client: self.client,
_marker: PhantomData,
}
}
pub fn collect<C: FromIterator<(T, Option<J>)> + 'static>(
self,
) -> LeftJoinCollectFuture<'a, T, J> {
LeftJoinCollectFuture {
executor: self,
_marker: PhantomData,
}
}
pub fn execute(self) -> LeftJoinCollectFuture<'a, T, J>
where
T: 'static,
J: 'static,
{
self.collect::<Vec<(T, Option<J>)>>()
}
}
pub struct LeftJoinCollectFuture<'a, T: Model, J: Model> {
executor: LeftJoinedSelectExecutor<'a, T, J>,
_marker: PhantomData<(T, J)>,
}
impl<'a, T: Model + 'static, J: Model + 'static> std::future::IntoFuture
for LeftJoinCollectFuture<'a, T, J>
{
type Output = Result<Vec<(T, Option<J>)>, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.executor.collect_inner().await })
}
}
impl<'a, T: Model, J: Model> LeftJoinedSelectExecutor<'a, T, J> {
async fn collect_inner<C: FromIterator<(T, Option<J>)>>(self) -> Result<C, crate::Error> {
let (sql, params) = self.select.to_sql_with_params(DbType::PostgreSQL);
let pg_params: Vec<Box<dyn postgres_types::ToSql + Sync>> = params
.into_iter()
.map(|v| match v {
crate::model::Value::Integer(i) => {
Box::new(i) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Text(t) => {
Box::new(t) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Real(r) => {
Box::new(r) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Null => {
Box::new(None::<i32>) as Box<dyn postgres_types::ToSql + Sync>
}
})
.collect();
let pg_params_refs: Vec<&(dyn postgres_types::ToSql + Sync)> =
pg_params.iter().map(|p| p.as_ref()).collect();
let rows = self
.client
.query(&sql, &pg_params_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut results = Vec::new();
let t_col_count = T::COLUMNS.len();
for row in rows {
let mut t_data = HashMap::new();
for (i, col_name) in T::COLUMNS.iter().enumerate() {
let rust_type = T::COLUMN_SCHEMA[i].rust_type;
let ormer_value = match rust_type {
"i32" => {
let v: Option<i32> = row.try_get(i).ok().flatten();
crate::model::Value::Integer(v.unwrap_or(0) as i64)
}
"i64" => {
let v: Option<i64> = row.try_get(i).ok().flatten();
crate::model::Value::Integer(v.unwrap_or(0))
}
"String" => {
let v: Option<String> = row.try_get(i).ok().flatten();
crate::model::Value::Text(v.unwrap_or_default())
}
"f32" | "f64" => {
let v: Option<f64> = row.try_get(i).ok().flatten();
crate::model::Value::Real(v.unwrap_or(0.0))
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
};
t_data.insert(col_name.to_string(), ormer_value);
}
let t_model = T::from_row(&Row::new(t_data))?;
let mut j_data = HashMap::new();
let mut j_is_null = true;
for (i, col_name) in J::COLUMNS.iter().enumerate() {
let idx = t_col_count + i;
let rust_type = J::COLUMN_SCHEMA[i].rust_type;
let ormer_value = match rust_type {
"i32" => {
let v: Option<i32> = row.try_get(idx).ok().flatten();
if v.is_some() {
j_is_null = false;
}
crate::model::Value::Integer(v.unwrap_or(0) as i64)
}
"i64" => {
let v: Option<i64> = row.try_get(idx).ok().flatten();
if v.is_some() {
j_is_null = false;
}
crate::model::Value::Integer(v.unwrap_or(0))
}
"String" => {
let v: Option<String> = row.try_get(idx).ok().flatten();
if v.is_some() {
j_is_null = false;
}
crate::model::Value::Text(v.unwrap_or_default())
}
"f32" | "f64" => {
let v: Option<f64> = row.try_get(idx).ok().flatten();
if v.is_some() {
j_is_null = false;
}
crate::model::Value::Real(v.unwrap_or(0.0))
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
};
j_data.insert(col_name.to_string(), ormer_value);
}
if j_is_null {
results.push((t_model, None));
} else {
let j_model = J::from_row(&Row::new(j_data))?;
results.push((t_model, Some(j_model)));
}
}
Ok(results.into_iter().collect())
}
}
impl<'a, T: Model, J: Model> InnerJoinedSelectExecutor<'a, T, J> {
pub fn filter<F>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> WhereExpr,
{
Self {
select: self.select.filter(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn range<RR: Into<crate::query::builder::RangeBounds>>(self, range: RR) -> Self {
Self {
select: self.select.range(range),
client: self.client,
_marker: PhantomData,
}
}
pub fn collect<C: FromIterator<(T, J)> + 'static>(self) -> InnerJoinCollectFuture<'a, T, J> {
InnerJoinCollectFuture {
executor: self,
_marker: PhantomData,
}
}
pub fn execute(self) -> InnerJoinCollectFuture<'a, T, J>
where
T: 'static,
J: 'static,
{
self.collect::<Vec<(T, J)>>()
}
}
pub struct InnerJoinCollectFuture<'a, T: Model, J: Model> {
executor: InnerJoinedSelectExecutor<'a, T, J>,
_marker: PhantomData<(T, J)>,
}
impl<'a, T: Model + 'static, J: Model + 'static> std::future::IntoFuture
for InnerJoinCollectFuture<'a, T, J>
{
type Output = Result<Vec<(T, J)>, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.executor.collect_inner().await })
}
}
impl<'a, T: Model, J: Model> InnerJoinedSelectExecutor<'a, T, J> {
async fn collect_inner<C: FromIterator<(T, J)>>(self) -> Result<C, crate::Error> {
let (sql, params) = self.select.to_sql_with_params(DbType::PostgreSQL);
let pg_params: Vec<Box<dyn postgres_types::ToSql + Sync>> = params
.into_iter()
.map(|v| match v {
crate::model::Value::Integer(i) => {
Box::new(i) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Text(t) => {
Box::new(t) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Real(r) => {
Box::new(r) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Null => {
Box::new(None::<i32>) as Box<dyn postgres_types::ToSql + Sync>
}
})
.collect();
let pg_params_refs: Vec<&(dyn postgres_types::ToSql + Sync)> =
pg_params.iter().map(|p| p.as_ref()).collect();
let rows = self
.client
.query(&sql, &pg_params_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut results = Vec::new();
let t_col_count = T::COLUMNS.len();
for row in rows {
let mut t_data = HashMap::new();
for (i, col_name) in T::COLUMNS.iter().enumerate() {
let rust_type = T::COLUMN_SCHEMA[i].rust_type;
let ormer_value = match rust_type {
"i32" => {
let v: i32 = row.get(i);
crate::model::Value::Integer(v as i64)
}
"i64" => {
let v: i64 = row.get(i);
crate::model::Value::Integer(v)
}
"String" => {
let v: String = row.get(i);
crate::model::Value::Text(v)
}
"f32" | "f64" => {
let v: f64 = row.get(i);
crate::model::Value::Real(v)
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
};
t_data.insert(col_name.to_string(), ormer_value);
}
let t_model = T::from_row(&Row::new(t_data))?;
let mut j_data = HashMap::new();
for (i, col_name) in J::COLUMNS.iter().enumerate() {
let idx = t_col_count + i;
let rust_type = J::COLUMN_SCHEMA[i].rust_type;
let ormer_value = match rust_type {
"i32" => {
let v: i32 = row.get(idx);
crate::model::Value::Integer(v as i64)
}
"i64" => {
let v: i64 = row.get(idx);
crate::model::Value::Integer(v)
}
"String" => {
let v: String = row.get(idx);
crate::model::Value::Text(v)
}
"f32" | "f64" => {
let v: f64 = row.get(idx);
crate::model::Value::Real(v)
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
};
j_data.insert(col_name.to_string(), ormer_value);
}
let j_model = J::from_row(&Row::new(j_data))?;
results.push((t_model, j_model));
}
Ok(results.into_iter().collect())
}
}
impl<'a, T: Model, J: Model> RightJoinedSelectExecutor<'a, T, J> {
pub fn filter<F>(self, f: F) -> Self
where
F: FnOnce(T::Where) -> WhereExpr,
{
Self {
select: self.select.filter(f),
client: self.client,
_marker: PhantomData,
}
}
pub fn range<RR: Into<crate::query::builder::RangeBounds>>(self, range: RR) -> Self {
Self {
select: self.select.range(range),
client: self.client,
_marker: PhantomData,
}
}
pub fn collect<C: FromIterator<(Option<T>, J)> + 'static>(
self,
) -> RightJoinCollectFuture<'a, T, J> {
RightJoinCollectFuture {
executor: self,
_marker: PhantomData,
}
}
pub fn execute(self) -> RightJoinCollectFuture<'a, T, J>
where
T: 'static,
J: 'static,
{
self.collect::<Vec<(Option<T>, J)>>()
}
}
pub struct RightJoinCollectFuture<'a, T: Model, J: Model> {
executor: RightJoinedSelectExecutor<'a, T, J>,
_marker: PhantomData<(T, J)>,
}
impl<'a, T: Model + 'static, J: Model + 'static> std::future::IntoFuture
for RightJoinCollectFuture<'a, T, J>
{
type Output = Result<Vec<(Option<T>, J)>, crate::Error>;
type IntoFuture = std::pin::Pin<Box<dyn std::future::Future<Output = Self::Output> + 'a>>;
fn into_future(self) -> Self::IntoFuture {
Box::pin(async move { self.executor.collect_inner().await })
}
}
impl<'a, T: Model, J: Model> RightJoinedSelectExecutor<'a, T, J> {
async fn collect_inner<C: FromIterator<(Option<T>, J)>>(self) -> Result<C, crate::Error> {
let (sql, params) = self.select.to_sql_with_params(DbType::PostgreSQL);
let pg_params: Vec<Box<dyn postgres_types::ToSql + Sync>> = params
.into_iter()
.map(|v| match v {
crate::model::Value::Integer(i) => {
Box::new(i) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Text(t) => {
Box::new(t) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Real(r) => {
Box::new(r) as Box<dyn postgres_types::ToSql + Sync>
}
crate::model::Value::Null => {
Box::new(None::<i32>) as Box<dyn postgres_types::ToSql + Sync>
}
})
.collect();
let pg_params_refs: Vec<&(dyn postgres_types::ToSql + Sync)> =
pg_params.iter().map(|p| p.as_ref()).collect();
let rows = self
.client
.query(&sql, &pg_params_refs)
.await
.map_err(|e| crate::Error::Database(e.to_string()))?;
let mut results = Vec::new();
let t_col_count = T::COLUMNS.len();
for row in rows {
let mut t_data = HashMap::new();
let mut t_is_null = true;
for (i, col_name) in T::COLUMNS.iter().enumerate() {
let rust_type = T::COLUMN_SCHEMA[i].rust_type;
let ormer_value = match rust_type {
"i32" => {
let v: Option<i32> = row.try_get(i).ok().flatten();
if v.is_some() {
t_is_null = false;
}
crate::model::Value::Integer(v.unwrap_or(0) as i64)
}
"i64" => {
let v: Option<i64> = row.try_get(i).ok().flatten();
if v.is_some() {
t_is_null = false;
}
crate::model::Value::Integer(v.unwrap_or(0))
}
"String" => {
let v: Option<String> = row.try_get(i).ok().flatten();
if v.is_some() {
t_is_null = false;
}
crate::model::Value::Text(v.unwrap_or_default())
}
"f32" | "f64" => {
let v: Option<f64> = row.try_get(i).ok().flatten();
if v.is_some() {
t_is_null = false;
}
crate::model::Value::Real(v.unwrap_or(0.0))
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
};
t_data.insert(col_name.to_string(), ormer_value);
}
let t_model = if t_is_null {
None
} else {
Some(T::from_row(&Row::new(t_data))?)
};
let mut j_data = HashMap::new();
for (i, col_name) in J::COLUMNS.iter().enumerate() {
let idx = t_col_count + i;
let rust_type = J::COLUMN_SCHEMA[i].rust_type;
let ormer_value = match rust_type {
"i32" => {
let v: i32 = row.get(idx);
crate::model::Value::Integer(v as i64)
}
"i64" => {
let v: i64 = row.get(idx);
crate::model::Value::Integer(v)
}
"String" => {
let v: String = row.get(idx);
crate::model::Value::Text(v)
}
"f32" | "f64" => {
let v: f64 = row.get(idx);
crate::model::Value::Real(v)
}
_ => {
return Err(crate::Error::Database(format!(
"Unsupported column type: {rust_type}"
)));
}
};
j_data.insert(col_name.to_string(), ormer_value);
}
let j_model = J::from_row(&Row::new(j_data))?;
results.push((t_model, j_model));
}
Ok(results.into_iter().collect())
}
}
fn convert_postgres_value(
row: &tokio_postgres::Row,
index: usize,
) -> Result<crate::model::Value, crate::Error> {
if let Ok(v) = row.try_get::<_, Option<i64>>(index) {
return Ok(match v {
Some(val) => crate::model::Value::Integer(val),
None => crate::model::Value::Null,
});
}
if let Ok(v) = row.try_get::<_, Option<String>>(index) {
return Ok(match v {
Some(val) => crate::model::Value::Text(val),
None => crate::model::Value::Null,
});
}
if let Ok(v) = row.try_get::<_, Option<f64>>(index) {
return Ok(match v {
Some(val) => crate::model::Value::Real(val),
None => crate::model::Value::Null,
});
}
if let Ok(v) = row.try_get::<_, Option<bool>>(index) {
return Ok(match v {
Some(true) => crate::model::Value::Integer(1),
Some(false) => crate::model::Value::Integer(0),
None => crate::model::Value::Null,
});
}
Err(crate::Error::Database(format!(
"Unsupported column type at index {index}"
)))
}