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 functionality of loading/saving for initial recording with new data model

Browse Source
This commit is contained in:
Florian Förster
2026-07-27 11:41:24 +02:00
parent dc9374f5b7
commit 1ff55038fa
5 changed files with 207 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wce_crm/backend/backend.py
+69 -30
View File
@@ -402,11 +402,16 @@ def page_initrec_to_db(
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0]
data_from_db = dict(from_db[0])
data_from_db = _rename_db_data(
data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
data.ids.pers_id = data_from_db["pers_id"]
data.geloescht = data_from_db["geloescht"]
data.form_data.person["Metadaten_aktualisierung"] = data_from_db["aktualisiert"]
data.form_data.person["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully")
@@ -416,7 +421,7 @@ def page_initrec_to_db(
db.t_grunderfassung_personen.update()
.where(db.t_grunderfassung_personen.c.pers_id == data.ids.pers_id)
.returning(
db.t_grunderfassung_personen.c.Metadaten_aktualisierung,
db.t_grunderfassung_personen.c.aktualisiert,
db.t_grunderfassung_personen.c.geloescht,
)
)
@@ -424,7 +429,10 @@ def page_initrec_to_db(
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excatly one returned row"
data_from_db = from_db[0]
data_from_db = dict(from_db[0])
data_from_db = _rename_db_data(
data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
data.geloescht = data_from_db["geloescht"]
data.form_data.person["Metadaten_aktualisierung"] = data_from_db[
@@ -450,14 +458,28 @@ def page_initrec_to_db(
)
conn.execute(stmt)
# additional information
# additional information --> linking table
dump_data = _rename_db_data(
data.form_data.company, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
)
dump_data["geloescht"] = data.geloescht
logger.debug(
"[InitRec -- backend] Company - additional info: Dump Data = \n%s",
pformat(dump_data),
)
if "un_id" not in dump_data:
# only case when updated
dump_data["un_id"] = data.ids.un_id
dump_data["an_id"] = data.ids.an_id
stmt = sqlite_insert(db.t_zuordnung_personen_unternehmen).values(
**dump_data, pers_id=data.ids.pers_id
)
# dump_data["geloescht"] = data.geloescht
# logger.debug(
# "[InitRec -- backend] Company - additional info: Dump Data = \n%s",
# pformat(dump_data),
# )
stmt = stmt.on_conflict_do_update(
index_elements=["un_id", "pers_id"],
index_where=sa.text("geloescht IS NULL"),
@@ -465,13 +487,17 @@ def page_initrec_to_db(
"an_id": stmt.excluded.an_id,
"Partnersuche__kanal_aufmerksamkeit": stmt.excluded.Partnersuche__kanal_aufmerksamkeit,
"aktualisiert": datetime.datetime.now(datetime.UTC),
"geloescht": data.geloescht,
},
).returning(db.t_zuordnung_personen_unternehmen.c.id)
ret = conn.execute(stmt)
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excatly one returned row"
data_from_db = from_db[0]
data_from_db = dict(from_db[0])
data_from_db = _rename_db_data(
data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
)
data.ids.link_id = data_from_db["id"]
return data
@@ -480,35 +506,18 @@ def page_initrec_to_db(
def page_initrec_from_db(
ids: EntityIds,
) -> InitRec:
logger.debug("[InitRec -- backend] Call database loading routine...")
recording_type = RecordingType.WITHOUT_COMPANY
if ids.valid(EntityType.COMPANY):
if ids.link_id is not None or ids.valid(EntityType.COMPANY):
recording_type = RecordingType.WITH_COMPANY
# person
assert ids.pers_id, "person ID not set"
logger.debug("[AutoForm -- backend] Call database loading routine...")
stmt = db.t_grunderfassung_personen.select().where(
db.t_grunderfassung_personen.c.pers_id == ids.pers_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_person_table = dict(from_db[0])
data_person_table = _rename_db_data(
data_person_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
geloescht = data_person_table["geloescht"]
del data_person_table["geloescht"]
data_company_table: dict[str, Any] = {}
if recording_type is RecordingType.WITH_COMPANY:
assert ids.un_id, "company ID not set"
logger.debug("[InitRec -- backend] With company. Link ID: %s", ids.link_id)
if ids.link_id:
stmt = sa.select(
db.t_zuordnung_personen_unternehmen.c.pers_id,
db.t_zuordnung_personen_unternehmen.c.un_id,
db.t_zuordnung_personen_unternehmen.c.an_id,
db.t_zuordnung_personen_unternehmen.c.Partnersuche__kanal_aufmerksamkeit,
@@ -516,7 +525,10 @@ def page_initrec_from_db(
db.t_zuordnung_personen_unternehmen.c.id == ids.link_id,
)
else:
assert ids.un_id, "company ID not set"
assert ids.pers_id, "person ID not set"
stmt = sa.select(
db.t_zuordnung_personen_unternehmen.c.pers_id,
db.t_zuordnung_personen_unternehmen.c.un_id,
db.t_zuordnung_personen_unternehmen.c.an_id,
db.t_zuordnung_personen_unternehmen.c.Partnersuche__kanal_aufmerksamkeit,
@@ -533,16 +545,43 @@ def page_initrec_from_db(
assert len(from_db) == 1, "not excactly one company initial recording obtained"
data_company_table = dict(from_db[0])
ids.pers_id = data_company_table.pop("pers_id")
ids.un_id = data_company_table["un_id"]
ids.an_id = data_company_table["an_id"]
data_company_table = _rename_db_data(
data_person_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
data_company_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
)
# person
assert ids.pers_id, "person ID not set"
logger.debug("[InitRec -- backend] Person... Person ID: %s", ids.pers_id)
stmt = db.t_grunderfassung_personen.select().where(
db.t_grunderfassung_personen.c.pers_id == ids.pers_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_person_table = dict(from_db[0])
data_person_table = _rename_db_data(
data_person_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
geloescht = data_person_table["geloescht"]
del data_person_table["geloescht"]
form_data = Page_InitRec_Form_Data(
metadata={},
person=data_person_table,
company=data_company_table,
)
if geloescht is not None:
logger.warning(
"[InitRec -- backend] Loaded entries which seem to be deleted - IDs: %s", ids
)
return InitRec(
ids=ids,
recording_type=recording_type,
src/wce_crm/data_models.py
+5 -9
View File
@@ -492,11 +492,11 @@ FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN: Final[DualDict[str, str]] = DualD
class Grunderfassung_Unternehmen(FlatBaseModel):
# default in SQLAlchemy with lambda and timezone-aware datetime
Metadaten_erstellung: AwareDatetime | None = None
Metadaten_aktualisierung: AwareDatetime | None = None # see above
Metadaten_nutzer: str | None
Grunderfassung_fallnummer: str
Grunderfassung_notiz: str | None
# Metadaten_erstellung: AwareDatetime | None = None
# Metadaten_aktualisierung: AwareDatetime | None = None # see above
# Metadaten_nutzer: str | None
# Grunderfassung_fallnummer: str
# Grunderfassung_notiz: str | None
Partnersuche: Grunderfassung_PartnerSuche
@@ -697,8 +697,6 @@ class Beratungsgespraech_Einzelgespraech(BaseModel):
INITREC_CONFIGS: Final[dict[EntityType, Page_InitRec_Form_Config]] = {
EntityType.COMPANY: Page_InitRec_Form_Config(
pydantic_model=Grunderfassung_Unternehmen,
# to_db=backend.initrec_company_to_db, # TODO remove
# from_db=backend.initrec_company_from_db, # TODO remove
ignored_keys=(
"Metadaten_erstellung",
"Metadaten_aktualisierung",
@@ -712,8 +710,6 @@ INITREC_CONFIGS: Final[dict[EntityType, Page_InitRec_Form_Config]] = {
),
EntityType.PERSON: Page_InitRec_Form_Config(
pydantic_model=Grunderfassung_Personen,
# to_db=backend.initrec_person_to_db, # TODO remove
# from_db=backend.initrec_person_from_db, # TODO remove
ignored_keys=(
"Metadaten_erstellung",
"Metadaten_aktualisierung",
src/wce_crm/form_defs.py
+1 -1
View File
@@ -957,7 +957,7 @@ INITREC_PERSON = [
required=False,
key="Metadaten_nutzer",
readonly=True,
ignore_get_data=True,
ignore_get_data=False,
),
FormField(
"Wiedereintrittsdatum",
src/wce_crm/gui.py
+120 -53
View File
@@ -1622,7 +1622,6 @@ class InitRecForm(QWidget):
logger_initrecform.debug("[InitRecForm] Loaded to Pydantic.")
logger_initrecform.debug("[InitRecForm] Convert to GUI structure...")
form_data_gui = model.to_gui()
# TODO include set data in this method?
self._set_data(form_data_gui)
def reset_form(self) -> None:
@@ -2996,7 +2995,7 @@ class Page_NewEntry(QWidget):
form_data=Page_InitRec_Form_Data(metadata={}, person={}, company={}),
locked=False,
)
logger_gui.debug("[Page -- NewInitRec] State to call: %s", req_state)
logger_gui.debug("[Page -- NewInitRec] State to call (IDs): %s", req_state.ids)
self.new_initrec_requested.emit(req_state)
def _sync_state_to_GUI(self) -> None:
@@ -3178,6 +3177,14 @@ class Page_InitRec(QWidget):
)
header_layout.addWidget(btn_state2gui)
btn_print_state = QPushButton("Print State")
btn_print_state.clicked.connect(self._debug_print_state)
btn_print_state.setFixedHeight(35)
btn_print_state.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
)
header_layout.addWidget(btn_print_state)
# self.test_button = QPushButton("Initialisiere Laden")
# self.test_button.clicked.connect(self._load_from_id)
# self.test_button.setFixedHeight(35)
@@ -3332,6 +3339,11 @@ class Page_InitRec(QWidget):
self.container_layout.addStretch()
self._update_layout()
def _debug_print_state(self) -> None:
logger_initrecform.info(
"[Page -- InitRec] Printing state IDs:\n%s", pformat(self.STATE.ids)
)
def _update_layout(self) -> None:
if self.STATE.recording_type is RecordingType.WITH_COMPANY:
# add company
@@ -3428,7 +3440,16 @@ class Page_InitRec(QWidget):
self.delete_btn.setEnabled(False)
def _deletion_valid(self) -> bool:
valid_id_states = (form.relevant_ids_set() for form in self.STATE.child_modules)
# valid_id_states: list[bool] = []
relevant_ent_types: set[EntityType] = {EntityType.METADATA, EntityType.PERSON}
if self.STATE.recording_type is RecordingType.WITH_COMPANY:
relevant_ent_types.add(EntityType.COMPANY)
valid_id_states = (
form.relevant_ids_set()
for form in self.STATE.child_modules
if form.STATE.ent_type in relevant_ent_types
)
return all(valid_id_states)
def _delete_data(self) -> None:
@@ -3465,18 +3486,50 @@ class Page_InitRec(QWidget):
logger_initrecform.info("[Page -- InitRec] Load data method...")
logger_initrecform.debug("[Page -- InitRec] Entity IDs: %s", ids)
self._reset()
ids = EntityIds(
# un_id=3,
# an_id=7,
pers_id=5,
)
ids = EntityIds(
link_id=7,
)
self.STATE.ids.update(ids, ignore_none=False)
if ids.empty():
logger_initrecform.debug("[Page -- InitRec] Loading: No ID set. Do nothing!")
return
# TODO add loading logic with backend call
# !! just a dummy
initrec: InitRec = InitRec(
ids=ids,
recording_type=RecordingType.WITHOUT_COMPANY,
geloescht=None,
form_data=Page_InitRec_Form_Data(
metadata={},
person={},
company={},
),
)
initrec = backend.page_initrec_from_db(ids)
# TODO re-enable later
# 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
self._load_from_pydantic(initrec, sync_to_GUI=False)
@@ -3487,28 +3540,31 @@ class Page_InitRec(QWidget):
sync_to_GUI: bool = True,
) -> None:
logger_initrecform.info("[Page -- InitRec] Load from Pydantic model...")
# TODO add logic to update internal state
self.STATE.ids.update(data.ids)
self.STATE.recording_type = data.recording_type
self.STATE.form_data = data.form_data
self.STATE.geloescht = True if data.geloescht else False
# set form data to correct initrec form states
for ent_type, state in self.STATE.initrec_form_states.items():
state.ids.update(self.STATE.ids)
state.geloescht = self.STATE.geloescht
match ent_type:
case EntityType.COMPANY:
state.form_data = self.STATE.form_data.company
# self.initrec_form_company.load_state(state)
if (
not update_only
and self.STATE.recording_type is RecordingType.WITH_COMPANY
):
self.initrec_form_company.load_state(state)
case EntityType.PERSON:
state.form_data = self.STATE.form_data.person
# self.initrec_form_person.load_state(state)
if not update_only:
self.initrec_form_person.load_state(state)
case _:
raise RuntimeError(f"Unspecified entity type {ent_type}")
if update_only:
...
else:
...
if sync_to_GUI:
self._sync_state_to_GUI(None)
@@ -3517,24 +3573,15 @@ class Page_InitRec(QWidget):
section: Literal["content", "banners", "locking"] | None,
) -> None:
logger_page_initrec.debug(
"[Page -- InitRec] Syncing state to GUI: State...\n%s", pformat(self.STATE)
"[Page -- InitRec] Syncing state to GUI: State IDs: %s", self.STATE.ids
)
if section is None or section == "content":
self._update_layout()
if not self.STATE.geloescht:
self._load_database(self.STATE.ids)
else:
self._reset()
# initrec_form_state_person = Page_InitRec_Form_State(
# session=self.STATE.session,
# cfg=CONFIG_GRUNDERFASSUNG_PERSONEN,
# ids=self.STATE.ids,
# ent_type=RecordingType.WITHOUT_COMPANY,
# locked=False,
# )
# self.initrec_form_person.load_state(initrec_form_state_person)
self._update_layout()
if section is None or section == "banners":
self._clear_info_banners()
@@ -3546,13 +3593,14 @@ class Page_InitRec(QWidget):
if section is None or section == "locking":
if self.STATE.locked:
self.edit_buttons.set_modify_state(False, set_delete_btn=True)
for form in self.STATE.child_modules:
form.lock()
self.edit_buttons.set_modify_state(False, set_delete_btn=True)
else:
self.edit_buttons.set_modify_state(True, set_delete_btn=False)
for form in self.STATE.child_modules:
form.unlock()
self.edit_buttons.set_modify_state(True, set_delete_btn=False)
self._activate_delete()
def _sync_GUI_to_state(self) -> None:
logger_page_initrec.debug("[Page -- InitRec] Syncing GUI to state...")
@@ -3641,7 +3689,13 @@ class Page_InitRec(QWidget):
def save_data(self) -> None:
logger_page_initrec.debug("[Page -- InitRec] Save button clicked...")
return
logger_page_initrec.info("[Page -- InitRec] Syncing current state...")
# state = self.get_state()
self._sync_GUI_to_state()
form_data = self.STATE.form_data
assert form_data, "form data not set"
logger_page_initrec.debug("[Page -- InitRec] Get form data\n%s", pformat(form_data))
db_data = Page_InitRec_Form_Data(metadata={}, person={}, company={})
if self.STATE.ids.empty() and self.STATE.geloescht:
logger_page_initrec.debug(
@@ -3656,12 +3710,13 @@ class Page_InitRec(QWidget):
)
deleted_datetime = datetime.datetime.now(datetime.UTC)
assert self.STATE.form_data, "cannot delete entry with uninitialised form data"
to_db = InitRec(
data_to_db = InitRec(
ids=self.STATE.ids,
recording_type=self.STATE.recording_type,
geloescht=deleted_datetime,
form_data={},
form_data=db_data,
)
returned = self.cfg.to_db(to_db)
returned = backend.page_initrec_to_db(data_to_db)
self._load_from_pydantic(returned)
msg_box = get_message_box(
QMessageBox.Icon.Information,
@@ -3679,33 +3734,45 @@ class Page_InitRec(QWidget):
logger_page_initrec.error(
"[Page -- InitRec] Error during GUI validation phase:\n%s", pformat(errors)
)
GUI_pydantic_validation_error_handling(errors)
GUI_validation_error_handling(errors)
self._enable_save()
return
logger_page_initrec.info("[Page -- InitRec] Success! All data is valid")
logger_page_initrec.info("[Page -- InitRec] Get form data call...")
state = self.get_state()
form_data = state.form_data
assert form_data, "form data not set"
logger_page_initrec.debug("[Page -- InitRec] Get form data\n%s", pformat(form_data))
try:
logger_page_initrec.debug("[Page -- InitRec] Call Pydantic...")
validated_data = self.cfg.pydantic_model(**form_data)
# validated_data = self.cfg.pydantic_model(**form_data)
for ent_type, state in self.STATE.initrec_form_states.items():
logger_page_initrec.debug(
"[Page -- InitRec] Ent type: %s, Current state CFG model: %s",
ent_type,
state.cfg.pydantic_model,
)
match ent_type:
case EntityType.COMPANY:
val_data = state.cfg.pydantic_model(**self.STATE.form_data.company)
db_data.company = val_data.to_db(exclude=state.cfg.ignored_keys)
case EntityType.PERSON:
val_data = state.cfg.pydantic_model(**self.STATE.form_data.person)
db_data.person = val_data.to_db(exclude=state.cfg.ignored_keys)
case _:
raise RuntimeError(f"Unspecified entity type {ent_type}")
except ValidationError as e:
logger_page_initrec.error(
"[Page -- InitRec] Error during Pydantic validation phase:\n%s",
pformat(e.errors()),
)
pydantic_validation_error_handling(e)
else:
if DEBUG_NO_DATABASE:
return
# !! 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)
# db_data = validated_data.to_db(exclude=self.cfg.ignored_keys)
logger_page_initrec.debug(
"[Page -- InitRec] Form data with 'exlude' (must be saved in the database):\n%s",
pformat(db_data),
@@ -3715,15 +3782,15 @@ class Page_InitRec(QWidget):
deleted_datetime = datetime.datetime.now(datetime.UTC)
try:
# TODO change to result pattern
to_db = InitRec(
data_to_db = InitRec(
ids=self.STATE.ids,
# rec_id=self.STATE.rec_id,
recording_type=self.STATE.recording_type,
geloescht=deleted_datetime,
form_data=db_data,
)
returned = self.cfg.to_db(to_db)
returned = backend.page_initrec_to_db(data_to_db)
self._load_from_pydantic(returned)
self._load_from_pydantic(returned, update_only=True, sync_to_GUI=False)
self.update_triggered.emit()
except Exception as err:
msg_box = get_message_box(
@@ -4958,7 +5025,7 @@ class Page_Consulting(QWidget):
logger_page_consulting.error(
"[Consulting Page] Error during GUI validation phase:\n%s", pformat(errors)
)
GUI_pydantic_validation_error_handling(errors)
GUI_validation_error_handling(errors)
self._enable_save()
return
@@ -5964,7 +6031,7 @@ def pydantic_validation_error_handling(
msg_box.exec()
def GUI_pydantic_validation_error_handling(
def GUI_validation_error_handling(
errors: list[str],
) -> None:
error_text = "Bitte füllen Sie die folgenden Pflichtfelder aus:\n\n▸ " + "\n▸ ".join(
src/wce_crm/types.py
+12 -9
View File
@@ -48,7 +48,9 @@ class EntityIds:
_field_names: frozenset[str] = dc.field(init=False, repr=False, compare=False)
def __post_init__(self) -> None:
self._field_names = frozenset(f.name for f in dc.fields(EntityIds))
field_names = set(f.name for f in dc.fields(EntityIds))
field_names.remove("_field_names")
self._field_names = frozenset(field_names)
def set_by_name(
self,
@@ -88,20 +90,21 @@ class EntityIds:
ent_type: EntityType | None = None,
) -> Generator[int | None, None, None]:
if ent_type is None:
relevant_fields = (getattr(self, field.name) for field in dc.fields(self))
relevant_fields = (getattr(self, field_name) for field_name in self._field_names)
elif ent_type is EntityType.COMPANY:
RELEVANT_ATTRIBUTES = {"un_id", "an_id", "link_id"}
# RELEVANT_ATTRIBUTES = {"un_id", "an_id", "link_id"}
RELEVANT_ATTRIBUTES = {"un_id", "an_id"}
relevant_fields = (
getattr(self, field.name)
for field in dc.fields(self)
if field.name in RELEVANT_ATTRIBUTES
getattr(self, field_name)
for field_name in self._field_names
if field_name in RELEVANT_ATTRIBUTES
)
elif ent_type is EntityType.PERSON:
RELEVANT_ATTRIBUTES = {"pers_id"}
relevant_fields = (
getattr(self, field.name)
for field in dc.fields(self)
if field.name in RELEVANT_ATTRIBUTES
getattr(self, field_name)
for field_name in self._field_names
if field_name in RELEVANT_ATTRIBUTES
)
else:
raise RuntimeError(f"Unspecified entity type {ent_type}")