djkato/CSQL
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

db pushing & transaction window [broken]

Browse source
This commit is contained in:
djkato 2023-04-09 14:07:33 +02:00
parent bcda74ff35
commit 471b3e7383
7 changed files with 461 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

128
src/backend/backend_manager.rs
View file

@ -1,11 +1,12 @@
use super::csv_handler::{DataEntry, ImportedData};
use super::database_handler::{DBLoginData, TableField, Tables};
use super::database_handler::{DBLoginData, QueryResult, Table, TableField, Tables};
use super::parser::parse;
use sqlx::mysql::MySqlQueryResult;
use sqlx::MySqlConnection;
use std::error::Error;
use std::sync::Arc;
use tokio::sync::mpsc::Receiver;
use tokio::sync::{oneshot, Mutex};
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::sync::{oneshot, Mutex, MutexGuard};
pub struct BackendManger {
pub db_login_data: DBLoginData,
pub imported_data: ImportedData,
@ -81,24 +82,20 @@ impl BackendManger {
.await;
if csv_data.are_headers {
match try_match_headers_to_fields(
&mut db_table_data
.tables
.get_mut(table_index)
.unwrap()
.fields
.as_mut()
.unwrap()
.as_mut(),
&mut db_table_data,
table_index,
&csv_data.data.iter_row(0),
) {
Ok(_) => {
for field in db_table_data
for (col_index, field) in db_table_data
.tables
.get_mut(table_index)
.unwrap()
.fields
.as_mut()
.unwrap()
.iter_mut()
.enumerate()
{
println!(
" > automapping field \"{}\" to col \"{:?}\"",
@ -111,6 +108,13 @@ impl BackendManger {
parse(cell);
}
}
if is_whole_col_parsed(&mut csv_data, col_index) {
println!("col \"{}\" is parsed whole!", &col_index);
csv_data.parsed_cols.push(col_index);
}
if is_whole_table_parsed(&csv_data) {
csv_data.is_parsed = true;
}
}
sender.send(true).unwrap_or_else(|e| {
println!(
@ -136,16 +140,107 @@ impl BackendManger {
parse(cell);
}
}
if is_whole_col_parsed(&mut csv_data, col_index) {
println!("col \"{}\" is parsed whole!", &col_index);
csv_data.parsed_cols.push(col_index - 1);
}
if is_whole_table_parsed(&csv_data) {
csv_data.is_parsed = true;
}
}
Communication::StartInserting(table_index, sender, oneshot_sender) => {
let csv_data = self.csv_data.lock().await;
let db_table_data = self.db_table_data.lock().await;
let table = db_table_data.tables.get(table_index).unwrap();
let trans_start =
Table::start_transaction(self.db_connection.as_mut().unwrap()).await;
sender
.send(trans_start)
.await
.unwrap_or_else(|_| println!("failed to send Transaction Start"));
let start_i: usize = csv_data.are_headers.into();
for i in start_i..csv_data.data.rows() {
let row: Vec<&DataEntry> = csv_data.data[i].iter().collect();
let res = table
.insert_into_table(self.db_connection.as_mut().unwrap(), row)
.await;
println!(
" | Query: {}\n > Result: {:?}",
res.query, res.result
);
sender
.send(res)
.await
.unwrap_or_else(|_| println!("failed to send insert into table"));
}
oneshot_sender
.send(true)
.unwrap_or_else(|_| println!("Failed to send end of insertin transaction"));
}
Communication::TryCommit(sender) => {
sender
.send(Table::transaction_commit(self.db_connection.as_mut().unwrap()).await)
.unwrap_or_else(|_| println!("Failed to respond to TryCommit"));
}
Communication::TryRollBack(sender) => {
sender
.send(
Table::transaction_rollback(self.db_connection.as_mut().unwrap()).await,
)
.unwrap_or_else(|_| println!("Failed to respond to TryRollBack"));
}
}
}
}
}
pub fn is_whole_table_parsed(csv_data: &MutexGuard<ImportedData>) -> bool {
if csv_data.data.cols() == csv_data.parsed_cols.len() {
true
} else {
false
}
}
pub fn is_whole_col_parsed(csv_data: &mut MutexGuard<ImportedData>, col_index: usize) -> bool {
let mut csv_iter = csv_data.data.iter_col(col_index);
if csv_data.are_headers {
csv_iter.next();
}
csv_iter.all(|cel| {
if let Some(parse) = &cel.is_parsed {
match parse {
Ok(_) => return true,
Err(_) => {
println!(
"Cel \"{}\" in Col \"{}\" isnt parsed :(",
cel.data, col_index
);
return false;
}
}
} else {
println!(
"Cel \"{}\" in Col \"{}\" isnt parsed :(",
cel.data, col_index
);
false
}
})
}
pub fn try_match_headers_to_fields(
db_fields: &mut Vec<TableField>,
tables: &mut MutexGuard<Tables>,
table_index: usize,
csv_headers: &std::slice::Iter<'_, DataEntry>,
) -> Result<(), ()> {
let db_fields = tables
.tables
.get_mut(table_index)
.unwrap()
.fields
.as_mut()
.unwrap();
let mut has_matched_some = false;
for field in db_fields {
for (i, header) in csv_headers.clone().enumerate() {
@ -170,4 +265,7 @@ pub enum Communication {
LoadImportFilePath(String, oneshot::Sender<bool>),
GetTableDescription(usize, oneshot::Sender<bool>),
TryParseCol(usize),
StartInserting(usize, Sender<QueryResult>, oneshot::Sender<bool>),
TryCommit(oneshot::Sender<QueryResult>),
TryRollBack(oneshot::Sender<QueryResult>),
}

5
src/backend/csv_handler.rs
View file

@ -7,6 +7,8 @@ pub struct ImportedData {
pub data: grid::Grid<DataEntry>,
pub path: String,
pub are_headers: bool,
pub parsed_cols: Vec<usize>,
pub is_parsed: bool,
}
#[derive(Clone, Default)]
pub struct DataEntry {
@ -19,7 +21,6 @@ impl ImportedData {
match csv::Reader::from_path(self.path.as_str()) {
Ok(mut rdr) => {
/* If there are headers, add to first row, else first row is empty */
if let Ok(headers) = rdr.headers() {
self.data = grid::Grid::new(0, 0);
let headers_vec: Vec<String> = headers.iter().map(|x| x.to_owned()).collect();
@ -65,6 +66,8 @@ impl Default for ImportedData {
data: grid::Grid::new(0, 0),
path: String::new(),
are_headers: false,
parsed_cols: vec![],
is_parsed: false,
}
}
}

100
src/backend/database_handler.rs
View file

@ -1,8 +1,11 @@
use crate::backend::csv_handler::DataEntry;
use core::num::ParseIntError;
use eframe::glow::Query;
use sqlx::mysql::MySqlQueryResult;
use sqlx::{mysql::MySqlConnectOptions, ConnectOptions};
use sqlx::{FromRow, MySqlConnection};
use std::error::Error;
use std::slice::Iter;
#[derive(Default)]
pub struct Tables {
pub tables: Vec<Table>,
@ -55,6 +58,95 @@ impl Tables {
}
impl Table {
pub async fn transaction_commit(connection: &mut MySqlConnection) -> QueryResult {
match sqlx::query("COMMIT").execute(connection).await {
Ok(res) => {
return QueryResult {
query: "ROLLBACK".to_owned(),
result: Ok(res),
}
}
Err(e) => {
return QueryResult {
query: "ROLLBACK".to_owned(),
result: Err(Box::new(e)),
}
}
}
}
pub async fn transaction_rollback(connection: &mut MySqlConnection) -> QueryResult {
match sqlx::query("ROLLBACK").execute(connection).await {
Ok(res) => {
return QueryResult {
query: "ROLLBACK".to_owned(),
result: Ok(res),
}
}
Err(e) => {
return QueryResult {
query: "ROLLBACK".to_owned(),
result: Err(Box::new(e)),
}
}
}
}
pub async fn start_transaction(connection: &mut MySqlConnection) -> QueryResult {
match sqlx::query("BEGIN").execute(connection).await {
Ok(res) => {
return QueryResult {
query: "ROLLBACK".to_owned(),
result: Ok(res),
}
}
Err(e) => {
return QueryResult {
query: "ROLLBACK".to_owned(),
result: Err(Box::new(e)),
}
}
}
}
pub async fn insert_into_table(
&self,
connection: &mut MySqlConnection,
csv_row: Vec<&DataEntry>,
) -> QueryResult {
/* Field_name, data_name */
let fields = csv_row
.iter()
.fold(("".to_owned(), "".to_owned()), |row, next_row| {
(
row.0
+ ", "
+ next_row
.curr_field_description
.as_ref()
.unwrap()
.field
.as_str(),
row.1 + "\'" + next_row.data.as_str() + "\', ",
)
});
let query = format!(
"INSERT INTO {}({}) VALUES({})",
self.name, fields.0, fields.1
);
match sqlx::query(&query).execute(connection).await {
Ok(res) => {
return QueryResult {
query,
result: Ok(res),
}
}
Err(e) => {
return QueryResult {
query,
result: Err(Box::new(e)),
}
}
}
}
pub async fn describe_table(&mut self, connection: &mut MySqlConnection) {
let qr_description: Vec<FieldDescription> =
sqlx::query_as(format!("DESCRIBE {}", self.name).as_str())
@ -85,6 +177,12 @@ struct QRTables {
#[sqlx(rename = "Tables_in_quotes")]
tables_in_quotes: String,
}
#[derive(Debug)]
pub struct QueryResult {
pub query: String,
pub result: Result<MySqlQueryResult, Box<dyn Error + Send>>,
}
#[derive(Default, Clone)]
pub struct DBLoginData {
pub user_name: String,

1
src/ui.rs
View file

@ -1,4 +1,5 @@
pub mod db_login_window;
pub mod db_transaction_window;
pub mod language;
pub mod table_window;
pub mod window_manager;

163
src/ui/db_transaction_window.rs Normal file
View file

@ -0,0 +1,163 @@
use egui::{Context, Ui};
use sqlx::mysql::MySqlQueryResult;
use std::error::Error;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::oneshot;
use crate::backend::backend_manager::Communication;
use crate::backend::database_handler::QueryResult;
pub struct DBTransactionWindow {
sender: Sender<Communication>,
/* (Query, Result) */
log_history: Vec<QueryResult>,
logs_receiver: Option<Receiver<QueryResult>>,
is_log_finished_receiver: Option<oneshot::Receiver<bool>>,
result: Option<Vec<Result<MySqlQueryResult, Box<dyn Error>>>>,
final_result_receiver: Option<oneshot::Receiver<QueryResult>>,
working_table_index: usize,
}
impl DBTransactionWindow {
pub fn default(
sender: Sender<Communication>,
working_table_index: usize,
) -> DBTransactionWindow {
DBTransactionWindow {
sender,
log_history: vec![],
result: None,
logs_receiver: None,
final_result_receiver: None,
is_log_finished_receiver: None,
working_table_index,
}
}
}
impl DBTransactionWindow {
pub fn show(&mut self, ctx: &Context, ui: &mut Ui) {
egui::Window::new("Database Transactions")
.id(egui::Id::new("Database Transactions"))
.resizable(false)
.collapsible(true)
.title_bar(true)
.movable(false)
.scroll2([false, true])
.enabled(true)
.anchor(egui::Align2::CENTER_CENTER, egui::Vec2::ZERO)
.show(ctx, |ui| {
self.log();
self.ui(ctx, ui);
});
}
pub fn log(&mut self) {
if self.logs_receiver.is_some() {
if let Ok(log) = self.logs_receiver.as_mut().unwrap().try_recv() {
self.log_history.push(log);
}
}
if self.is_log_finished_receiver.is_some() {
if let Ok(finished) = self.is_log_finished_receiver.as_mut().unwrap().try_recv() {
//self.result = Some(finished);
println!("FINISHED THE QUERY!!!");
}
}
let (log_sender, log_receiver) = channel(2);
let (finished_sender, finished_receiver) = oneshot::channel();
self.is_log_finished_receiver = Some(finished_receiver);
self.logs_receiver = Some(log_receiver);
self.sender
.try_send(Communication::StartInserting(
self.working_table_index,
log_sender,
finished_sender,
))
.unwrap_or_else(|_| println!("Failed to send startInserting"));
}
pub fn ui(&mut self, ctx: &Context, ui: &mut Ui) {
ui.add_enabled_ui(self.result.is_some(), |ui| {
if ui.button("Roll back").clicked() {
if self.final_result_receiver.is_none() {
let (sender, receiver) = oneshot::channel();
self.final_result_receiver = Some(receiver);
self.sender
.try_send(Communication::TryRollBack(sender))
.unwrap_or_else(|_| println!("failed sending TryCommit receiver"));
}
}
if ui.button("Commit").clicked() {
if self.final_result_receiver.is_none() {
let (sender, receiver) = oneshot::channel();
self.final_result_receiver = Some(receiver);
self.sender
.try_send(Communication::TryCommit(sender))
.unwrap_or_else(|_| println!("failed sending TryCommit receiver"));
}
}
});
/* DB Output Stuff */
egui::ScrollArea::vertical()
.auto_shrink([false; 2])
.show(ui, |ui| {
egui_extras::StripBuilder::new(ui)
.size(egui_extras::Size::remainder().at_least(100.0))
.size(egui_extras::Size::remainder())
.vertical(|mut strip| {
strip.cell(|ui| {
egui::ScrollArea::horizontal().show(ui, |ui| {
let mut table = egui_extras::TableBuilder::new(ui)
.striped(true)
.resizable(true)
.cell_layout(egui::Layout::left_to_right(egui::Align::LEFT))
.column(egui_extras::Column::auto().clip(false))
.column(egui_extras::Column::auto().clip(false))
.min_scrolled_height(0.0);
table
.header(20.0, |mut header| {
header.col(|ui| {
ui.strong("Query");
});
header.col(|ui| {
ui.strong("Result");
});
})
.body(|mut body| {
body.rows(
15.0,
self.log_history.len(),
|row_index, mut row| {
let log = self.log_history.get(row_index).unwrap();
row.col(|ui| {
ui.label(&log.query);
});
row.col(|ui| {
match &log.result {
Ok(rows) => {
ui.style_mut()
.visuals
.override_text_color =
Some(egui::Color32::RED);
ui.label(format!(
"Success! Rows affected: {}",
rows.rows_affected()
));
}
Err(e) => {
ui.style_mut()
.visuals
.override_text_color =
Some(egui::Color32::RED);
ui.label(format!("Error! {}", e));
}
}
ui.reset_style();
});
},
);
});
ui.scroll_to_cursor(Some(egui::Align::BOTTOM));
});
});
})
});
}
}

69
src/ui/table_window.rs
View file

@ -1,6 +1,6 @@
use crate::backend::backend_manager::Communication;
use crate::backend::csv_handler::ImportedData;
use crate::backend::database_handler::{Tables,TableField};
use crate::backend::database_handler::{TableField, Tables};
use egui::{ComboBox, Context, Ui};
use egui_extras::{Column, Size, StripBuilder, TableBuilder};
use std::error::Error;
@ -17,11 +17,12 @@ pub struct SpreadSheetWindow {
is_table_described_receiver: Option<oneshot::Receiver<bool>>,
is_table_described: bool,
is_table_matched_with_headers: bool,
commit_to_db_sender: Sender<usize>,
}
impl SpreadSheetWindow {
pub fn default(
sender: Sender<Communication>,
commit_to_db_sender: Sender<usize>,
csv_data_handle: Arc<Mutex<ImportedData>>,
db_table_data_handle: Arc<Mutex<Tables>>,
) -> SpreadSheetWindow {
@ -35,6 +36,7 @@ impl SpreadSheetWindow {
csv_data_handle,
db_table_data_handle,
current_table: None,
commit_to_db_sender,
}
}
}
@ -58,7 +60,21 @@ impl SpreadSheetWindow {
if ui.button("Save File").clicked() {
println!("Saving file lol");
}
//if ui.button("Current Working Table: ")
ui.add_space(ui.available_width() * 0.8);
let is_csv_parsed: bool;
if let Ok(csv_data) = self.csv_data_handle.try_lock() {
is_csv_parsed = csv_data.is_parsed.clone();
} else {
is_csv_parsed = false;
}
ui.add_enabled_ui(is_csv_parsed, |ui| {
if ui.button("Commit to Database").clicked() {
self.commit_to_db_sender
.try_send(self.current_table.unwrap())
.unwrap_or_else(|_| println!("failed to send committodb-UI"))
}
});
});
/* if db isn't connected, don't allow imports */
@ -126,24 +142,28 @@ impl SpreadSheetWindow {
});
/* If a table is selected, try if it's fields are discovered */
if !self.is_table_described{
if !self.is_table_described {
if let Some(table_i) = self.current_table {
if db_table_data.tables.get(table_i).unwrap().fields.is_some() {
self.is_table_described = true;
} else if self.is_table_described_receiver.is_some(){
match self.is_table_described_receiver.as_mut().unwrap().try_recv(){
Ok(res) => {self.is_table_matched_with_headers = res},
Err(e) => println!("Failed receiving is_table_described, {}",e),
} else if self.is_table_described_receiver.is_some() {
match self
.is_table_described_receiver
.as_mut()
.unwrap()
.try_recv()
{
Ok(res) => self.is_table_matched_with_headers = res,
Err(e) => println!("Failed receiving is_table_described, {}", e),
}
} else {
let (sender, receiver) = oneshot::channel();
self.is_table_described_receiver = Some(receiver);
self.sender
.try_send(Communication::GetTableDescription(table_i, sender))
.unwrap_or_else(|e| println!("Failed asking to describe table, {}", e));
}
} else{
let (sender, receiver ) = oneshot::channel();
self.is_table_described_receiver = Some(receiver);
self.sender
.try_send(Communication::GetTableDescription(table_i, sender))
.unwrap_or_else(|e| println!("Failed asking to describe table, {}", e));
}
}
}
}
}
@ -155,7 +175,13 @@ impl SpreadSheetWindow {
if let Ok(csv_data) = &mut self.csv_data_handle.try_lock() {
if let Ok(db_table_data) = &mut self.db_table_data_handle.try_lock() {
//ref to all fields in curr db table
let mut curr_db_table_fields: &mut Vec<TableField> = db_table_data.tables.get_mut(self.current_table.unwrap()).unwrap().fields.as_mut().unwrap();
let mut curr_db_table_fields: &mut Vec<TableField> = db_table_data
.tables
.get_mut(self.current_table.unwrap())
.unwrap()
.fields
.as_mut()
.unwrap();
let mut table = TableBuilder::new(ui)
.striped(true)
.cell_layout(egui::Layout::left_to_right(egui::Align::Center));
@ -171,6 +197,12 @@ impl SpreadSheetWindow {
.header(20., |mut header| {
for i in 0..csv_data.data.cols() {
header.col(|ui| {
if csv_data.parsed_cols.contains(&i){
/* If the whole col is parsed, change header bg to green, else normal */
ui.style_mut().visuals.override_text_color= Some(egui::Color32::GREEN);
} else {
ui.style_mut().visuals.override_text_color= Some(egui::Color32::RED);
}
let mut combo_box: ComboBox;
if csv_data.are_headers {
combo_box = ComboBox::new(
@ -217,6 +249,7 @@ impl SpreadSheetWindow {
}
}
});
ui.reset_style();
});
}
})

29
src/ui/window_manager.rs
View file

@ -5,10 +5,12 @@ use crate::ui::db_login_window::DBLoginWindow;
use crate::ui::table_window::SpreadSheetWindow;
use eframe::{run_native, App, NativeOptions};
use std::sync::Arc;
use tokio::sync::mpsc::Sender;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::oneshot;
use tokio::sync::Mutex;
use super::db_transaction_window::DBTransactionWindow;
pub fn create_ui(
sender: Sender<Communication>,
csv_data_handle: Arc<Mutex<ImportedData>>,
@ -18,7 +20,7 @@ pub fn create_ui(
drag_and_drop_support: true,
..Default::default()
};
let (open_db_transaction_window_sender, open_db_transaction_window_receiver) = channel(2);
run_native(
"CSQL",
win_option,
@ -26,15 +28,19 @@ pub fn create_ui(
Box::new(CSQL {
spreadsheet_window: SpreadSheetWindow::default(
sender.clone(),
open_db_transaction_window_sender,
csv_data_handle.clone(),
db_table_data_handle.clone(),
),
db_login_window: DBLoginWindow::default(sender.clone()),
db_transaction_window: None,
is_db_connection_verified_receiver: None,
sender,
is_db_connection_verified: false,
csv_data_handle,
db_table_data_handle,
open_db_transaction_window_receiver,
should_open_db_transaction_window: None,
})
}),
)
@ -43,11 +49,14 @@ pub fn create_ui(
struct CSQL {
spreadsheet_window: SpreadSheetWindow,
db_login_window: DBLoginWindow,
db_transaction_window: Option<DBTransactionWindow>,
sender: Sender<Communication>,
is_db_connection_verified_receiver: Option<oneshot::Receiver<bool>>,
is_db_connection_verified: bool,
csv_data_handle: Arc<Mutex<ImportedData>>,
db_table_data_handle: Arc<Mutex<Tables>>,
open_db_transaction_window_receiver: Receiver<usize>,
should_open_db_transaction_window: Option<usize>,
}
impl App for CSQL {
@ -58,6 +67,20 @@ impl App for CSQL {
self.spreadsheet_window
.show(ctx, ui, self.is_db_connection_verified);
if let Ok(resp) = self.open_db_transaction_window_receiver.try_recv() {
self.should_open_db_transaction_window = Some(resp);
}
if self.should_open_db_transaction_window.is_some() {
let db_transaction_window = DBTransactionWindow::default(
self.sender.clone(),
self.should_open_db_transaction_window.unwrap(),
);
self.db_transaction_window = Some(db_transaction_window);
self.db_transaction_window.as_mut().unwrap().show(ctx, ui);
}
/* Changes self if db connection is verified */
if let Some(oneshot_receiver) = &mut self.is_db_connection_verified_receiver {
if let Ok(boole) = oneshot_receiver.try_recv() {
self.is_db_connection_verified = boole;
@ -71,7 +94,7 @@ impl App for CSQL {
.try_send(Communication::AreCreditentialsValidated(sed))
{
Ok(_) => (),
Err(e) => println!("Failed to send oneshot, {}", e),
Err(e) => println!("Failed to send-verify db conn oneshot, {}", e),
}
}
});