ske087/db_interface
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: persistent DB host config, async DB ops, Exit button, test connection

Browse Source 
- Save/load DB host from config.json (works both in dev and built .exe)
- All DB operations moved to background threads to prevent UI blocking
- Connection timeout set to 5s to avoid long hangs on unreachable host
- Added Test Connection button to Settings popup
- Added red Exit button fixed to bottom-right corner of main screen
- Updated DatabaseApp.spec with full Kivy deps (sdl2, glew, angle) and hidden imports
- PyInstaller-aware base path using sys.frozen for config.json persistence
This commit is contained in:
scheianu
2026-04-01 21:00:45 +03:00
parent 8ae60a77e4
commit c912bac2dc
4 changed files with 220 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -8,7 +8,6 @@ venv/
env/
ENV/
*.egg-info/
dist/
build/
*.egg

29
DatabaseApp.spec
View File

@@ -1,12 +1,36 @@
# -*- mode: python ; coding: utf-8 -*-
from kivy_deps import sdl2, glew, angle
a = Analysis(
['main.py'],
pathex=[],
binaries=[],
datas=[],
hiddenimports=['mysql.connector', 'kivy.core.window.window_sdl2', 'win32timezone'],
hiddenimports=[
'mysql.connector',
'mysql.connector.locales',
'mysql.connector.locales.eng',
'kivy',
'kivy.core.window',
'kivy.core.window.window_sdl2',
'kivy.core.text',
'kivy.core.text.text_sdl2',
'kivy.core.image',
'kivy.core.image.img_sdl2',
'kivy.core.image.img_pil',
'kivy.core.audio',
'kivy.core.clipboard',
'kivy.core.clipboard.clipboard_sdl2',
'kivy.core.spelling',
'kivy.graphics',
'kivy.graphics.cgl_backend',
'kivy.graphics.cgl_backend.cgl_glew',
'kivy.input.providers',
'kivy.input.providers.wm_touch',
'kivy.input.providers.wm_pen',
'win32timezone',
'pkg_resources.py2_compat',
],
hookspath=[],
hooksconfig={},
runtime_hooks=[],
@@ -21,6 +45,7 @@ exe = EXE(
a.scripts,
a.binaries,
a.datas,
*[Tree(p) for p in (sdl2.dep_bins + glew.dep_bins + angle.dep_bins)],
[],
name='DatabaseApp',
debug=False,

60
database_manager.py
View File

@@ -1,6 +1,18 @@
import mysql.connector
from mysql.connector import Error
from typing import List, Tuple, Optional
import json
import os
import sys
# When frozen by PyInstaller, __file__ points to a temp folder that is deleted on exit.
# sys.executable points to the .exe location, which is persistent.
if getattr(sys, 'frozen', False):
_BASE_DIR = os.path.dirname(sys.executable)
else:
_BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CONFIG_FILE = os.path.join(_BASE_DIR, 'config.json')
class DatabaseManager:
"""
@@ -9,12 +21,53 @@ class DatabaseManager:
"""
def __init__(self):
self.host = "localhost"
self.host = self._load_host()
self.database = "cantare_injectie"
self.user = "omron"
self.password = "Initial01!"
self.connection = None
self.init_database()
# init_database() is called asynchronously from the UI layer to avoid blocking
def _load_host(self) -> str:
"""Load the database host from the config file, falling back to localhost."""
try:
if os.path.exists(CONFIG_FILE):
with open(CONFIG_FILE, 'r') as f:
data = json.load(f)
return data.get('host', 'localhost')
except Exception as e:
print(f"Could not read config file: {e}")
return 'localhost'
def save_host(self, host: str):
"""Persist the database host to the config file."""
try:
data = {}
if os.path.exists(CONFIG_FILE):
with open(CONFIG_FILE, 'r') as f:
data = json.load(f)
data['host'] = host
with open(CONFIG_FILE, 'w') as f:
json.dump(data, f, indent=2)
except Exception as e:
print(f"Could not save config file: {e}")
def test_connection(self, host: str) -> tuple:
"""Test connectivity to the database using the given host. Returns (success: bool, message: str)."""
try:
test_conn = mysql.connector.connect(
host=host,
database=self.database,
user=self.user,
password=self.password,
connection_timeout=5
)
if test_conn.is_connected():
test_conn.close()
return True, f"Connected successfully to '{host}'"
except Error as e:
return False, str(e)
return False, "Connection failed"
def get_connection(self):
"""Get a database connection."""
@@ -24,7 +77,8 @@ class DatabaseManager:
host=self.host,
database=self.database,
user=self.user,
password=self.password
password=self.password,
connection_timeout=5
)
return self.connection
except Error as e:

187
main.py
View File

@@ -1,5 +1,8 @@
import threading
from kivy.app import App
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.anchorlayout import AnchorLayout
from kivy.uix.floatlayout import FloatLayout
from kivy.uix.gridlayout import GridLayout
from kivy.uix.label import Label
from kivy.uix.textinput import TextInput
@@ -18,8 +21,13 @@ class DatabaseApp(App):
def build(self):
# Set window to fullscreen
Window.fullscreen = 'auto'
# Root float layout so we can overlay the exit button
root = FloatLayout()
# Main layout with better spacing for fullscreen
main_layout = BoxLayout(orientation='vertical', padding=40, spacing=20)
main_layout = BoxLayout(orientation='vertical', padding=40, spacing=20,
size_hint=(1, 1), pos_hint={'x': 0, 'y': 0})
# Top spacer for vertical centering
main_layout.add_widget(Label(size_hint_y=0.15))
@@ -115,11 +123,31 @@ class DatabaseApp(App):
# Bottom spacer for vertical centering
main_layout.add_widget(Label(size_hint_y=0.15))
# Removed database contents frame
# Load initial data
Clock.schedule_once(self.refresh_data, 0.1)
root.add_widget(main_layout)
return main_layout
# Exit button — fixed to the bottom-right corner
exit_btn = Button(
text='Exit',
font_size=20,
bold=True,
background_color=(0.85, 0.1, 0.1, 1),
color=(1, 1, 1, 1),
size_hint=(None, None),
size=(120, 55),
pos_hint={'right': 1, 'y': 0}
)
exit_btn.bind(on_press=lambda x: self.stop())
root.add_widget(exit_btn)
# Init DB and load data in background so the UI appears immediately
def _init_db(dt):
def _do():
self.db_manager.init_database()
Clock.schedule_once(lambda dt2: self.refresh_data(None))
threading.Thread(target=_do, daemon=True).start()
Clock.schedule_once(_init_db, 0.1)
return root
def set_update_frame_enabled(self, enabled):
self.update_id_input.readonly = not enabled
self.update_mass_input.readonly = not enabled
@@ -149,20 +177,24 @@ class DatabaseApp(App):
self.show_status("ID must be 20 characters or less", error=True)
return
# Show searching status
self.show_status("Searching...", error=False)
try:
record = self.db_manager.search_by_id(record_id)
if record:
self.mass_input.text = str(record[1]) # Set the mass field
self.show_status(f"Found: {record[0]} = {record[1]}")
self.highlight_record(record_id)
else:
self.show_status(f"ID '{record_id}' not found in database", error=True)
self.mass_input.text = ""
except Exception as e:
self.show_status(f"Search error: {str(e)}", error=True)
def _do():
try:
record = self.db_manager.search_by_id(record_id)
def _update(dt):
if record:
self.mass_input.text = str(record[1])
self.show_status(f"Found: {record[0]} = {record[1]}")
self.highlight_record(record_id)
else:
self.show_status(f"ID '{record_id}' not found in database", error=True)
self.mass_input.text = ""
Clock.schedule_once(_update)
except Exception as e:
err = str(e)
Clock.schedule_once(lambda dt: self.show_status(f"Search error: {err}", error=True))
threading.Thread(target=_do, daemon=True).start()
def add_update_record(self, instance):
"""Add or update a record from the update frame."""
@@ -179,20 +211,25 @@ class DatabaseApp(App):
except ValueError:
self.show_status("Mass must be a valid number", error=True)
return
success = self.db_manager.add_or_update_record(record_id, mass)
if success:
existing = self.db_manager.search_by_id(record_id)
if existing:
self.show_status(f"Successfully added/updated: {record_id} = {mass}")
self.refresh_data(None)
# Clear update frame after successful operation
self.update_id_input.text = ""
self.update_mass_input.text = ""
self.set_update_frame_enabled(False)
else:
self.show_status("Operation completed but record not found", error=True)
else:
self.show_status("Failed to add/update record", error=True)
self.show_status("Saving...", error=False)
def _do():
try:
success = self.db_manager.add_or_update_record(record_id, mass)
def _update(dt):
if success:
self.show_status(f"Successfully added/updated: {record_id} = {mass}")
self.update_id_input.text = ""
self.update_mass_input.text = ""
self.set_update_frame_enabled(False)
self.refresh_data(None)
else:
self.show_status("Failed to add/update record", error=True)
Clock.schedule_once(_update)
except Exception as e:
err = str(e)
Clock.schedule_once(lambda dt: self.show_status(f"Save error: {err}", error=True))
threading.Thread(target=_do, daemon=True).start()
def delete_record(self, instance):
"""Delete a record using the update frame fields."""
@@ -209,17 +246,26 @@ class DatabaseApp(App):
def confirm_delete(self, record_id):
"""Confirm and execute deletion."""
success = self.db_manager.delete_record(record_id)
if success:
self.show_status(f"Successfully deleted: {record_id}")
self.refresh_data(None)
self.clear_fields()
# Clear update frame fields
self.update_id_input.text = ""
self.update_mass_input.text = ""
self.set_update_frame_enabled(False)
else:
self.show_status(f"Failed to delete ID '{record_id}' (not found)", error=True)
self.show_status("Deleting...", error=False)
def _do():
try:
success = self.db_manager.delete_record(record_id)
def _update(dt):
if success:
self.show_status(f"Successfully deleted: {record_id}")
self.clear_fields()
self.update_id_input.text = ""
self.update_mass_input.text = ""
self.set_update_frame_enabled(False)
self.refresh_data(None)
else:
self.show_status(f"Failed to delete ID '{record_id}' (not found)", error=True)
Clock.schedule_once(_update)
except Exception as e:
err = str(e)
Clock.schedule_once(lambda dt: self.show_status(f"Delete error: {err}", error=True))
threading.Thread(target=_do, daemon=True).start()
def clear_fields(self):
"""Clear the ID and mass fields."""
@@ -234,13 +280,16 @@ class DatabaseApp(App):
self.show_status("Fields cleared", error=False)
def refresh_data(self, instance):
"""Refresh the data display."""
try:
records = self.db_manager.read_all_data()
count = len(records)
self.show_status(f"Data refreshed - {count} records found")
except Exception as e:
self.show_status(f"Error refreshing data: {str(e)}", error=True)
"""Refresh the data display (runs in a background thread)."""
def _do():
try:
records = self.db_manager.read_all_data()
count = len(records)
Clock.schedule_once(lambda dt: self.show_status(f"Data refreshed - {count} records found"))
except Exception as e:
err = str(e)
Clock.schedule_once(lambda dt: self.show_status(f"Error refreshing data: {err}", error=True))
threading.Thread(target=_do, daemon=True).start()
def highlight_record(self, record_id):
"""Highlight a specific record in the display."""
@@ -292,7 +341,20 @@ class DatabaseApp(App):
height=60
)
content.add_widget(info_label)
# Test connection button + result label
test_btn = Button(text='Test Connection', font_size=20, bold=True, size_hint_y=None, height=55)
content.add_widget(test_btn)
test_result_label = Label(
text='',
font_size=17,
size_hint_y=None,
height=40,
bold=True
)
content.add_widget(test_result_label)
# Buttons
buttons = BoxLayout(size_hint_y=None, height=60, spacing=15)
@@ -306,13 +368,14 @@ class DatabaseApp(App):
popup = Popup(
title='Settings',
content=content,
size_hint=(0.6, 0.5)
size_hint=(0.6, 0.65)
)
def save_settings():
new_host = ip_input.text.strip()
if new_host:
self.db_manager.host = new_host
self.db_manager.save_host(new_host)
# Reconnect with new settings
if self.db_manager.connection:
try:
@@ -325,6 +388,28 @@ class DatabaseApp(App):
else:
self.show_status("Please enter a valid IP address", error=True)
def test_connection(instance):
host = ip_input.text.strip()
if not host:
test_result_label.text = 'Please enter a host first.'
test_result_label.color = (1, 0.6, 0, 1)
return
test_result_label.text = 'Testing...'
test_result_label.color = (1, 1, 1, 1)
def _do():
success, message = self.db_manager.test_connection(host)
def _update(dt):
if success:
test_result_label.text = f'OK: {message}'
test_result_label.color = (0, 0.85, 0, 1)
else:
test_result_label.text = f'Failed: {message}'
test_result_label.color = (1, 0.2, 0.2, 1)
Clock.schedule_once(_update)
threading.Thread(target=_do, daemon=True).start()
test_btn.bind(on_press=test_connection)
save_btn.bind(on_press=lambda x: save_settings())
cancel_btn.bind(on_press=popup.dismiss)