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

Dev #7

Merged
foefl merged 74 commits from dev into main 2026-07-31 14:19:01 +00:00
Conversation 0 Commits 74 Files Changed 38
+151 -22
3 changed files with 151 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 482ae436eb - Show all commits
src/wce_crm/backend/backend.py
+45 -3
View File
@@ -564,6 +564,48 @@ def page_consulting_to_db(
return consultation_data
@wrap_result(10)
def page_consulting_delete(
cons_id: ConsId,
) -> None:
timestamp_deleted = datetime.datetime.now(datetime.UTC)
dump_data: dict[str, Any] = {"geloescht": timestamp_deleted}
with db.ENGINE.begin() as conn:
stmt = db.beratung_vorgang.update().where(db.beratung_vorgang.c.vorgang_id == cons_id)
conn.execute(stmt, dump_data)
stmt = db.beratung_einzelberatung.update().where(
db.beratung_einzelberatung.c.vorgang_id == cons_id,
db.beratung_einzelberatung.c.geloescht.is_(None),
)
conn.execute(stmt, dump_data)
return
stmt = db.beratung_einzelberatung.select().where(
db.beratung_einzelberatung.c.vorgang_id == cons_id,
db.beratung_einzelberatung.c.geloescht.is_(None),
)
ret = conn.execute(stmt)
cons_sessions = ret.mappings().all()
# empty results possible
if not cons_sessions:
logger.debug("[Consulting Page] No sessions, return directly...")
return
rows_for_db_update: list[dict[str, Any]] = []
for row_data in cons_sessions:
row_data_dump = copy.deepcopy(dump_data)
row_data_dump["b_beratung_id"] = row_data["beratung_id"]
rows_for_db_update.append(row_data_dump)
if 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)
@wrap_result(11)
def page_consulting_from_db(
cons_id: ConsId,
@@ -611,12 +653,12 @@ def page_consulting_from_db(
db.beratung_einzelberatung.c.geloescht.is_(None),
)
ret = conn.execute(stmt)
cons_sessions = ret.mappings().all()
# empty results possible
if ret.rowcount == 0:
if not cons_sessions:
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"],
@@ -649,7 +691,7 @@ def companyprofile_page_get_consultations(
db.beratung_vorgang.c.aktualisiert,
db.beratung_vorgang.c.titel,
db.beratung_vorgang.c.beratungs_typ,
).where(db.beratung_vorgang.c.un_id == un_id)
).where(db.beratung_vorgang.c.un_id == un_id, db.beratung_vorgang.c.geloescht.is_(None))
with db.ENGINE.connect() as conn:
res = conn.execute(stmt)
src/wce_crm/data_models.py
+1 -1
View File
@@ -262,7 +262,7 @@ class Page_Consulting_State(PageState[Module]):
cons_sessions: list[Page_Consulting_ConsultingSession_State] = dc.field(
default_factory=list
)
ist_geloescht: bool = False
geloescht: bool = False
locked: bool
src/wce_crm/gui.py
+104 -17
View File
@@ -1590,6 +1590,7 @@ class AutoForm(QWidget):
reset_form(self.widget_registry)
self.STATE.rec_id = None
self.STATE.form_data = None
self.STATE.geloescht = False
def lock_form(self) -> None:
self.STATE.locked = True
@@ -1742,6 +1743,7 @@ class AutoForm(QWidget):
if self.STATE.geloescht:
deleted_datetime = datetime.datetime.now(datetime.UTC)
try:
# TODO change to result pattern
to_db = Initrec(
rec_id=self.STATE.rec_id,
geloescht=deleted_datetime,
@@ -3706,11 +3708,14 @@ class Page_CompanyProfile(QWidget):
widget.deleteLater()
def _update_consultations(self) -> None:
assert self.STATE.rec_id is not None, (
"cannot update consultations if RecID is not set"
)
if self.STATE.rec_id is None:
return
consultations = backend.companyprofile_page_get_consultations(un_id=self.STATE.rec_id)
logger_gui.debug(
"[Page -- Company Profile] Consultations: %s", pformat(consultations)
)
self._clear_consultations()
for entry in consultations.pauschal:
@@ -4179,13 +4184,26 @@ class Page_Consulting(QWidget):
def _delete_data(self) -> None:
logger_page_consulting.debug("[Consulting Page] Call data delete...")
assert self.STATE.vorgang_id is not None, "deletion initialised despite no index set"
confirm = get_user_confirmation(
self,
"Löschen bestätigen",
"Soll der Eintrag wirklich gelöscht werden?",
)
if not confirm:
return
self.STATE.geloescht = True
self._delete_entry()
self.update_triggered.emit()
def _disable_save(self) -> None:
self.save_btn.setEnabled(False)
self.save_btn.setText(self.edit_buttons.save_btn_txt_disabled)
def _activate_delete(self) -> None:
if self.STATE.un_id is not None or self.STATE.pers_id is not None:
# if self.STATE.un_id is not None or self.STATE.pers_id is not None:
if self.STATE.vorgang_id is not None:
self.delete_btn.setEnabled(True)
else:
self.delete_btn.setEnabled(False)
@@ -4241,7 +4259,7 @@ class Page_Consulting(QWidget):
except ValueError:
return
self.load_database(db_index)
self._load_database(db_index)
def _open_linking_dialogue(self) -> None:
logger_page_consulting.info("[Consulting Page] Linking dialogue triggered")
@@ -4271,10 +4289,21 @@ class Page_Consulting(QWidget):
self.info_banners.append(banner)
self.info_banners_layout.addWidget(banner)
def _reset(self) -> None:
self.STATE.vorgang_id = None
self.STATE.cons_sessions.clear()
self.STATE.geloescht = False
def _sync_state_to_GUI(
self,
section: Literal["cons_type", "banners", "table", "locking"] | None,
section: Literal["content", "cons_type", "banners", "table", "locking"] | None,
) -> None:
if section is None or section == "content":
if not self.STATE.geloescht:
self._load_database(self.STATE.vorgang_id)
else:
self._reset()
if section is None or section == "cons_type" or section == "locking":
match self.STATE.beratungs_typ:
case ConsultingType.PAUSCHAL:
@@ -4333,10 +4362,11 @@ class Page_Consulting(QWidget):
if section is None or section == "locking":
if self.STATE.locked:
self.table_sessions.lock()
self.edit_buttons.set_modify_state(False)
self.edit_buttons.set_modify_state(False, set_delete_btn=True)
else:
self.table_sessions.unlock()
self.edit_buttons.set_modify_state(True)
self.edit_buttons.set_modify_state(True, set_delete_btn=False)
self._activate_delete()
def _sync_GUI_to_state(self) -> None:
# call to child modules' syncing method not needed because their syncing method is
@@ -4361,12 +4391,19 @@ class Page_Consulting(QWidget):
self._sync_state_to_GUI(None)
def load_database(
def _load_database(
self,
cons_id: ConsId,
cons_id: ConsId | None,
) -> None:
# TODO add routine
# load data model from database
logger_auto_form.info("[Consulting Page] Load data method...")
logger_auto_form.debug("[Consulting Page] Lookup ID: %s", cons_id)
self._reset()
if cons_id is None:
logger_auto_form.debug("[Consulting Page] Loading: Lookup ID NULL. Do nothing!")
return
res = backend.page_consulting_from_db(cons_id)
if res.status != STATUS_HANDLER.SUCCESS:
logger_page_consulting.error(
@@ -4394,19 +4431,20 @@ class Page_Consulting(QWidget):
# instantiate state from data model
consulting_process = res.unwrap()
self._load_from_pydantic(consulting_process)
self._load_from_pydantic(consulting_process, sync_to_GUI=False)
def _load_from_pydantic(
self,
data: Beratungsgespraech_Vorgang,
update_only: bool = False,
sync_to_GUI: bool = True,
) -> 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
self.STATE.ist_geloescht = data.geloescht is not None
self.STATE.geloescht = data.geloescht is not None
self.STATE.locked = True
if update_only:
@@ -4438,6 +4476,7 @@ class Page_Consulting(QWidget):
)
self.STATE.cons_sessions.append(cons_session_state)
if sync_to_GUI:
self._sync_state_to_GUI(None)
def get_state(self) -> Page_Consulting_State:
@@ -4445,8 +4484,55 @@ class Page_Consulting(QWidget):
return self.STATE
def _delete_entry(self) -> None:
assert self.STATE.vorgang_id is not None and self.STATE.geloescht
logger_auto_form.debug(
"[Consulting Page] This was saved in the database "
"and is marked for deletion - fast path."
)
res = backend.page_consulting_delete(self.STATE.vorgang_id)
if res.status != STATUS_HANDLER.SUCCESS:
logger_page_consulting.error(
(
"[Consulting Page] There were errors during the database "
"deletion procedure."
),
stack_info=True,
)
exc_formatted = (
f"Exception: {res.status.ExceptionType}\nMessage: {res.status.message}"
)
msg_box = get_message_box(
QMessageBox.Icon.Warning,
"Löschen fehlgeschlagen",
(
"Beim Löschen der Daten ist ein Fehler aufgetreten. Details sind "
"unten angefügt."
),
detailed_text=exc_formatted,
)
msg_box.exec()
return
logger_page_consulting.info("[Consulting Page] Data deleted successfully")
msg_box = get_message_box(
QMessageBox.Icon.Information,
"Löschen erfolgreich",
"Die Daten wurden erfolgreich gelöscht",
)
msg_box.exec()
self._sync_state_to_GUI(None)
def save_data(self) -> None:
logger_page_consulting.debug("[Consulting Page] Call saving data...")
if self.STATE.vorgang_id is None and self.STATE.geloescht:
logger_auto_form.debug(
"[Consulting Page] This was never saved in the database "
"and is marked for deletion - ignore."
)
return
self._disable_save()
errors = self.validate()
@@ -4460,9 +4546,9 @@ class Page_Consulting(QWidget):
logger_page_consulting.info("[Consulting Page] Success! All data is valid")
logger_page_consulting.info("[Consulting Page] Get form data call...")
state = self.get_state()
self._sync_GUI_to_state()
logger_page_consulting.debug("[Consulting Page] Get state:\n%s", state)
logger_page_consulting.debug("[Consulting Page] Current state:\n%s", self.STATE)
try:
logger_page_consulting.debug("[Consulting Page] Call Pydantic...")
@@ -4470,7 +4556,7 @@ class Page_Consulting(QWidget):
pydantic_sessions: list[Beratungsgespraech_Einzelgespraech] = []
timestamp_deleted: datetime.datetime | None = None
if self.STATE.ist_geloescht:
if self.STATE.geloescht:
timestamp_deleted = datetime.datetime.now(datetime.UTC)
for row in self.STATE.cons_sessions:
@@ -4553,6 +4639,7 @@ class Page_Consulting(QWidget):
"[Consulting Page] Load state from modified Pydantic model"
)
self._load_from_pydantic(updated_consulting_process, update_only=True)
self.update_triggered.emit()
logger_page_consulting.info("[Consulting Page] Data saved successfully")
msg_box = get_message_box(
@@ -4561,7 +4648,7 @@ class Page_Consulting(QWidget):
"Die Daten wurden erfolgreich gespeichert",
)
msg_box.exec()
self.update_triggered.emit()
# self._activate_delete()
finally:
# always re-enable save, even if error occurred