use std::io;
use sqlx::{ColumnIndex, Database, Decode, Encode, Executor, IntoArguments};
use crate::sql::{
dialects::{
condition::Where,
context,
page::{Order, PageRequest},
schema::schema::Schema,
},
schema::{column::Column, index::Index, table::TableSchema},
};
pub struct MySchema {
pub ctx: context::Context,
}
impl MySchema {
pub fn new(ctx: context::Context) -> impl Schema {
Self { ctx }
}
async fn check_upgrade_table<'c, E, DB: Database>(&self, conn: &mut E) -> io::Result<()>
where
for<'e> &'e mut E: Executor<'e, Database = DB>,
for<'a> DB::Arguments<'a>: IntoArguments<'a, DB>,
for<'a> &'a str: ColumnIndex<DB::Row>,
for<'a> std::string::String: Encode<'a, DB> + sqlx::Type<DB>,
for<'a> bool: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
{
Ok(())
}
}
impl Schema for MySchema {
fn sql_create_table(&self, table: &TableSchema) -> std::io::Result<Vec<String>> {
Ok(vec![])
}
fn sql_table_rename(&self, table: &TableSchema, new_table_name: &String) -> String {
format!(
"ALTER TABLE {} RENAME TO {}",
self.ctx.quote(&table.name_with_schema()),
self.ctx.quote(&new_table_name)
)
}
fn sql_drop_table(&self, table: &TableSchema) -> String {
"".to_string()
}
fn sql_drop_index(&self, table: &TableSchema, index: &Index) -> String {
"".to_string()
}
fn sql_create_index(&self, table: &TableSchema, index: &Index) -> Option<String> {
Some("".to_string())
}
fn sql_create_indexes(&self, table: &TableSchema) -> Vec<String> {
vec![]
}
async fn get_tables<C, DB: Database>(&self, conn: &mut C) -> std::io::Result<Vec<TableSchema>>
where
for<'a> &'a mut C: Executor<'a, Database = DB>,
for<'a> <DB as Database>::Arguments<'a>: IntoArguments<'a, DB>,
for<'a> &'a str: ColumnIndex<DB::Row>,
for<'a> bool: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
for<'a> i32: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
for<'a> i64: sqlx::Decode<'a, DB> + sqlx::Type<DB> + Encode<'a, DB>,
for<'a> std::string::String: Decode<'a, DB> + Encode<'a, DB> + sqlx::Type<DB>,
{
Ok(vec![])
}
fn sql_add_column(&self, table: &TableSchema, col: &Column) -> String {
"".to_string()
}
fn is_table_name_equal(&self, table1: &TableSchema, table2: &TableSchema) -> bool {
self.ctx
.is_table_name_equal(&table1.name_with_schema(), &table2.name_with_schema())
}
fn sql_alter_column(
&self,
table: &TableSchema,
old: &Column,
new: &Column,
) -> io::Result<Vec<String>> {
Ok(vec![])
}
fn sql_drop_column(&self, table: &TableSchema, col: &Column) -> String {
"".to_string()
}
fn table_name_with_schema(&self, table: &TableSchema) -> String {
self.ctx.table_name_with_schema(&table.name_with_schema())
}
async fn query_upgrade_tags<C, DB: Database>(
&self,
conn: &mut C,
table_name: &String,
tag: &String,
) -> io::Result<Vec<String>>
where
for<'a> &'a mut C: Executor<'a, Database = DB>,
for<'a> DB::Arguments<'a>: IntoArguments<'a, DB>,
for<'a> &'a str: ColumnIndex<DB::Row>,
for<'a> bool: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
for<'a> i32: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
for<'a> i64: sqlx::Decode<'a, DB> + sqlx::Type<DB> + Encode<'a, DB>,
for<'a> std::string::String: Decode<'a, DB> + Encode<'a, DB> + sqlx::Type<DB>,
{
Ok(vec![])
}
async fn insert_upgrade_tag<C, DB: Database>(
&self,
conn: &mut C,
table_name: &String,
tag: &String,
tag_value: &String,
) -> io::Result<()>
where
for<'a> &'a mut C: Executor<'a, Database = DB>,
for<'a> <DB as Database>::Arguments<'a>: IntoArguments<'a, DB>,
for<'a> &'a str: ColumnIndex<DB::Row>,
for<'a> bool: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
for<'a> i32: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
for<'a> i64: sqlx::Decode<'a, DB> + sqlx::Type<DB> + Encode<'a, DB>,
for<'a> std::string::String: Decode<'a, DB> + Encode<'a, DB> + sqlx::Type<DB>,
{
self.check_upgrade_table(&mut *conn).await?;
Ok(())
}
async fn execute_sql<'s, C, DB: Database>(
&self,
conn: &mut C,
sql: &'s str,
) -> io::Result<<DB as Database>::QueryResult>
where
for<'a> &'a mut C: Executor<'a, Database = DB>,
for<'a> DB::Arguments<'a>: IntoArguments<'a, DB>,
for<'a> &'a str: ColumnIndex<DB::Row>,
for<'a> bool: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
for<'a> i32: sqlx::Decode<'a, DB> + sqlx::Type<DB>,
for<'a> i64: sqlx::Decode<'a, DB> + sqlx::Type<DB> + Encode<'a, DB>,
for<'a> std::string::String: Decode<'a, DB> + Encode<'a, DB> + sqlx::Type<DB>,
{
sqlx::query(sql).execute(&mut *conn).await.map_err(|err| {
tracing::error!("Execute sql {sql} error: {err}");
io::Error::new(
io::ErrorKind::Other,
format!("Execute sql {sql} error: {err}"),
)
})
}
fn sql_insert(&self, table: &TableSchema, return_sql: bool) -> String {
self.sql_insert_columns(
table,
&table
.columns
.iter()
.map(|c| c.get_column_name())
.collect::<Vec<String>>(),
return_sql,
)
}
fn sql_insert_columns(
&self,
table: &TableSchema,
cols: &Vec<String>,
return_sql: bool,
) -> String {
let mut column_names = "".to_string();
let mut column_value_holder = "".to_string();
for (n, col) in cols.iter().enumerate() {
if n > 0 {
column_names.push_str(",");
column_value_holder.push_str(",");
}
column_names.push_str(self.ctx.quote(&col).as_str());
column_value_holder.push_str(format!("${}", n + 1).as_str());
}
let return_cols = "".to_string();
format!(
"insert into {} ({column_names}) values ({column_value_holder}) {return_cols}",
self.ctx.quote(&self.table_name_with_schema(table))
)
}
fn sql_update_columns(
&self,
table: &TableSchema,
cols: &Vec<String>,
wh: Option<Where>,
return_sql: bool,
) -> String {
let mut columns = "".to_string();
for (n, col) in cols.iter().enumerate() {
if n > 0 {
columns.push_str(",");
}
columns.push_str(self.ctx.quote(&col).as_str());
columns.push_str("=");
columns.push_str(format!("${}", n + 1).as_str());
}
let mut where_str = String::from("");
if let Some(w) = wh {
let (ws, _) = w.sql(cols.len() + 1, &self.quoter());
if !ws.is_empty() {
where_str.push_str(" where ");
where_str.push_str(&ws);
}
}
let return_cols = "".to_string();
format!(
"update {} set {columns} {where_str} {return_cols}",
self.ctx.quote(&self.table_name_with_schema(table))
)
}
fn quoter(&self) -> crate::sql::utils::quote::Quoter {
self.ctx.quoter.clone()
}
fn sql_delete(&self, table: &TableSchema, wh: Option<Where>, return_sql: bool) -> String {
let mut where_str = String::from("");
if let Some(w) = wh {
let (ws, _) = w.sql(1, &self.quoter());
if !ws.is_empty() {
where_str.push_str(" where ");
where_str.push_str(&ws);
}
}
let return_cols = "".to_string();
format!(
"delete from {} {where_str} {return_cols}",
self.ctx.quote(&self.table_name_with_schema(table))
)
}
fn sql_count(&self, table: &TableSchema, wh: Option<Where>) -> String {
let mut where_str = String::from("");
if let Some(w) = wh {
let (ws, _) = w.sql(1, &self.quoter());
if !ws.is_empty() {
where_str.push_str(" where ");
where_str.push_str(&ws);
}
}
format!(
r#"select count(0) as "count" from {} {where_str}"#,
self.ctx.quote(&self.table_name_with_schema(table))
)
}
fn sql_select(
&self,
table: &TableSchema,
wh: Option<Where>,
orders: &Vec<Order>,
pg: Option<&PageRequest>,
) -> String {
self.sql_select_columns(table, &table.columns, wh, orders, pg)
}
fn sql_select_columns(
&self,
table: &TableSchema,
columns: &Vec<Column>,
wh: Option<Where>,
orders: &Vec<Order>,
pg: Option<&PageRequest>,
) -> String {
let cols: Vec<String> = columns
.iter()
.map(|c| c.get_query_column_name(&self.quoter()))
.collect();
let mut where_str = String::from("");
if let Some(w) = wh {
let (ws, _) = w.sql(1, &self.quoter());
if !ws.is_empty() {
where_str.push_str(" where ");
where_str.push_str(&ws);
}
}
let mut order_str = String::from("");
if !orders.is_empty() {
order_str.push_str(" order by ");
let items: Vec<String> = orders
.iter()
.map(|o| format!("{} {}", self.ctx.quote(&o.field), o.order_type.sql()))
.collect();
order_str.push_str(items.join(", ").as_str());
}
let mut page_str = String::from("");
if let Some(page) = pg {
page_str.push_str(
format!(
"offset {} limit {}",
(page.get_page_no() - 1) * page.get_page_size(),
page.get_page_size()
)
.as_str(),
);
}
format!(
"select {} from {} {where_str} {order_str} {page_str}",
cols.join(","),
self.ctx.quote(&self.table_name_with_schema(table))
)
}
}