kdv/pygadmin
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add the create and drop feature

Browse Source 
The table, specified by the user, can now be dropped and created in the
csv import dialog.
This commit is contained in:
Lea Laux
2021-02-18 14:06:17 +01:00
committed by KDV Admin
parent 7906123f47
commit ef9d6e6a20
2 changed files with 213 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
pygadmin/csv_importer.py
View File

@@ -1,4 +1,5 @@
import copy
import logging
import os
import csv
import re
@@ -90,20 +91,31 @@ class CSVImporter:
# Check the data type of the current column.
for check_column in range(len(current_row)):
# Get the old/previous data type for comparison.
old_data_type = self.data_types[check_column]
# If the data type is TEXT, break, because there is nothing to change. This data type works in every
# case.
if old_data_type != "TEXT":
# Get the current value.
value = current_row[check_column]
# Get the data type of the current value.
data_type = self.get_data_type(value)
# Try to parse and assume the data types. This process could fail caused by a wrong delimiter for the
# csv file, so the list of data types has a completely different length, causing an error.
try:
# Get the old/previous data type for comparison.
old_data_type = self.data_types[check_column]
# If the data type is TEXT, break, because there is nothing to change. This data type works in every
# case.
if old_data_type != "TEXT":
# Get the current value.
value = current_row[check_column]
# Get the data type of the current value.
data_type = self.get_data_type(value)
# If the data type is not null, write the data type in the data type list. Converting REAL to INT is
# not allowed.
if data_type != "NULL" and (not (old_data_type == "REAL" and data_type == "INT")):
self.data_types[check_column] = data_type
# If the data type is not null, write the data type in the data type list. Converting REAL to
# INT is not allowed.
if data_type != "NULL" and (not (old_data_type == "REAL" and data_type == "INT")):
self.data_types[check_column] = data_type
# Catch the index error of data type list, caused by a wrong delimiter in the csv file.
except IndexError:
# Log the incident.
logging.critical("The data types can not be assumed based on an index error. This is caused by a "
"wrong delimiter for the csv file.", exc_info=True)
# Raise the exception for further handling, for example in the user interface.
raise
def get_data_type(self, value):
"""
@@ -131,18 +143,18 @@ class CSVImporter:
# Return TEXT, if a match could not be made.
return "TEXT"
def create_table_for_csv_data(self):
def create_table_for_csv_data(self, create_statement=None):
"""
Create the table to store the csv data in the database.
Create the table to store the csv data in the database. Set the default parameter for the create statement to
None, so a statement is created out of the given csv file. If the parameter is used, a new create statement can
be set, for example by a user.
"""
# Get the create statement of the table.
create_statement = self.get_create_statement()
if create_statement is None:
# Get the create statement of the table.
create_statement = self.get_create_statement()
# Assign the create statement as query to the table.
self.database_query_executor.database_query = create_statement
# Execute!
self.database_query_executor.submit_and_execute_query()
self.execute_query(create_statement)
def get_create_statement(self, check_ddl=True):
"""
@@ -278,16 +290,7 @@ class CSVImporter:
insert_query = "{}{}".format(insert_query, value_query)
# Execute the insert query.
self.execute_insert_query(insert_query, parameter_list)
def execute_insert_query(self, insert_query, insert_parameters):
"""
Get the query and parameters for an insert and execute it with the database query executor.
"""
self.database_query_executor.database_query = insert_query
self.database_query_executor.database_query_parameter = insert_parameters
self.database_query_executor.submit_and_execute_query()
self.execute_query(insert_query, parameter_list)
def create_insert_query_begin(self):
"""
@@ -318,6 +321,46 @@ class CSVImporter:
# Return the begin of the query.
return insert_query
def drop_table(self, table_name=None):
"""
Drop the given table, defined by the table name. If the table name stays None as default parameter, use the
deduced name.
"""
# Get the statement.
drop_statement = self.get_drop_statement(table_name)
# Execute the query.
self.execute_query(drop_statement)
def get_drop_statement(self, table_name=None):
"""
Create the drop statement for the existing table.
"""
# If the given table name is None, use the attribute of the class, defined by the name of the csv file.
if table_name is None:
self.get_table_name()
table_name = self.table_name
# Create the statement.
drop_statement = "DROP TABLE {};".format(table_name)
# Return the statement.
return drop_statement
def execute_query(self, query, parameters=None):
"""
Execute the given query with the parameters. If the parameters are not given, the default None is used, so only
the query is executed.
"""
# Assign the query as database query for the executor.
self.database_query_executor.database_query = query
# Use the parameters.
self.database_query_executor.database_query_parameter = parameters
# Execute!
self.database_query_executor.submit_and_execute_query()
@staticmethod
def check_ddl_parameter(parameter):
"""

