Взаимодействие с БД Postgres
Работаю с формой, на форме выполняются результаты выполнения запросов, но в самой бд никаких изменений. Также не работает запрос на вывод данных бд за условием. С чем может связана причина? Или библиотеки psycopg2 не достаточно здесь и нужно искать решения с помощью SQLAlchemy?
def connect_to_database(self):
try:
self.conn = psycopg2.connect(**self.db_params)
print("Connected to the database.")
return self.conn
except (Exception, psycopg2.Error) as error:
print(f"Error while connecting to PostgreSQL: {error}")
return None
def select_data(self):
self.set_column_widths()
table_entry = self.tables_combobox.get()
text_field_entry = self.query_text_field.get("1.0", "end-1c")
if text_field_entry == "WHERE":
self.cursor.execute(f"SELECT * FROM {table_entry}")
result = self.cursor.fetchall()
else:
sql_query = f"SELECT * FROM {table_entry} {text_field_entry}"
print(sql_query)
self.cursor.execute(sql_query)
result = self.cursor.fetchall()
if result:
self.data_grid.delete(*self.data_grid.get_children())
for row in result:
self.data_grid.insert("", "end", values=list(row))
def delete_data(self):
self.set_column_widths()
table_name = self.tables_combobox.get()
selected_index = self.data_grid.selection()[0]
selected_item = self.data_grid.item(selected_index)
selected_values = selected_item['values']
if not self.data_grid.selection():
messagebox.showwarning("Попередження", "Будь ласка, виберіть рядок для оновлення.")
return
query = f"DELETE FROM {table_name} WHERE id = {selected_values[0]}"
try:
self.cursor.execute(query)
except Exception as e:
print(f"An error occurred: {e}")
self.data_grid.delete(selected_index)
def update_data(self):
self.set_column_widths()
table_name = self.tables_combobox.get()
if not self.data_grid.selection():
messagebox.showwarning("Попередження", "Будь ласка, виберіть рядок для оновлення.")
return
index = self.data_grid.selection()[0]
id_value = self.data_grid.item(index)['values'][0]
new_values = []
for entry in self.entries:
value = entry.get()
if isinstance(value, str):
new_values.append(f"'{value}'")
else:
new_values.append(str(value))
query = f"UPDATE {table_name} SET {', '.join([f'{col}={val}' for col, val in zip(self.data_grid['columns'], new_values)])} WHERE id = {id_value}"
self.cursor.execute(query)
self.data_grid.item(index, values=new_values)
def add_data(self):
self.set_column_widths()
table_name = self.tables_combobox.get()
new_values = []
for entry in self.entries:
value = entry.get()
if isinstance(value, str):
new_values.append(f"'{value}'")
else:
new_values.append(str(value))
query = f"INSERT INTO {table_name} ({', '.join(self.data_grid['columns'])}) VALUES ({', '.join(new_values)})"
self.cursor.execute(query)
self.data_grid.insert("", "end", values=new_values)