From a40213a1ed4d1f4cbafafd1ef0f9eb947635c0c4 Mon Sep 17 00:00:00 2001 From: foefl Date: Wed, 15 Jul 2026 17:28:24 +0200 Subject: [PATCH] prepare new loading scheme for auto form data (not working yet) --- src/wce_crm/backend/backend.py | 238 +++++++++++++++++++++++++++++++++ src/wce_crm/data_models.py | 30 +++++ src/wce_crm/gui.py | 67 +++++++--- 3 files changed, 316 insertions(+), 19 deletions(-) diff --git a/src/wce_crm/backend/backend.py b/src/wce_crm/backend/backend.py index c3f701d..eeb8b30 100644 --- a/src/wce_crm/backend/backend.py +++ b/src/wce_crm/backend/backend.py @@ -1,5 +1,6 @@ from __future__ import annotations +import copy import datetime from pprint import pformat from typing import TYPE_CHECKING, Any, cast @@ -13,6 +14,7 @@ from wce_crm.constants import TIMEZONE_CEST from wce_crm.data_models import ( Beratungsgespraech_Einzelgespraech, Beratungsgespraech_Vorgang, + Initrec_FromDb, ) from wce_crm.logging import logger_back as logger from wce_crm.types import ( @@ -26,6 +28,7 @@ from wce_crm.types import ( ) if TYPE_CHECKING: + from wce_crm.data_models import AutoForm_State, Initrec from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId @@ -145,6 +148,241 @@ def initrec_company_update_initial_recording( conn.execute(stmt, data) +def initrec_company_to_db( + auto_form_data: Initrec, +) -> Initrec: + logger.debug("[AutoForm -- backend] Call database saving routine...") + + dump_data = copy.deepcopy(auto_form_data.form_data) + dump_data["geloescht"] = auto_form_data.geloescht + + with db.ENGINE.begin() as conn: + if auto_form_data.rec_id is None: + logger.debug("[AutoForm -- backend] Insert...") + + stmt = db.grunderfassung_unternehmen.insert().returning( + db.grunderfassung_unternehmen.c.un_id, + db.grunderfassung_unternehmen.c.Metadaten_aktualisierung, + db.grunderfassung_unternehmen.c.geloescht, + ) + ret = conn.execute(stmt, dump_data) + if ret.rowcount == 0: + raise IOError("Entry was not inserted correctly") + + from_db = ret.mappings().fetchall() + assert len(from_db) == 1, "expected excactly one returned row" + data_from_db = from_db[0] + auto_form_data.rec_id = data_from_db["un_id"] + auto_form_data.geloescht = data_from_db["geloescht"] + auto_form_data.form_data["Metadaten_aktualisierung"] = data_from_db[ + "Metadaten_aktualisierung" + ] + + logger.debug("[AutoForm -- backend] Inserted InitRec successfully") + + else: + logger.debug("[AutoForm -- backend] Update...") + stmt = ( + db.beratung_vorgang.update() + .where(db.grunderfassung_unternehmen.c.un_id == auto_form_data.rec_id) + .returning( + db.grunderfassung_unternehmen.c.Metadaten_aktualisierung, + db.grunderfassung_unternehmen.c.geloescht, + ) + ) + ret = conn.execute(stmt, dump_data) + + from_db = ret.mappings().fetchall() + assert len(from_db) == 1, "expected excatly one returned row" + data_from_db = from_db[0] + + auto_form_data.geloescht = data_from_db["geloescht"] + auto_form_data.form_data["Metadaten_aktualisierung"] = data_from_db[ + "Metadaten_aktualisierung" + ] + + logger.debug( + "[AutoForm -- backend] Updated InitRec with ID %d successfully", + auto_form_data.rec_id, + ) + + return auto_form_data + + +def initrec_company_from_db( + id_: RecId, +) -> Initrec: + logger.debug("[AutoForm -- backend] Call database loading routine...") + stmt = db.grunderfassung_unternehmen.select().where( + db.grunderfassung_unternehmen.c.un_id == id_ + ) + with db.ENGINE.connect() as conn: + ret = conn.execute(stmt) + + from_db = ret.mappings().all() + assert len(from_db) == 1, "not excactly one company initial recording obtained" + data_from_db = dict(from_db[0]) + + geloescht = data_from_db["geloescht"] + del data_from_db["geloescht"] + + return Initrec( + rec_id=id_, + geloescht=geloescht, + form_data=data_from_db, + ) + + +def initrec_person_to_db( + auto_form_data: Initrec, +) -> Initrec: + logger.debug("[AutoForm -- backend] Call database saving routine...") + + dump_data = copy.deepcopy(auto_form_data.form_data) + dump_data["geloescht"] = auto_form_data.geloescht + + with db.ENGINE.begin() as conn: + if auto_form_data.rec_id is None: + logger.debug("[AutoForm -- backend] Insert...") + + stmt = db.grunderfassung_personen.insert().returning( + db.grunderfassung_personen.c.pers_id, + db.grunderfassung_personen.c.Metadaten_aktualisierung, + db.grunderfassung_personen.c.geloescht, + ) + ret = conn.execute(stmt, dump_data) + if ret.rowcount == 0: + raise IOError("Entry was not inserted correctly") + + from_db = ret.mappings().fetchall() + assert len(from_db) == 1, "expected excactly one returned row" + data_from_db = from_db[0] + auto_form_data.rec_id = data_from_db["un_id"] + auto_form_data.geloescht = data_from_db["geloescht"] + auto_form_data.form_data["Metadaten_aktualisierung"] = data_from_db[ + "Metadaten_aktualisierung" + ] + + logger.debug("[AutoForm -- backend] Inserted InitRec successfully") + + else: + logger.debug("[AutoForm -- backend] Update...") + stmt = ( + db.beratung_vorgang.update() + .where(db.grunderfassung_personen.c.pers_id == auto_form_data.rec_id) + .returning( + db.grunderfassung_personen.c.Metadaten_aktualisierung, + db.grunderfassung_personen.c.geloescht, + ) + ) + ret = conn.execute(stmt, dump_data) + + from_db = ret.mappings().fetchall() + assert len(from_db) == 1, "expected excatly one returned row" + data_from_db = from_db[0] + + auto_form_data.geloescht = data_from_db["geloescht"] + auto_form_data.form_data["Metadaten_aktualisierung"] = data_from_db[ + "Metadaten_aktualisierung" + ] + + logger.debug( + "[AutoForm -- backend] Updated InitRec with ID %d successfully", + auto_form_data.rec_id, + ) + + return auto_form_data + + +def initrec_person_from_db( + id_: RecId, +) -> Initrec: + logger.debug("[AutoForm -- backend] Call database loading routine...") + stmt = db.grunderfassung_personen.select().where( + db.grunderfassung_personen.c.pers_id == id_ + ) + with db.ENGINE.connect() as conn: + ret = conn.execute(stmt) + + from_db = ret.mappings().all() + assert len(from_db) == 1, "not excactly one company initial recording obtained" + data_from_db = dict(from_db[0]) + + geloescht = data_from_db["geloescht"] + del data_from_db["geloescht"] + + return Initrec( + rec_id=id_, + geloescht=geloescht, + form_data=data_from_db, + ) + + +# def initrec_person_to_db( +# auto_form_data: Initrec, +# ) -> Initrec_FromDb: +# logger.debug("[AutoForm -- backend] Call database saving routine...") + +# dump_data = copy.deepcopy(auto_form_data.form_data) +# dump_data["geloescht"] = auto_form_data.geloescht + +# return_from_db: Initrec_FromDb + +# with db.ENGINE.begin() as conn: +# if auto_form_data.rec_id is None: +# # insert new "Vorgang" +# logger.debug("[AutoForm -- backend] Insert...") + +# stmt = db.grunderfassung_personen.insert().returning( +# db.grunderfassung_personen.c.pers_id, +# db.grunderfassung_personen.c.Metadaten_aktualisierung, +# db.grunderfassung_personen.c.geloescht, +# ) +# ret = conn.execute(stmt, dump_data) +# if ret.rowcount == 0: +# raise IOError("Entry was not inserted correctly") + +# from_db = ret.mappings().fetchall() +# assert len(from_db) == 1, "expected excactly one returned row" +# data_from_db = from_db[0] +# return_from_db = Initrec_FromDb( +# rec_id=data_from_db["un_id"], +# geloescht=data_from_db["geloescht"], +# Metadaten_aktualisierung=data_from_db["Metadaten_aktualisierung"], +# ) + +# logger.debug("[AutoForm -- backend] Inserted InitRec successfully") + +# else: +# logger.debug("[AutoForm -- backend] Update...") +# stmt = ( +# db.beratung_vorgang.update() +# .where(db.grunderfassung_personen.c.pers_id == auto_form_data.rec_id) +# .returning( +# db.grunderfassung_personen.c.Metadaten_aktualisierung, +# db.grunderfassung_personen.c.geloescht, +# ) +# ) +# ret = conn.execute(stmt, dump_data) + +# from_db = ret.mappings().fetchall() +# assert len(from_db) == 1, "expected excatly one returned row" +# data_from_db = from_db[0] + +# return_from_db = Initrec_FromDb( +# rec_id=auto_form_data.rec_id, +# geloescht=data_from_db["geloescht"], +# Metadaten_aktualisierung=data_from_db["Metadaten_aktualisierung"], +# ) + +# logger.debug( +# "[AutoForm -- backend] Updated InitRec with ID %d successfully", +# auto_form_data.rec_id, +# ) + +# return return_from_db + + def initrec_company_get_initial_recording( id_: RecId, ) -> dict[str, Any]: diff --git a/src/wce_crm/data_models.py b/src/wce_crm/data_models.py index f4e2dce..2894cf8 100644 --- a/src/wce_crm/data_models.py +++ b/src/wce_crm/data_models.py @@ -177,6 +177,20 @@ class AutoFormDelete(Protocol): ) -> None: ... +class AutoFormToDb(Protocol): + def __call__( + self, + auto_form_data: Initrec, + ) -> Initrec: ... + + +class AutoFormFromDb(Protocol): + def __call__( + self, + id_: int, + ) -> Initrec: ... + + @dc.dataclass(slots=True) class Session: user_id: UserId @@ -209,6 +223,8 @@ class Page_InitRecCompany_State(PageState[Module]): @dc.dataclass(slots=True, kw_only=True) class AutoFormConfig: model: type[FlatBaseModel] + to_db: AutoFormToDb + from_db: AutoFormFromDb data_insert: AutoFormInsert data_update: AutoFormUpdate data_get: AutoFormGet @@ -225,6 +241,7 @@ class AutoForm_State(PageState[Module]): rec_id: RecId | None = None form_data: dict[str, Any] | None = None locked: bool + geloescht: bool = False @dc.dataclass(slots=True, kw_only=True) @@ -395,6 +412,19 @@ class FlatBaseModel(BaseModel): # ** InitRec +class Initrec_FromDb(BaseModel): + rec_id: RecId + geloescht: AwareDatetime | None + Metadaten_aktualisierung: AwareDatetime + form_data: dict[str, Any] + + +class Initrec(BaseModel): + rec_id: RecId | None + geloescht: AwareDatetime | None = None + form_data: dict[str, Any] + + class Grunderfassung_Unternehmen(FlatBaseModel): # default in SQLAlchemy with lambda and timezone-aware datetime Metadaten_erstellung: AwareDatetime | None = None diff --git a/src/wce_crm/gui.py b/src/wce_crm/gui.py index 37bb5d8..af92cce 100644 --- a/src/wce_crm/gui.py +++ b/src/wce_crm/gui.py @@ -86,6 +86,7 @@ from wce_crm.data_models import ( Beratungsgespraech_Vorgang, Grunderfassung_Personen, Grunderfassung_Unternehmen, + Initrec, Page_CompanyProfile_State, Page_Consulting_ConsultingSession_State, Page_Consulting_State, @@ -111,7 +112,7 @@ from wce_crm.logging import ( logger_gui, logger_page_consulting, ) -from wce_crm.types import CompanyProfileConsultationEntry, ConsultingType +from wce_crm.types import CompanyProfileConsultationEntry, ConsultingType, RecId if TYPE_CHECKING: from wce_crm.data_models import AutoForm_State, Module, WrapperModule @@ -1417,7 +1418,7 @@ class AutoForm(QWidget): self.main_layout.addWidget(self.registry_button) self.test_button = QPushButton("Initialisiere Laden") - self.test_button.clicked.connect(self._load_data) + self.test_button.clicked.connect(self._load_from_id) self.test_button.setFixedHeight(35) self.test_button.setSizePolicy( QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed @@ -1535,9 +1536,9 @@ class AutoForm(QWidget): self.update_triggered.emit() self.reset_form() - def _load_data( + def _load_from_id( self, - lookup_id: int | None, + id_: RecId | None, ) -> None: if DEBUG_NO_DATABASE: return @@ -1545,28 +1546,33 @@ class AutoForm(QWidget): logger_auto_form.info("[Auto-Form] Load data method...") self.reset_form() - logger_auto_form.debug("[Auto-Form] Lookup ID: %s", lookup_id) + logger_auto_form.debug("[Auto-Form] Lookup ID: %s", id_) - if lookup_id is not None: + if id_ is not None: logger_auto_form.debug("Load from DB:") - loaded_data = self.cfg.data_get(lookup_id) + returned = self.cfg.from_db(id_) + # loaded_data = self.cfg.data_get(lookup_id) # TODO add path to avoid unnecessary loading (where the same ID is loaded # twice in a row) -- use an internal marker (not state-related) else: logger_auto_form.debug("[Auto-Form] Loading: Lookup ID NULL. Do nothing!") return + # logger_auto_form.debug( + # "[Auto-Form] Loaded data dict:\n%s Passing to Pydantic...", pformat(loaded_data) + # ) logger_auto_form.debug( - "[Auto-Form] Loaded data dict:\n%s Passing to Pydantic...", pformat(loaded_data) + "[Auto-Form] Loaded data dict:\n%s Passing to Pydantic...", + pformat(returned.form_data), ) - model = self.cfg.model(**loaded_data) + model = self.cfg.model(**returned.form_data) logger_auto_form.debug("[Auto-Form] Loaded to Pydantic.") logger_auto_form.debug("[Auto-Form] Convert to GUI structure...") form_data = model.to_gui() logger_auto_form.debug("[Auto-Form] Set form data...") logger_auto_form.debug("[Auto-Form] Form data:\n%s", pformat(form_data)) self._set_form_data(form_data) - self.STATE.rec_id = lookup_id + self.STATE.rec_id = id_ self._activate_delete() def reset_form(self) -> None: @@ -1608,7 +1614,7 @@ class AutoForm(QWidget): section: Literal["form", "locking"] | None, ) -> None: if section is None or section == "form": - self._load_data(self.STATE.rec_id) + self._load_from_id(self.STATE.rec_id) if section is None or section == "locking": if self.STATE.locked: @@ -1641,6 +1647,13 @@ class AutoForm(QWidget): return self.STATE + def _load_from_pydantic(self, data: Initrec) -> None: + self.STATE.rec_id = data.rec_id + self.STATE.form_data = data.form_data + self.STATE.geloescht = True if data.geloescht else False + + self._sync_state_to_GUI(None) # TODO change to correct loading + def save_data(self) -> None: self._disable_save() errors = self.validate() @@ -1679,15 +1692,27 @@ class AutoForm(QWidget): "[Auto-Form] Form data with 'exlude' (must be saved in the database):\n%s", pformat(db_data), ) + deleted_datetime: datetime.datetime | None = None + if self.STATE.geloescht: + deleted_datetime = datetime.datetime.now(datetime.UTC) try: - if self.STATE.rec_id is None: - logger_auto_form.debug("[Auto-Form] Insert triggered...") - 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("[Auto-Form] Update triggered...") - self.cfg.data_update(self.STATE.rec_id, db_data) + to_db = Initrec( + rec_id=self.STATE.rec_id, + geloescht=deleted_datetime, + form_data=db_data, + ) + returned = self.cfg.to_db(to_db) + + self._load_from_pydantic(returned) + + # if self.STATE.rec_id is None: + # logger_auto_form.debug("[Auto-Form] Insert triggered...") + # 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("[Auto-Form] Update triggered...") + # self.cfg.data_update(self.STATE.rec_id, db_data) self.update_triggered.emit() self._activate_delete() @@ -2876,6 +2901,8 @@ class Page_NewInitRec(QWidget): CONFIG_GRUNDERFASSUNG_UNTERNEHMEN: Final[AutoFormConfig] = AutoFormConfig( model=Grunderfassung_Unternehmen, + to_db=backend.initrec_company_to_db, + from_db=backend.initrec_company_from_db, data_insert=backend.initrec_company_insert_initial_recording, data_update=backend.initrec_company_update_initial_recording, data_get=backend.initrec_company_get_initial_recording, @@ -2889,6 +2916,8 @@ CONFIG_GRUNDERFASSUNG_UNTERNEHMEN: Final[AutoFormConfig] = AutoFormConfig( CONFIG_GRUNDERFASSUNG_PERSONEN: Final[AutoFormConfig] = AutoFormConfig( model=Grunderfassung_Personen, + to_db=backend.initrec_person_to_db, + from_db=backend.initrec_person_from_db, data_insert=backend.initrec_person_insert_initial_recording, data_update=backend.initrec_person_update_initial_recording, data_get=backend.initrec_person_get_initial_recording,