WCE/NAFKA-crm-gui
1
0
Fork 0
generated from dopt-python/py311
Code Issues Pull Requests Actions Packages Projects 1 Releases Wiki Activity

basic saving and loading of consultation sessions

Browse Source
This commit is contained in:
Florian Förster
2026-07-10 11:52:49 +02:00
parent 6f7593ac9d
commit 7c79bf8e91
2 changed files with 346 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wce_crm/backend/backend.py
+143 -44
View File
@@ -6,10 +6,11 @@ from typing import TYPE_CHECKING, Any, cast
import polars as pl
import sqlalchemy as sql
from dopt_basics.result_pattern import wrap_result
from wce_crm import db
from wce_crm.constants import TIMEZONE_CEST
from wce_crm.data_models import Beratungsgespraech_Vorgang
from wce_crm.data_models import Beratungsgespraech_Einzelgespraech, Beratungsgespraech_Vorgang
from wce_crm.logging import logger_back as logger
from wce_crm.types import CompanyInfo, ContactPersonInfo, InitRecType, MainPageEntry
@@ -140,16 +141,19 @@ def initrec_company_get_initial_recording(
stmt = db.grunderfassung_unternehmen.select().where(
db.grunderfassung_unternehmen.c.un_id == id_
)
with db.ENGINE.begin() as conn:
with db.ENGINE.connect() as conn:
ret = conn.execute(stmt)
if ret.rowcount == 0:
raise KeyError(f"Database ID {id_} not found")
results = ret.mappings().all()
if not results:
raise KeyError(f"Database ID {id_} not found")
row = ret.fetchone()
assert len(results) == 1, "more than one company initial recording obtained"
row = results[0]
assert row, "row was not obtained"
return row._asdict() # type: ignore
return dict(row)
def initrec_company_delete_initial_recording(
@@ -202,16 +206,19 @@ def initrec_person_get_initial_recording(
stmt = db.grunderfassung_personen.select().where(
db.grunderfassung_personen.c.pers_id == id_
)
with db.ENGINE.begin() as conn:
with db.ENGINE.connect() as conn:
ret = conn.execute(stmt)
if ret.rowcount == 0:
raise KeyError(f"Database ID {id_} not found")
results = ret.mappings().all()
if not results:
raise KeyError(f"Database ID {id_} not found")
row = ret.fetchone()
assert len(results) == 1, "more than one person initial recording obtained"
row = results[0]
assert row, "row was not obtained"
return row._asdict() # type: ignore
return dict(row)
def initrec_person_delete_initial_recording(
@@ -228,50 +235,142 @@ def initrec_person_delete_initial_recording(
raise KeyError(f"Database ID {id_} not found for deletion")
@wrap_result(10)
def page_consulting_to_db(
data: Beratungsgespraech_Vorgang,
) -> None:
logger.debug("[Consulting Page] Call Database Routine...")
fk_vorgang_id: int
if data.vorgang_id is None:
# insert new "Vorgang"
insert_data = data.model_dump(exclude={"vorgang_id", "beratungen"})
logger.debug("[Consulting Page] Call insert with data:\n%s", pformat(insert_data))
fk_vorgang_id = 42 # TODO replace for real inserted PK
else:
fk_vorgang_id = data.vorgang_id
logger.debug("[Consulting Page] VorgangID already set. ID: %d", fk_vorgang_id)
consultation_data: Beratungsgespraech_Vorgang,
) -> Beratungsgespraech_Vorgang:
logger.debug("[Consulting Page] Call database saving routine...")
rows_for_db_insert: list[dict[str, Any]] = []
rows_for_db_update: list[dict[str, Any]] = []
with db.ENGINE.begin() as conn:
if consultation_data.vorgang_id is None:
# insert new "Vorgang"
insert_data = consultation_data.model_dump(exclude={"vorgang_id", "beratungen"})
logger.debug(
"[Consulting Page] Call insert 'Vorgang' with data:\n%s", pformat(insert_data)
)
stmt = db.beratung_vorgang.insert()
ret = conn.execute(stmt, insert_data)
if ret.rowcount == 0:
raise IOError("Entry was not inserted correctly")
prim_keys = ret.inserted_primary_key
assert prim_keys
consultation_data.vorgang_id = cast("ConsId", prim_keys[0])
logger.debug("[Consulting Page] Inserted 'Vorgang' successfully")
for cons_session in data.beratungen:
cons_session.vorgang_id = fk_vorgang_id
if cons_session.beratung_id is None:
row_data = cons_session.model_dump(exclude={"beratung_id"})
rows_for_db_insert.append(row_data)
else:
row_data = cons_session.model_dump()
rows_for_db_update.append(row_data)
logger.debug(
"[Consulting Page] VorgangID already set. ID: %d",
consultation_data.vorgang_id,
)
if rows_for_db_update:
# ... update
logger.debug(
"[Consulting Page] Call update for sessions:\n%s", pformat(rows_for_db_update)
)
if rows_for_db_insert:
# ... insert
logger.debug(
"[Consulting Page] Call insert for sessions:\n%s", pformat(rows_for_db_insert)
)
rows_for_db_insert: list[dict[str, Any]] = []
rows_for_db_update: list[dict[str, Any]] = []
cons_sessions_inserted: list[Beratungsgespraech_Einzelgespraech] = []
for cons_session in consultation_data.beratungen:
cons_session.vorgang_id = consultation_data.vorgang_id
if cons_session.beratung_id is None:
row_data = cons_session.model_dump(exclude={"beratung_id"})
rows_for_db_insert.append(row_data)
cons_sessions_inserted.append(cons_session)
else:
row_data = cons_session.model_dump()
# new bind param to avoid name clashes
row_data["b_beratung_id"] = row_data["beratung_id"]
del row_data["beratung_id"]
rows_for_db_update.append(row_data)
if rows_for_db_update:
# ... update
logger.debug(
"[Consulting Page] Call update for sessions:\n%s", pformat(rows_for_db_update)
)
stmt = db.beratung_einzelberatung.update().where(
db.beratung_einzelberatung.c.beratung_id == sql.bindparam("b_beratung_id")
)
conn.execute(stmt, rows_for_db_update)
if rows_for_db_insert:
# ... insert
logger.debug(
"[Consulting Page] Call insert for sessions:\n%s", pformat(rows_for_db_insert)
)
stmt = db.beratung_einzelberatung.insert().returning(
db.beratung_einzelberatung.c.beratung_id
)
res = conn.execute(stmt, rows_for_db_insert)
new_cons_session_ids = cast(list[int], [row[0] for row in res.fetchall()])
assert len(cons_sessions_inserted) == len(new_cons_session_ids)
for cons_session, new_id in zip(cons_sessions_inserted, new_cons_session_ids):
cons_session.beratung_id = new_id
return consultation_data
@wrap_result(11)
def page_consulting_from_db(
cons_id: ConsId,
) -> Beratungsgespraech_Vorgang:
# TODO add Routine
# get "Vorgang" and all associated sessions to instantiate Pydantic data model
...
logger.debug("[Consulting Page] Call database reading routine...")
with db.ENGINE.connect() as conn:
# get the consultation process
stmt = db.beratung_vorgang.select().where(db.beratung_vorgang.c.vorgang_id == cons_id)
ret = conn.execute(stmt)
results = ret.mappings().all()
if not results:
raise KeyError(f"Database ID {cons_id} not found")
assert len(results) == 1, "more than one consulting process obtained"
consultation_data_db = results[0]
consultation_data = Beratungsgespraech_Vorgang(
vorgang_id=consultation_data_db["vorgang_id"],
un_id=consultation_data_db["un_id"],
pers_id=consultation_data_db["pers_id"],
titel=consultation_data_db["titel"],
beratungs_typ=consultation_data_db["beratungs_typ"],
erstellt=consultation_data_db["erstellt"],
aktualisiert=consultation_data_db["aktualisiert"],
beratungen=[],
)
# get all consultation sessions of this process
stmt = db.beratung_einzelberatung.select().where(
db.beratung_einzelberatung.c.vorgang_id == cons_id
)
ret = conn.execute(stmt)
# empty results possible
if ret.rowcount == 0:
logger.debug("[Consulting Page] No sessions, return directly...")
return consultation_data
cons_sessions = ret.mappings()
for session in cons_sessions:
cons_session_pydantic = Beratungsgespraech_Einzelgespraech(
beratung_id=session["beratung_id"],
vorgang_id=session["vorgang_id"],
nutzer_id=session["nutzer_id"],
nutzer_name=session["nutzer_name"],
zeitstempel=session["zeitstempel"],
ansprechpartner=session["ansprechpartner"],
kommunikationsweg=session["kommunikationsweg"],
thema_crm_matrix=session["thema_crm_matrix"],
anmerkungen=session["anmerkungen"],
rueckmeldung=session["rueckmeldung"],
erstellt=session["erstellt"],
aktualisiert=session["aktualisiert"],
)
consultation_data.beratungen.append(cons_session_pydantic)
logger.debug("[Consulting Page] Returning with sessions attached...")
return consultation_data
# // main page interaction
src/wce_crm/gui.py
+203 -78
View File
@@ -25,6 +25,7 @@ from typing import (
)
from typing_extensions import override
from dopt_basics.result_pattern import STATUS_HANDLER
from pydantic import (
ValidationError,
)
@@ -1625,7 +1626,7 @@ class AutoForm(QWidget):
logger_auto_form.error(
"[Auto-Form] Error during GUI validation phase:\n%s", pformat(errors)
)
GUI_validation_error_handling(errors)
GUI_pydantic_validation_error_handling(errors)
self._enable_save()
return
@@ -2629,6 +2630,7 @@ class Page_NewInitRec(QWidget):
back_requested = Signal() # Signal back button
company_requested = Signal(Page_InitRecCompany_State) # Signal "Unternehmen"
person_requested = Signal(Page_InitRecPerson_State) # Signal "Individualperson"
consulting_requested = Signal(Page_Consulting_State) # Signal "Beratungsgespräch"
def __init__(
self,
@@ -2661,18 +2663,25 @@ class Page_NewInitRec(QWidget):
btn_person.setFixedHeight(40)
btn_person.clicked.connect(self._request_initrec_person)
btn_consulting = QPushButton("Beratungsgespräch →")
btn_consulting.setFixedWidth(300)
btn_consulting.setFixedHeight(40)
btn_consulting.clicked.connect(self._request_consulting)
layout.addWidget(back_btn)
layout.addSpacing(15)
layout.addWidget(self.title_label)
layout.addWidget(btn_company)
layout.addWidget(btn_person)
layout.addSpacing(10)
layout.addWidget(btn_consulting)
def _request_initrec_company(self) -> None:
req_state = Page_InitRecCompany_State(
session=self.STATE.session,
un_id=None,
)
logger_gui.debug("[Page -- InitRec Company] State to call: %s", req_state)
logger_gui.debug("[Page -- InitRec] State to call: %s", req_state)
self.company_requested.emit(req_state)
def _request_initrec_person(self) -> None:
@@ -2680,9 +2689,18 @@ class Page_NewInitRec(QWidget):
session=self.STATE.session,
pers_id=None,
)
logger_gui.debug("[Page -- InitRec Company] State to call: %s", req_state)
logger_gui.debug("[Page -- InitRec] State to call: %s", req_state)
self.person_requested.emit(req_state)
def _request_consulting(self) -> None:
req_state = Page_Consulting_State(
session=self.STATE.session,
vorgang_id=None,
beratungs_typ=ConsultingType.PAUSCHAL,
)
logger_gui.debug("[Page -- InitRec] State to call: %s", req_state)
self.consulting_requested.emit(req_state)
def _sync_state_to_GUI(self) -> None:
pass
@@ -3628,12 +3646,25 @@ class Page_Consulting(QWidget):
btn_get_data_table = QPushButton("Debug call 'get_state' from table and page")
btn_get_data_table.clicked.connect(self._debug_get_state)
container_layout.addWidget(btn_get_data_table)
btn_validate_data_table = QPushButton(
"Debuig call 'validate' from table and page"
)
btn_validate_data_table = QPushButton("Debug call 'validate' from table and page")
btn_validate_data_table.clicked.connect(self._debug_validate)
container_layout.addWidget(btn_validate_data_table)
db_index_layout = QHBoxLayout()
self.db_index_field = QLineEdit(placeholderText="setze Index für VorgangID")
self.db_index_field.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred
)
db_index_set_button = QPushButton("setze und lade Vorgang-ID (Datenbankindex)")
db_index_set_button.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Preferred
)
db_index_set_button.clicked.connect(self._debug_load_database)
db_index_layout.addWidget(self.db_index_field)
db_index_layout.addWidget(db_index_set_button)
container_layout.addLayout(db_index_layout)
separator2 = QFrame()
separator2.setFrameShape(QFrame.Shape.HLine)
separator2.setFrameShadow(QFrame.Shadow.Sunken)
@@ -3762,6 +3793,18 @@ class Page_Consulting(QWidget):
errors = self.validate()
logger_page_consulting.debug("[Consulting Page] Page errors:\n%s", pformat(errors))
def _debug_load_database(self) -> None:
value = self.db_index_field.text().strip()
if not value:
return
try:
db_index = int(value)
except ValueError:
return
self.load_database(db_index)
def _sync_state_to_GUI(self) -> None:
match self.STATE.beratungs_typ:
case ConsultingType.PAUSCHAL:
@@ -3774,9 +3817,18 @@ class Page_Consulting(QWidget):
if self.STATE.un_id or self.STATE.pers_id:
self.type_pauschal_btn.setEnabled(False)
self.type_individual_btn.setEnabled(False)
else:
self.type_pauschal_btn.setEnabled(True)
self.type_individual_btn.setEnabled(True)
# child modules
# call to child modules' syncing method not needed because these are initialised by
# their own "load_state" method which again calls the modules sync method
table_state = Page_Consulting_Table_State(
session=self.STATE.session,
row_states=self.STATE.cons_sessions,
)
self.table_sessions.load_state(table_state)
def _sync_GUI_to_state(self) -> None:
# call to child modules' syncing method not needed because their syncing method is
@@ -3798,12 +3850,74 @@ class Page_Consulting(QWidget):
) -> None:
logger_page_consulting.debug("[Consulting] Loading with request:\n%s", new_state)
set_page_state(self.STATE, new_state)
# child modules
table_state = Page_Consulting_Table_State(
session=self.STATE.session,
row_states=self.STATE.cons_sessions,
)
self.table_sessions.load_state(table_state)
self._sync_state_to_GUI()
def load_database(
self,
cons_id: ConsId,
) -> None:
# TODO add routine
# load data model from database
res = backend.page_consulting_from_db(cons_id)
if res.status != STATUS_HANDLER.SUCCESS:
logger_page_consulting.error(
(
"[Consulting Page] There were errors during the database "
"loading procedure."
),
stack_info=True,
)
exc_formatted = (
f"Exception: {res.status.ExceptionType}\nMessage: {res.status.message}"
)
msg_box = get_message_box(
QMessageBox.Icon.Warning,
"Laden fehlgeschlagen",
(
"Beim Laden der Daten ist ein Fehler aufgetreten. Details sind "
"unten angefügt."
),
detailed_text=exc_formatted,
)
msg_box.exec()
return
# instantiate state from data model
consulting_process = res.unwrap()
self._load_from_pydantic(consulting_process)
def _load_from_pydantic(
self,
data: Beratungsgespraech_Vorgang,
update_only: bool = False,
) -> None:
self.STATE.vorgang_id = data.vorgang_id
self.STATE.beratungs_typ = ConsultingType(data.beratungs_typ)
self.STATE.un_id = data.un_id
self.STATE.pers_id = data.pers_id
self.STATE.titel = data.titel
if update_only:
for gui_state, pydantic_state in zip(self.STATE.cons_sessions, data.beratungen):
gui_state.beratung_id = pydantic_state.beratung_id
else:
self.STATE.cons_sessions.clear()
for pydantic_state in data.beratungen:
cons_session_state = Page_Consulting_ConsultingSession_State(
session=self.STATE.session,
nutzer_id=pydantic_state.nutzer_id,
nutzer_name=pydantic_state.nutzer_name,
beratung_id=pydantic_state.beratung_id,
zeitstempel=pydantic_state.zeitstempel,
ansprechpartner=pydantic_state.ansprechpartner,
kommunikationsweg=pydantic_state.kommunikationsweg,
thema_crm_matrix=pydantic_state.thema_crm_matrix,
anmerkungen=pydantic_state.anmerkungen,
rueckmeldung=pydantic_state.rueckmeldung,
)
self.STATE.cons_sessions.append(cons_session_state)
self._sync_state_to_GUI()
@@ -3821,7 +3935,7 @@ class Page_Consulting(QWidget):
logger_page_consulting.error(
"[Consulting Page] Error during GUI validation phase:\n%s", pformat(errors)
)
GUI_validation_error_handling(errors)
GUI_pydantic_validation_error_handling(errors)
self._enable_save()
return
@@ -3862,48 +3976,54 @@ class Page_Consulting(QWidget):
else:
if DEBUG_NO_DATABASE:
return
# logger_page_consulting.debug(
# "[Consulting Page] Validated Pydantic successfully. Model:\n%s",
# pformat(consulting_process.model_dump()),
# )
logger_page_consulting.debug("[Consulting Page] Validated Pydantic successfully.")
# !! should be placed in a try block to handle errors
backend.page_consulting_to_db(consulting_process)
res = backend.page_consulting_to_db(consulting_process)
if res.status != STATUS_HANDLER.SUCCESS:
logger_page_consulting.error(
(
"[Consulting Page] There were errors during the database "
"saving/updating procedure."
),
stack_info=True,
)
# # !! this code is only called if the 'try' block was successful
# # save data to database
# # db_data = validated_data.to_db(exclude=self.cfg.ignored_keys)
# # logger_auto_form.debug(
# # "[Consulting Page] Form data with 'exlude' (must be saved in the database):\n%s",
# # pformat(db_data),
# # )
exc_formatted = (
f"Exception: {res.status.ExceptionType}\nMessage: {res.status.message}"
)
msg_box = get_message_box(
QMessageBox.Icon.Warning,
"Speichern fehlgeschlagen",
(
"Beim Speichern der Daten ist ein Fehler aufgetreten. Details sind "
"unten angefügt."
),
detailed_text=exc_formatted,
)
msg_box.exec()
return
# if self.STATE.rec_id is None:
# logger_auto_form.debug("[Consulting Page] Insert triggered...")
# # should return "vorgang_id"
# # init_rec_id = self.cfg.data_insert(db_data)
# # assert isinstance(init_rec_id, int)
# # self.STATE.rec_id = init_rec_id
# else:
# logger_auto_form.debug("[Consulting Page] Update triggered...")
# # self.cfg.data_update(self.STATE.rec_id, db_data)
updated_data = res.unwrap()
logger_page_consulting.info(
"[Consulting Page] Load state from modified Pydantic model"
)
self._load_from_pydantic(updated_data, update_only=True)
# logger_auto_form.info("[Consulting Page] Data saved successfully")
# self.update_triggered.emit()
# self._activate_delete()
logger_page_consulting.info("[Consulting Page] Data saved successfully")
# TODO status message successful to user in GUI
msg_box = get_message_box(
QMessageBox.Icon.Information,
"Speichern erfolgreich",
"Die Daten wurden erfolgreich gespeichert",
)
msg_box.exec()
# self.update_triggered.emit()
# self._activate_delete()
finally:
# always re-enable save, even if error occurred
self._enable_save()
def load_database(
self,
cons_id: ConsId,
) -> None:
# TODO add routine
# load data model from database
consulting_process = backend.page_consulting_from_db(cons_id)
# instantiate state from data model
...
class Page_Consulting_ConsultingSession(QWidget):
"""child of 'Page_Consulting_DynamicTable'"""
@@ -4095,32 +4215,6 @@ class Page_Consulting_ConsultingSession(QWidget):
self._sync_state_to_GUI()
def set_page_state(
target: S,
source: S,
deep: bool = False,
) -> None:
"""
transfer all field values generically from 'source' to 'target'
"""
if not (dc.is_dataclass(target) and dc.is_dataclass(source)):
raise TypeError("Both source and atrget must be dataclasses")
if type(target) is not type(source):
raise TypeError("Both source and target must have the same type")
for field in dc.fields(source):
if field.name == "child_modules":
continue
new_value = getattr(source, field.name)
if deep:
new_value = copy.deepcopy(new_value)
setattr(target, field.name, new_value)
class Page_Consulting_Table(QWidget):
"""child of 'Page_Consulting'"""
@@ -4314,10 +4408,14 @@ class Page_Consulting_Table(QWidget):
self.remove_row(self.STATE.child_modules[0])
def _sync_state_to_GUI(self) -> None:
self.setUpdatesEnabled(False)
self.reset()
for row_state in self.STATE.row_states:
self.add_row(row_state)
self.setUpdatesEnabled(True)
def _sync_GUI_to_state(self) -> None:
logger_page_consulting.debug("[Page -- Consulting] Table: Call syncing GUI to state")
row_data: list[Page_Consulting_ConsultingSession_State] = []
@@ -4367,6 +4465,32 @@ class Page_Consulting_Table(QWidget):
####################################################################
def set_page_state(
target: S,
source: S,
deep: bool = False,
) -> None:
"""
transfer all field values generically from 'source' to 'target'
"""
if not (dc.is_dataclass(target) and dc.is_dataclass(source)):
raise TypeError("Both source and atrget must be dataclasses")
if type(target) is not type(source):
raise TypeError("Both source and target must have the same type")
for field in dc.fields(source):
if field.name == "child_modules":
continue
new_value = getattr(source, field.name)
if deep:
new_value = copy.deepcopy(new_value)
setattr(target, field.name, new_value)
def clear_layout(
layout: QLayout | None,
) -> None:
@@ -4416,6 +4540,7 @@ class MainWindow(QMainWindow):
self.new_initrec_select.back_requested.connect(self.show_main_page)
self.new_initrec_select.company_requested.connect(self.show_page_initrec_company)
self.new_initrec_select.person_requested.connect(self.show_page_initrec_person)
self.new_initrec_select.consulting_requested.connect(self.show_page_consulting)
self.stack.addWidget(self.new_initrec_select)
# SITE: 'Grunderfassung Unternehmen'
initrec_company_state = Page_InitRecCompany_State(session=self.STATE.session)
@@ -4666,7 +4791,7 @@ def global_exception_handler(exc_type, exc_value, exc_traceback):
# message to user
# check if QApplication exists otherwise crashing of crash handler possible)
if QApplication.instance():
msg_box = get_error_message_box(
msg_box = get_message_box(
QMessageBox.Icon.Critical,
"Kritischer Fehler",
"Ein unerwarteter Fehler ist aufgetreten. Die Details wurden protokolliert.",
@@ -4684,7 +4809,7 @@ def global_exception_handler(exc_type, exc_value, exc_traceback):
sys.exit(1)
def get_error_message_box(
def get_message_box(
msg_type: QMessageBox.Icon,
title: str,
message: str,
@@ -4713,7 +4838,7 @@ def pydantic_validation_error_handling(
error_texts.append(f"- {error_field}: {reason}, (Pfad: {path})")
msg_box = get_error_message_box(
msg_box = get_message_box(
QMessageBox.Icon.Warning,
"Fehler bei der Validierung der Eingabedaten",
(
@@ -4725,13 +4850,13 @@ def pydantic_validation_error_handling(
msg_box.exec()
def GUI_validation_error_handling(
def GUI_pydantic_validation_error_handling(
errors: list[str],
) -> None:
error_text = "Bitte füllen Sie die folgenden Pflichtfelder aus:\n\n▸ " + "\n▸ ".join(
errors
)
msg_box = get_error_message_box(
msg_box = get_message_box(
QMessageBox.Icon.Warning,
"Fehlende oder fehlerhafte Angaben",
error_text,