168
pygadmin/widgets/csv_import.py
View File

@@ -1,7 +1,8 @@
import sys
import time
from PyQt5.QtWidgets import QDialog, QGridLayout, QLabel, QApplication, QPushButton, QMessageBox, QLineEdit
from PyQt5.QtWidgets import QDialog, QGridLayout, QLabel, QApplication, QPushButton, QMessageBox, QLineEdit, \
QInputDialog
from pygadmin.connectionfactory import global_connection_factory
from pygadmin.csv_importer import CSVImporter
@@ -20,29 +21,65 @@ class CSVImportDialog(QDialog):
icon_adder = IconAdder()
icon_adder.add_icon_to_widget(self)
# TODO: Workaround for delimiter
self.csv_importer = CSVImporter(database_connection, csv_file)
# Get the result of an input dialog for getting the delimiter of the csv file.
delimiter_result = self.init_delimiter_question(csv_file)
# The second variable in the tuple of the input dialog result contains the success. So if the user cancels the
# dialog, the result contains False.
if delimiter_result[1] is False:
# Inform the user about the aborting of the process.
QMessageBox.information(self, "Abort", "The csv import process has been aborted by the user.")
# End the function with a return.
return
# At this point, the input dialog for the delimiter was successful, so the result given by the user can be used
# as delimiter for the csv file.
self.csv_importer = CSVImporter(database_connection, csv_file, delimiter=delimiter_result[0])
# If the csv file exists, initialize the relevant parts of the dialog.
if self.csv_importer.check_existence_csv_file():
# Parse the given csv file.
self.csv_importer.parse_csv_file()
# Try to get the data types of the csv file. A fail or an error is possible and caused by a wrong delimiter.
try:
# Get the data types of the given csv file.
self.csv_importer.assume_data_types()
# Catch the error with showing the error to the user and end the initialization of the dialog.
except IndexError:
self.init_error_ui("The given delimiter is wrong, so the csv file {} cannot be processed.".format(
csv_file))
return
# Initialize the user interface.
self.init_ui()
# Initialize the grid layout.
self.init_grid()
self.csv_importer.database_query_executor.result_data.connect(self.show_success)
self.csv_importer.database_query_executor.error.connect(self.show_error)
# TODO: Better functions for connecting for the signals of the database query executor.
# Show an error for a non existing file.
else:
self.init_error_ui(csv_file)
self.init_error_ui("The given csv file {} is invalid".format(csv_file))
def init_delimiter_question(self, csv_file):
"""
Create an input dialog for getting the delimiter of the csv file.
"""
# Use an input dialog for getting the csv delimiter with the default of a comma, because it is called comma
# separated values.
delimiter_result = QInputDialog.getText(self, "CSV Delimiter", "What is the delimiter of your csv file "
"{}?".format(csv_file), text=",")
# Return the result of the dialog with the input and the success of the dialog.
return delimiter_result
def init_ui(self):
"""
Initialize the user interface.
"""
# Parse the given csv file.
self.csv_importer.parse_csv_file()
# Get the data types of the given csv file.
self.csv_importer.assume_data_types()
# Get the create statement of the given csv file based on the column names in the file and the data types.
create_statement = self.csv_importer.get_create_statement()
@@ -58,7 +95,7 @@ class CSVImportDialog(QDialog):
# Delete the first element, because this element does only contain a "CREATE TABLE".
del column_definition_list[0]
# Delete the last element as closing part of the statement.
del column_definition_list[len(column_definition_list)-1]
del column_definition_list[len(column_definition_list) - 1]
# Split the column definition at commas.
column_definition_list = [column_definition.replace(",", "") for column_definition in column_definition_list]
@@ -71,7 +108,7 @@ class CSVImportDialog(QDialog):
# Split the column at whitespaces, so the
identifier_list = column.split(" ")
# Get the index of the last element based on the length of the identifier list.
last_element_index = len(identifier_list)-1
last_element_index = len(identifier_list) - 1
# The last item is the column data type, which is only one word with any whitespaces.
column_data_type = identifier_list[last_element_index]
# Delete the last element from the list, because it is not necessary anymore and possibly disturbs the
@@ -101,11 +138,23 @@ class CSVImportDialog(QDialog):
# Define a label for closing the statement.
self.close_label = QLabel(");")
# Use a button for creating the given table.
self.create_button = QPushButton("Create Table")
# Connect the button with the function for creating the table.
self.create_button.clicked.connect(self.create_table)
# Create a drop button for dropping the table.
self.drop_button = QPushButton("Drop Table")
# Connect the button with the signal for actually dropping the table.
self.drop_button.clicked.connect(self.drop_table)
# Connect the signals for the result data and the error with functions for processing.
self.csv_importer.database_query_executor.result_data.connect(self.process_create_drop_success)
self.csv_importer.database_query_executor.error.connect(self.process_create_drop_error)
# Under construction TODO
self.insert_button = QPushButton("Insert")
self.create_button = QPushButton("Create")
self.insert_button.clicked.connect(self.insert_data)
self.create_button.clicked.connect(self.create_table)
self.show()
def init_grid(self):
@@ -134,23 +183,27 @@ class CSVImportDialog(QDialog):
# Increment the count, so the next tuple of line edits is placed under those ones.
line_edit_count += 1
# Place the close label at the end of CREATE
# Place the close label at the end of CREATE.
grid_layout.addWidget(self.close_label, line_edit_count, 0)
# Set the create button at the end of the statement, created by the line edits.
grid_layout.addWidget(self.create_button, line_edit_count, 1)
# Set the drop button on the right of the widget.
grid_layout.addWidget(self.drop_button, 0, 4)
# Set the spacing of the grid.
grid_layout.setSpacing(10)
# Set the layout to the grid layout.
self.setLayout(grid_layout)
def init_error_ui(self, csv_file):
def init_error_ui(self, message):
"""
Initialize the user interface for the error case and show the name of the invalid csv file.
Initialize the user interface for the error case and use an error message for specifying the error.
"""
# Get the layout as grid layout.
grid_layout = QGridLayout(self)
# Add a label with an error.
grid_layout.addWidget(QLabel("The given csv file {} is invalid".format(csv_file)), 0, 0)
grid_layout.addWidget(QLabel(message), 0, 0)
self.setLayout(grid_layout)
self.setMaximumSize(10, 100)
self.showMaximized()
@@ -160,24 +213,82 @@ class CSVImportDialog(QDialog):
def create_table(self):
"""
TODO: Build create statement and use the csv importer for execution
Build the create statement specified by the user in the line edits and use the csv importer for execution of the
statement.
"""
self.csv_importer.create_table_for_csv_data()
create_statement = self.get_user_create_statement()
self.csv_importer.database_query_executor.result_data.connect(self.process_create_drop_success)
self.csv_importer.database_query_executor.error.connect(self.process_create_drop_error)
self.csv_importer.create_table_for_csv_data(create_statement)
def get_user_create_statement(self):
"""
Build the create statement specified by the user in the line edits.
"""
# Initialize the statement with the begin, the table name and the opening (.
create_statement = "{}{} {}\n".format(self.create_table_start_label.text(), self.table_name_line_edit.text(),
self.open_label.text())
# Iterate over the list of column line edits for getting the user data.
for line_edit_count in range(len(self.column_line_edit_list)):
# Get the line edit tuple.
line_edit = self.column_line_edit_list[line_edit_count]
# The first element contains the line edit for name of the column.
name_line_edit = line_edit[0]
# The second element contains the line edit for the data type of the column.
datatype_line_edit = line_edit[1]
# If the current line edit tuple is not the last one, use a comma as comma value for separating the column
# definitions.
if line_edit_count != len(self.column_line_edit_list) - 1:
comma_value = ","
# If the current line edit tuple is the last one, a comma is not necessary.
else:
comma_value = ""
# Add the column definition to the create statement.
create_statement = "{} {} {}{}\n".format(create_statement, name_line_edit.text(), datatype_line_edit.text(),
comma_value)
# Add the closing text to the create statement.
create_statement = "{}{}".format(create_statement, self.close_label.text())
# Return the result.
return create_statement
def drop_table(self):
"""
Drop the table specified in the line edit with the table name.
"""
table_name = self.table_name_line_edit.text()
self.csv_importer.drop_table(table_name)
def process_create_drop_success(self, result):
"""
Process the result of the table creation or dropping.
"""
# The parameter result contains a list. If the list is not empty, a success can be processed.
if result:
QMessageBox.information(self, "Create Success", "The result is {}".format(result))
def process_create_drop_error(self, result):
"""
Process the error during the create or drop process.
"""
QMessageBox.critical(self, "Create Error", "An error occurred during the create process: {}".format(result))
# under construction: TODO
def insert_data(self):
begin = time.time()
self.csv_importer.create_and_execute_insert_queries()
end = time.time()
print("Runtime: {}".format(end-begin))
def show_success(self, result):
QMessageBox.information(self, "Success", "The result is {}".format(result))
def show_error(self, error_message):
QMessageBox.critical(self, "Error", "{}".format(error_message))
print(error_message)
print("Runtime: {}".format(end - begin))
if __name__ == "__main__":
@@ -185,4 +296,3 @@ if __name__ == "__main__":
csv_import = CSVImportDialog(global_connection_factory.get_database_connection("localhost", "testuser", "testdb"),
"/home/sqlea/test.csv")
sys.exit(app.exec())