mirror of https://gitlab.com/pamhyr/pamhyr2
SQL: Move SQL class to separate file.
parent
b79fc28a5f
commit
41e14208e4
|
|
@ -23,7 +23,7 @@ import logging
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from functools import reduce
|
from functools import reduce
|
||||||
|
|
||||||
from tools import SQL
|
from SQL import SQL
|
||||||
from Model.Except import NotImplementedMethodeError
|
from Model.Except import NotImplementedMethodeError
|
||||||
|
|
||||||
logger = logging.getLogger()
|
logger = logging.getLogger()
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,133 @@
|
||||||
|
# SQL.py -- Pamhyr
|
||||||
|
# Copyright (C) 2023-2024 INRAE
|
||||||
|
#
|
||||||
|
# This program is free software: you can redistribute it and/or modify
|
||||||
|
# it under the terms of the GNU General Public License as published by
|
||||||
|
# the Free Software Foundation, either version 3 of the License, or
|
||||||
|
# (at your option) any later version.
|
||||||
|
#
|
||||||
|
# This program is distributed in the hope that it will be useful,
|
||||||
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||||
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||||
|
# GNU General Public License for more details.
|
||||||
|
#
|
||||||
|
# You should have received a copy of the GNU General Public License
|
||||||
|
# along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||||
|
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
|
import os
|
||||||
|
import logging
|
||||||
|
import sqlite3
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from tools import timer
|
||||||
|
|
||||||
|
logger = logging.getLogger()
|
||||||
|
|
||||||
|
|
||||||
|
class SQL(object):
|
||||||
|
def _init_db_file(self, db):
|
||||||
|
exists = Path(db).exists()
|
||||||
|
|
||||||
|
os.makedirs(
|
||||||
|
os.path.dirname(db),
|
||||||
|
exist_ok=True
|
||||||
|
)
|
||||||
|
|
||||||
|
self._db = sqlite3.connect(db)
|
||||||
|
self._cur = self._db.cursor()
|
||||||
|
|
||||||
|
if not exists:
|
||||||
|
self._create() # Create db
|
||||||
|
self._save() # Save
|
||||||
|
else:
|
||||||
|
self._update() # Update db scheme if necessary
|
||||||
|
self._load() # Load data
|
||||||
|
|
||||||
|
def __init__(self, filename=None):
|
||||||
|
self._db = None
|
||||||
|
|
||||||
|
if filename is not None:
|
||||||
|
self._init_db_file(filename)
|
||||||
|
|
||||||
|
def commit(self):
|
||||||
|
logger.debug("SQL - commit")
|
||||||
|
self._db.commit()
|
||||||
|
|
||||||
|
def _close(self):
|
||||||
|
self.commit()
|
||||||
|
self._db.close()
|
||||||
|
|
||||||
|
def _fetch_string(self, s):
|
||||||
|
return s.replace("'", "'")
|
||||||
|
|
||||||
|
def _fetch_tuple(self, tup):
|
||||||
|
res = []
|
||||||
|
for v in tup:
|
||||||
|
if type(v) is str:
|
||||||
|
v = self._fetch_string(v)
|
||||||
|
res.append(v)
|
||||||
|
|
||||||
|
return res
|
||||||
|
|
||||||
|
def _fetch_list(self, lst):
|
||||||
|
res = []
|
||||||
|
for v in lst:
|
||||||
|
if type(v) is str:
|
||||||
|
v = self._fetch_string(v)
|
||||||
|
elif type(v) is tuple:
|
||||||
|
v = self._fetch_tuple(v)
|
||||||
|
res.append(v)
|
||||||
|
|
||||||
|
return res
|
||||||
|
|
||||||
|
def _fetch(self, res, one):
|
||||||
|
if one:
|
||||||
|
value = res.fetchone()
|
||||||
|
else:
|
||||||
|
value = res.fetchall()
|
||||||
|
res = value
|
||||||
|
|
||||||
|
if type(value) is list:
|
||||||
|
res = self._fetch_list(value)
|
||||||
|
elif type(value) is tuple:
|
||||||
|
res = self._fetch_tuple(value)
|
||||||
|
|
||||||
|
return res
|
||||||
|
|
||||||
|
def _db_format(self, value):
|
||||||
|
# Replace ''' by ''' to preserve SQL injection
|
||||||
|
if type(value) is str:
|
||||||
|
value = value.replace("'", "'")
|
||||||
|
return value
|
||||||
|
|
||||||
|
@timer
|
||||||
|
def execute(self, cmd, fetch_one=True, commit=False):
|
||||||
|
logger.debug(f"SQL - {cmd}")
|
||||||
|
|
||||||
|
value = None
|
||||||
|
try:
|
||||||
|
res = self._cur.execute(cmd)
|
||||||
|
|
||||||
|
if commit:
|
||||||
|
self._db.commit()
|
||||||
|
|
||||||
|
value = self._fetch(res, fetch_one)
|
||||||
|
except Exception as e:
|
||||||
|
logger_exception(e)
|
||||||
|
finally:
|
||||||
|
return value
|
||||||
|
|
||||||
|
def _create(self):
|
||||||
|
logger.warning("TODO: Create")
|
||||||
|
|
||||||
|
def _update(self):
|
||||||
|
logger.warning("TODO: Update")
|
||||||
|
|
||||||
|
def _save(self):
|
||||||
|
logger.warning("TODO: Save")
|
||||||
|
|
||||||
|
def _load(self):
|
||||||
|
logger.warning("TODO: LOAD")
|
||||||
|
|
@ -29,6 +29,7 @@ from PyQt5.QtGui import (
|
||||||
|
|
||||||
from PyQt5.QtCore import (
|
from PyQt5.QtCore import (
|
||||||
Qt, QTranslator, QEvent, QUrl, QTimer,
|
Qt, QTranslator, QEvent, QUrl, QTimer,
|
||||||
|
QCoreApplication,
|
||||||
)
|
)
|
||||||
from PyQt5.QtWidgets import (
|
from PyQt5.QtWidgets import (
|
||||||
QMainWindow, QApplication, QAction,
|
QMainWindow, QApplication, QAction,
|
||||||
|
|
@ -72,6 +73,8 @@ except Exception as e:
|
||||||
|
|
||||||
from Model.Study import Study
|
from Model.Study import Study
|
||||||
|
|
||||||
|
_translate = QCoreApplication.translate
|
||||||
|
|
||||||
logger = logging.getLogger()
|
logger = logging.getLogger()
|
||||||
|
|
||||||
no_model_action = [
|
no_model_action = [
|
||||||
|
|
@ -296,16 +299,24 @@ class ApplicationWindow(QMainWindow, ListedSubWindow, WindowToolKit):
|
||||||
|
|
||||||
def setup_debug_mode(self, init=False):
|
def setup_debug_mode(self, init=False):
|
||||||
menu = self.findChild(QMenu, "menu_help")
|
menu = self.findChild(QMenu, "menu_help")
|
||||||
|
menu.setToolTipsVisible(True)
|
||||||
|
|
||||||
self.set_title()
|
self.set_title()
|
||||||
|
|
||||||
if init:
|
if init:
|
||||||
self.debug_action = QAction("Debug", self)
|
self.debug_action = QAction("Debug", self)
|
||||||
self.debug_action.setStatusTip("Debug")
|
self.debug_action.setToolTip(
|
||||||
|
_translate("MainWindow", "Open debug window")
|
||||||
|
)
|
||||||
self.debug_action.triggered.connect(self.open_debug)
|
self.debug_action.triggered.connect(self.open_debug)
|
||||||
|
|
||||||
self.debug_sqlite_action = QAction("Debug SQLite", self)
|
self.debug_sqlite_action = QAction("Debug SQLite", self)
|
||||||
self.debug_sqlite_action.setStatusTip(
|
self.debug_sqlite_action.setToolTip(
|
||||||
"Open SQLite debuging tool (sqlitebrowser)")
|
_translate(
|
||||||
|
"MainWindow",
|
||||||
|
"Open SQLite debuging tool ('sqlitebrowser')"
|
||||||
|
)
|
||||||
|
)
|
||||||
self.debug_sqlite_action.triggered.connect(self.open_sqlite)
|
self.debug_sqlite_action.triggered.connect(self.open_sqlite)
|
||||||
|
|
||||||
if self.conf.debug:
|
if self.conf.debug:
|
||||||
|
|
@ -599,6 +610,7 @@ class ApplicationWindow(QMainWindow, ListedSubWindow, WindowToolKit):
|
||||||
|
|
||||||
def _do_update_window_list(self):
|
def _do_update_window_list(self):
|
||||||
menu = self.findChild(QMenu, "menu_windows")
|
menu = self.findChild(QMenu, "menu_windows")
|
||||||
|
menu.setToolTipsVisible(True)
|
||||||
|
|
||||||
# Remove all actions
|
# Remove all actions
|
||||||
menu.clear()
|
menu.clear()
|
||||||
|
|
@ -622,9 +634,9 @@ class ApplicationWindow(QMainWindow, ListedSubWindow, WindowToolKit):
|
||||||
return lambda: self._activate_window(h)
|
return lambda: self._activate_window(h)
|
||||||
|
|
||||||
action = QAction(window._title, self)
|
action = QAction(window._title, self)
|
||||||
# action.setStatusTip(
|
action.setToolTip(
|
||||||
# _translate("MainWindowdow", "Activate this windowdow")
|
_translate("MainWindow", "Activate this window")
|
||||||
# )
|
)
|
||||||
h = window.hash()
|
h = window.hash()
|
||||||
fn = lambda_generator(h)
|
fn = lambda_generator(h)
|
||||||
action.triggered.connect(fn)
|
action.triggered.connect(fn)
|
||||||
|
|
|
||||||
|
|
@ -20,7 +20,7 @@ import os
|
||||||
import pickle
|
import pickle
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
from tools import SQL
|
from SQL import SQL
|
||||||
|
|
||||||
from Model.Stricklers.Stricklers import Stricklers
|
from Model.Stricklers.Stricklers import Stricklers
|
||||||
from Model.Stricklers.StricklersList import StricklersList
|
from Model.Stricklers.StricklersList import StricklersList
|
||||||
|
|
|
||||||
115
src/tools.py
115
src/tools.py
|
|
@ -18,7 +18,6 @@
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import time
|
import time
|
||||||
import sqlite3
|
|
||||||
import logging
|
import logging
|
||||||
import traceback
|
import traceback
|
||||||
|
|
||||||
|
|
@ -251,120 +250,6 @@ def old_pamhyr_date_to_timestamp(date: str):
|
||||||
|
|
||||||
return ts
|
return ts
|
||||||
|
|
||||||
#######
|
|
||||||
# SQL #
|
|
||||||
#######
|
|
||||||
|
|
||||||
# This class is an abstract class to make class with save and load
|
|
||||||
# from sqlite3.
|
|
||||||
|
|
||||||
|
|
||||||
class SQL(object):
|
|
||||||
def _init_db_file(self, db):
|
|
||||||
exists = Path(db).exists()
|
|
||||||
|
|
||||||
os.makedirs(
|
|
||||||
os.path.dirname(db),
|
|
||||||
exist_ok=True
|
|
||||||
)
|
|
||||||
|
|
||||||
self._db = sqlite3.connect(db)
|
|
||||||
self._cur = self._db.cursor()
|
|
||||||
|
|
||||||
if not exists:
|
|
||||||
self._create() # Create db
|
|
||||||
self._save() # Save
|
|
||||||
else:
|
|
||||||
self._update() # Update db scheme if necessary
|
|
||||||
self._load() # Load data
|
|
||||||
|
|
||||||
def __init__(self, filename=None):
|
|
||||||
self._db = None
|
|
||||||
|
|
||||||
if filename is not None:
|
|
||||||
self._init_db_file(filename)
|
|
||||||
|
|
||||||
def commit(self):
|
|
||||||
logger.debug("SQL - commit")
|
|
||||||
self._db.commit()
|
|
||||||
|
|
||||||
def _close(self):
|
|
||||||
self.commit()
|
|
||||||
self._db.close()
|
|
||||||
|
|
||||||
def _fetch_string(self, s):
|
|
||||||
return s.replace("'", "'")
|
|
||||||
|
|
||||||
def _fetch_tuple(self, tup):
|
|
||||||
res = []
|
|
||||||
for v in tup:
|
|
||||||
if type(v) is str:
|
|
||||||
v = self._fetch_string(v)
|
|
||||||
res.append(v)
|
|
||||||
|
|
||||||
return res
|
|
||||||
|
|
||||||
def _fetch_list(self, lst):
|
|
||||||
res = []
|
|
||||||
for v in lst:
|
|
||||||
if type(v) is str:
|
|
||||||
v = self._fetch_string(v)
|
|
||||||
elif type(v) is tuple:
|
|
||||||
v = self._fetch_tuple(v)
|
|
||||||
res.append(v)
|
|
||||||
|
|
||||||
return res
|
|
||||||
|
|
||||||
def _fetch(self, res, one):
|
|
||||||
if one:
|
|
||||||
value = res.fetchone()
|
|
||||||
else:
|
|
||||||
value = res.fetchall()
|
|
||||||
res = value
|
|
||||||
|
|
||||||
if type(value) is list:
|
|
||||||
res = self._fetch_list(value)
|
|
||||||
elif type(value) is tuple:
|
|
||||||
res = self._fetch_tuple(value)
|
|
||||||
|
|
||||||
return res
|
|
||||||
|
|
||||||
def _db_format(self, value):
|
|
||||||
# Replace ''' by ''' to preserve SQL injection
|
|
||||||
if type(value) is str:
|
|
||||||
value = value.replace("'", "'")
|
|
||||||
return value
|
|
||||||
|
|
||||||
@timer
|
|
||||||
def execute(self, cmd, fetch_one=True, commit=False):
|
|
||||||
logger.debug(f"SQL - {cmd}")
|
|
||||||
|
|
||||||
value = None
|
|
||||||
try:
|
|
||||||
res = self._cur.execute(cmd)
|
|
||||||
|
|
||||||
if commit:
|
|
||||||
self._db.commit()
|
|
||||||
|
|
||||||
value = self._fetch(res, fetch_one)
|
|
||||||
except Exception as e:
|
|
||||||
logger_exception(e)
|
|
||||||
finally:
|
|
||||||
return value
|
|
||||||
|
|
||||||
def _create(self):
|
|
||||||
logger.warning("TODO: Create")
|
|
||||||
|
|
||||||
def _update(self):
|
|
||||||
logger.warning("TODO: Update")
|
|
||||||
|
|
||||||
def _save(self):
|
|
||||||
logger.warning("TODO: Save")
|
|
||||||
|
|
||||||
def _load(self):
|
|
||||||
logger.warning("TODO: LOAD")
|
|
||||||
|
|
||||||
|
|
||||||
#######################
|
#######################
|
||||||
# COMMAND LINE PARSER #
|
# COMMAND LINE PARSER #
|
||||||
#######################
|
#######################
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue