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

handle metadata separately in GUI, but common in database

Browse Source
This commit is contained in:
Florian Förster
2026-07-27 19:50:24 +02:00
parent ba26baa225
commit 449d401c04
6 changed files with 271 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wce_crm/backend/backend.py
+47 -26
View File
@@ -2,8 +2,9 @@ from __future__ import annotations
import copy
import datetime
from collections.abc import Container, Mapping
from pprint import pformat
from typing import TYPE_CHECKING, Any, cast
from typing import TYPE_CHECKING, Any, TypeVar, cast
import polars as pl
import sqlalchemy as sa
@@ -15,6 +16,7 @@ from wce_crm.constants import TIMEZONE_CEST
from wce_crm.data_models import (
FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN,
FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN,
FIELD_DB_SPLIT_METADATEN,
Beratungsgespraech_Einzelgespraech,
Beratungsgespraech_Vorgang,
InitRec,
@@ -40,6 +42,9 @@ if TYPE_CHECKING:
from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId
K = TypeVar("K")
V = TypeVar("V")
def _transform_for_gui_output(
data: pl.DataFrame,
@@ -355,6 +360,30 @@ def initrec_comp_contact_person_search_get_info(
# return dict(row)
def split_dict(
dic: dict[K, V],
split_keys: Container[K],
) -> tuple[dict[K, V], dict[K, V]]:
keys_in: dict[K, V] = {}
keys_out = copy.deepcopy(dic)
for check_key in dic:
if check_key in split_keys:
keys_in[check_key] = keys_out.pop(check_key)
return keys_in, keys_out
def dic_deep_update(
dic: dict[K, V],
data: Mapping[K, V],
) -> dict[K, V]:
dic_copy = copy.deepcopy(dic)
dic_copy.update(data)
return dic_copy
def _rename_db_data(
db_data: dict[str, Any],
mapping: DualDict,
@@ -379,18 +408,15 @@ def page_initrec_to_db(
with db.ENGINE.begin() as conn:
# person
dump_data = _rename_db_data(
data.form_data.person, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
dump_data = dic_deep_update(data.form_data.person, data.form_data.metadata)
dump_data = _rename_db_data(dump_data, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN)
dump_data["geloescht"] = data.geloescht
if data.ids.pers_id is None:
logger.debug("[InitRec -- backend] Insert...")
stmt = sa.insert(db.t_grunderfassung_personen).returning(
db.t_grunderfassung_personen.c.pers_id,
db.t_grunderfassung_personen.c.aktualisiert,
db.t_grunderfassung_personen.c.geloescht,
db.t_grunderfassung_personen
)
stmt_compiled = str(stmt.compile(db.ENGINE))
logger.debug(
@@ -409,10 +435,6 @@ def page_initrec_to_db(
)
data.ids.pers_id = data_from_db["pers_id"]
data.geloescht = data_from_db["geloescht"]
data.form_data.person["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully")
@@ -421,10 +443,7 @@ def page_initrec_to_db(
stmt = (
db.t_grunderfassung_personen.update()
.where(db.t_grunderfassung_personen.c.pers_id == data.ids.pers_id)
.returning(
db.t_grunderfassung_personen.c.aktualisiert,
db.t_grunderfassung_personen.c.geloescht,
)
.returning(db.t_grunderfassung_personen)
)
ret = conn.execute(stmt, dump_data)
@@ -435,16 +454,22 @@ def page_initrec_to_db(
data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
data.geloescht = data_from_db["geloescht"]
data.form_data.person["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
# data.geloescht = data_from_db["geloescht"]
# data.form_data.person["Metadaten_aktualisierung"] = data_from_db[
# "Metadaten_aktualisierung"
# ]
logger.debug(
"[InitRec -- backend] Updated InitRec Person with ID %d successfully",
data.ids.pers_id,
)
data.geloescht = data_from_db["geloescht"]
del data_from_db["geloescht"]
in_filter, out_filter = split_dict(data_from_db, FIELD_DB_SPLIT_METADATEN)
data.form_data.metadata = in_filter
data.form_data.person = out_filter
# company
if data.recording_type is RecordingType.WITH_COMPANY:
# insert into company table and add additional information
@@ -475,11 +500,6 @@ def page_initrec_to_db(
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"],
@@ -571,10 +591,11 @@ def page_initrec_from_db(
)
geloescht = data_person_table["geloescht"]
del data_person_table["geloescht"]
in_filter, out_filter = split_dict(data_person_table, FIELD_DB_SPLIT_METADATEN)
form_data = Page_InitRec_Form_Data(
metadata={},
person=data_person_table,
metadata=in_filter,
person=out_filter,
company=data_company_table,
)
src/wce_crm/data_models.py
+66 -18
View File
@@ -19,7 +19,7 @@ from pydantic import (
)
from pydantic_core import ErrorDetails
from wce_crm.form_defs import INITREC_COMP, INITREC_PERSON, FormField
from wce_crm.form_defs import INITREC_COMP, INITREC_METADATEN, INITREC_PERSON, FormField
from wce_crm.types import ConsultingType, EntityIds, EntityType, RecordingType
if TYPE_CHECKING:
@@ -222,10 +222,22 @@ class Page_NewEntry_State(PageState[Module]):
session: Session
INITREC_REC_TO_ENTITIES: Final[dict[RecordingType, frozenset[EntityType]]] = {
RecordingType.WITH_COMPANY: frozenset((EntityType.COMPANY, EntityType.PERSON)),
RecordingType.WITHOUT_COMPANY: frozenset((EntityType.PERSON,)),
}
# TODO check removal
# INITREC_REC_TO_ENTITIES: Final[dict[RecordingType, frozenset[EntityType]]] = {
# RecordingType.WITH_COMPANY: frozenset(
# (
# EntityType.METADATA,
# EntityType.COMPANY,
# EntityType.PERSON,
# )
# ),
# RecordingType.WITHOUT_COMPANY: frozenset(
# (
# EntityType.METADATA,
# EntityType.PERSON,
# )
# ),
# }
@dc.dataclass(slots=True, kw_only=True)
@@ -251,7 +263,16 @@ class Page_InitRec_State(PageState["InitRecForm"]):
# pydantic_models: dict[RecordingType, type[FlatBaseModel]] = dc.field(default_factory=dict)
def __post_init__(self) -> None:
relevant_entities = INITREC_REC_TO_ENTITIES[self.recording_type]
# TODO check removal
# relevant_entities = INITREC_REC_TO_ENTITIES[self.recording_type]
relevant_entities = frozenset(
(
EntityType.METADATA,
EntityType.COMPANY,
EntityType.PERSON,
)
)
for ent_type in relevant_entities:
initrec_form_state = Page_InitRec_Form_State(
session=self.session,
@@ -483,11 +504,33 @@ FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN: Final[DualDict[str, str]] = DualDict
Metadaten_erstellung="erstellt",
Metadaten_aktualisierung="aktualisiert",
)
FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN: Final[DualDict[str, str]] = DualDict(
Partnersuche__un_suche="un_id",
Partnersuche__person_suche="an_id",
)
FIELD_DB_SPLIT_METADATEN: Final[frozenset[str]] = frozenset(
(
"Metadaten_erstellung",
"Metadaten_aktualisierung",
"Metadaten_nutzer",
"Metadaten_wiedereintrittsdatum",
"Grunderfassung_fallnummer",
"Grunderfassung_notiz",
)
)
class Grunderfassung_Metadaten(FlatBaseModel):
# displayed and handled separately, on database level included in 'Person'
Metadaten_erstellung: AwareDatetime | None = None
Metadaten_aktualisierung: AwareDatetime | None = None # see above
Metadaten_nutzer: str | None
Metadaten_wiedereintrittsdatum: datetime.date | None = None
Grunderfassung_fallnummer: str
Grunderfassung_notiz: str | None
class Grunderfassung_Unternehmen(FlatBaseModel):
# default in SQLAlchemy with lambda and timezone-aware datetime
@@ -502,12 +545,9 @@ class Grunderfassung_Unternehmen(FlatBaseModel):
class Grunderfassung_Personen(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
Metadaten_wiedereintrittsdatum: datetime.date | None = None
Grunderfassung_fallnummer: str
Grunderfassung_notiz: str | None
# Metadaten_erstellung: AwareDatetime | None = None
# Metadaten_aktualisierung: AwareDatetime | None = None # see above
# Metadaten_nutzer: str | None
Projektrelevanz: Grunderfassung_Projektrelevanz
Kontaktperson: Grunderfassung_Kontaktperson
@@ -694,12 +734,19 @@ class Beratungsgespraech_Einzelgespraech(BaseModel):
INITREC_CONFIGS: Final[dict[EntityType, Page_InitRec_Form_Config]] = {
EntityType.COMPANY: Page_InitRec_Form_Config(
pydantic_model=Grunderfassung_Unternehmen,
EntityType.METADATA: Page_InitRec_Form_Config(
pydantic_model=Grunderfassung_Metadaten,
ignored_keys=(
"Metadaten_erstellung",
"Metadaten_aktualisierung",
),
form_fields=INITREC_METADATEN,
add_buttons=False,
id_mapping={},
),
EntityType.COMPANY: Page_InitRec_Form_Config(
pydantic_model=Grunderfassung_Unternehmen,
ignored_keys=tuple(),
form_fields=INITREC_COMP,
add_buttons=False,
id_mapping={
@@ -709,10 +756,11 @@ INITREC_CONFIGS: Final[dict[EntityType, Page_InitRec_Form_Config]] = {
),
EntityType.PERSON: Page_InitRec_Form_Config(
pydantic_model=Grunderfassung_Personen,
ignored_keys=(
"Metadaten_erstellung",
"Metadaten_aktualisierung",
),
ignored_keys=tuple(),
# ignored_keys=(
# "Metadaten_erstellung",
# "Metadaten_aktualisierung",
# ),
form_fields=INITREC_PERSON,
add_buttons=False,
id_mapping={},
src/wce_crm/form_defs.py
+63 -17
View File
@@ -187,7 +187,7 @@ COUNTRY_LIST: Final[CountryList] = get_country_list_german()
GERMAN_STATE_LIST: Final[CountryList] = get_list_germany_states()
INITREC_COMP_SEARCH_HEAD = [
_INITREC_COMP_SEARCH_HEAD = [
FormField(
"Suche",
FormFieldType.EXTENDED_DROPDOWN,
@@ -329,7 +329,7 @@ INITREC_COMP_SEARCH_HEAD = [
),
]
INITREC_CONTACT_PERSON = [
_INITREC_CONTACT_PERSON = [
# TODO check removal
FormField(
"Name Unternehmen/Netzwerkpartner (vorausgefüllt von Suche)",
@@ -400,7 +400,7 @@ INITREC_CONTACT_PERSON = [
]
INITREC_MASTER_DATA = [
_INITREC_MASTER_DATA = [
FormField(
"Titel",
FormFieldType.TEXT,
@@ -560,7 +560,7 @@ INITREC_MASTER_DATA = [
]
INITREC_ADDITIONAL_DATA = [
_INITREC_ADDITIONAL_DATA = [
FormField(
"Deutsch als Kommunikationssprache",
FormFieldType.DROPDOWN,
@@ -635,7 +635,7 @@ INITREC_ADDITIONAL_DATA = [
]
INITREC_SCHOOL = [
_INITREC_SCHOOL = [
FormField("Abschluss", FormFieldType.TEXT, required=False, key="SB_abschluss"),
FormField(
"Abschlussgrad laut Dokument",
@@ -673,7 +673,7 @@ INITREC_SCHOOL = [
]
INITREC_HIGHER_EDUCATION = [
_INITREC_HIGHER_EDUCATION = [
FormField(
"Anerkennung",
FormFieldType.TEXT,
@@ -742,7 +742,7 @@ INITREC_HIGHER_EDUCATION = [
]
INITREC_WORK_EXPERIENCE = [
_INITREC_WORK_EXPERIENCE = [
FormField(
"Branche",
FormFieldType.DROPDOWN,
@@ -835,7 +835,7 @@ INITREC_WORK_EXPERIENCE = [
]
INITREC_LANGUAGES = [
_INITREC_LANGUAGES = [
FormField(
"Sprache",
FormFieldType.TEXT,
@@ -930,11 +930,11 @@ INITREC_COMP = [
FormFieldType.CUSTOM,
custom_widget="grunderfassung_unternehmen_suche",
key="Partnersuche",
children=INITREC_COMP_SEARCH_HEAD,
children=_INITREC_COMP_SEARCH_HEAD,
),
]
INITREC_PERSON = [
INITREC_METADATEN = [
FormField(
"Ersteintrag Datum",
FormFieldType.TEXT_DATETIME,
@@ -978,6 +978,52 @@ INITREC_PERSON = [
required=False,
key="Grunderfassung_notiz",
),
]
INITREC_PERSON = [
# FormField(
# "Ersteintrag Datum",
# FormFieldType.TEXT_DATETIME,
# required=False,
# key="Metadaten_erstellung",
# readonly=True,
# ignore_get_data=True,
# ),
# FormField(
# "Aktualisierung Datum",
# FormFieldType.TEXT_DATETIME,
# required=False,
# key="Metadaten_aktualisierung",
# readonly=True,
# ignore_get_data=True,
# ),
# FormField(
# "Aktualisierung Nutzer",
# FormFieldType.TEXT,
# required=False,
# key="Metadaten_nutzer",
# readonly=True,
# ignore_get_data=False,
# ),
# FormField(
# "Wiedereintrittsdatum",
# FormFieldType.DATE,
# required=False,
# key="Metadaten_wiedereintrittsdatum",
# readonly=False,
# ),
# FormField(
# "Fallnummer",
# FormFieldType.TEXT,
# required=True,
# key="Grunderfassung_fallnummer",
# ),
# FormField(
# "Notizen",
# FormFieldType.LONGTEXT,
# required=False,
# key="Grunderfassung_notiz",
# ),
FormField(
"Status && Projektrelevanz",
FormFieldType.GROUP,
@@ -1009,42 +1055,42 @@ INITREC_PERSON = [
"Daten Kontaktperson",
FormFieldType.GROUP,
key="Kontaktperson",
children=INITREC_CONTACT_PERSON,
children=_INITREC_CONTACT_PERSON,
),
FormField(
"Stammdaten",
FormFieldType.GROUP,
key="Stammdaten",
children=INITREC_MASTER_DATA,
children=_INITREC_MASTER_DATA,
),
FormField(
"Weitere Informationen",
FormFieldType.GROUP,
key="WeitereInfos",
children=INITREC_ADDITIONAL_DATA,
children=_INITREC_ADDITIONAL_DATA,
),
FormField(
"Schulbildung",
FormFieldType.DYNAMIC_LIST,
children=INITREC_SCHOOL,
children=_INITREC_SCHOOL,
key="Schulbildung",
),
FormField(
"Studium/Ausbildung",
FormFieldType.DYNAMIC_LIST,
children=INITREC_HIGHER_EDUCATION,
children=_INITREC_HIGHER_EDUCATION,
key="HoehereBildung",
),
FormField(
"Arbeitserfahrung",
FormFieldType.DYNAMIC_LIST,
children=INITREC_WORK_EXPERIENCE,
children=_INITREC_WORK_EXPERIENCE,
key="Arbeitserfahrung",
),
FormField(
"Sprachkenntnisse",
FormFieldType.DYNAMIC_LIST,
children=INITREC_LANGUAGES,
children=_INITREC_LANGUAGES,
key="Sprachkenntnisse",
),
]
src/wce_crm/gui.py
+56 -4
View File
@@ -1616,6 +1616,11 @@ class InitRecForm(QWidget):
pformat(self.STATE.form_data),
)
# TODO error catching?
logger_initrecform.info(
"[InitRecForm] Pydantic model name: %s",
self.STATE.cfg.pydantic_model.__name__,
)
model = self.cfg.pydantic_model(**self.STATE.form_data)
logger_initrecform.debug("[InitRecForm] Loaded to Pydantic.")
logger_initrecform.debug("[InitRecForm] Convert to GUI structure...")
@@ -3097,6 +3102,12 @@ class CollapsibleBox(QWidget):
layout.addWidget(widget)
self.setContentLayout(layout)
def set_open(
self,
open: bool,
) -> None:
self._on_toggle(open)
class Page_InitRec(QWidget):
back_main_requested = Signal() # back to main page
@@ -3247,6 +3258,28 @@ class Page_InitRec(QWidget):
# now we try to render the layout within the collapsable box
self.container_layout.addSpacing(20)
# ** metadata
self.initrec_form_metadata_box = CollapsibleBox("Metadaten")
ent_type = EntityType.METADATA
self.initrec_form_metadata = InitRecForm(self.STATE.initrec_form_states[ent_type])
assert self.STATE.initrec_form_states[ent_type] is self.initrec_form_metadata.STATE
self.STATE.child_modules.append(self.initrec_form_metadata)
self.initrec_form_metadata_box.setContentWidget(self.initrec_form_metadata)
self.container_layout.addWidget(self.initrec_form_metadata_box)
# --- CUSTOM LOGIC ---
# ** 'Wiedereintrittsdatum' -> only relevant without company
search_res = search_widgets_by_key(
self.initrec_form_metadata.widget_registry, "wiedereintrittsdatum"
)
print(f">>>> {pformat(search_res)}")
assert len(search_res) == 1
field_wiedereintrittsdatum = search_res[0]["widget"]
assert isinstance(field_wiedereintrittsdatum, QDateEdit)
self.field_wiedereintrittsdatum = field_wiedereintrittsdatum
# ** person
self.initrec_form_person_box = CollapsibleBox("Angaben zur Person")
ent_type = EntityType.PERSON
@@ -3286,7 +3319,7 @@ class Page_InitRec(QWidget):
assert self.STATE.initrec_form_states[ent_type] is self.initrec_form_company.STATE
target_idx = self.container_layout.indexOf(self.initrec_form_person_box) - 1
target_idx = self.container_layout.indexOf(self.initrec_form_person_box)
self.initrec_form_company_box.setContentWidget(self.initrec_form_company)
self.container_layout.insertWidget(target_idx, self.initrec_form_company_box)
@@ -3321,9 +3354,13 @@ class Page_InitRec(QWidget):
self.company_profile_requested.emit(req_state)
def _update_layout(self) -> None:
self.initrec_form_metadata_box.set_open(True)
if self.STATE.recording_type is RecordingType.WITH_COMPANY:
# add company
self.initrec_form_company_box.setVisible(True)
self.initrec_form_metadata.top_level_form_layout.setRowVisible(
self.field_wiedereintrittsdatum, False
)
self.comp_profile_btn.setVisible(True)
if self.STATE.ids.valid(EntityType.COMPANY):
self.comp_profile_btn.setEnabled(True)
@@ -3332,6 +3369,9 @@ class Page_InitRec(QWidget):
else:
self.initrec_form_company_box.setVisible(False)
self.initrec_form_metadata.top_level_form_layout.setRowVisible(
self.field_wiedereintrittsdatum, True
)
self.comp_profile_btn.setVisible(False)
self.comp_profile_btn.setEnabled(False)
@@ -3454,9 +3494,9 @@ class Page_InitRec(QWidget):
pers_id=5,
)
ids = EntityIds(
link_id=7,
)
# ids = EntityIds(
# link_id=7,
# )
self.STATE.ids.update(ids, ignore_none=False)
if ids.empty():
@@ -3523,6 +3563,13 @@ class Page_InitRec(QWidget):
state.form_data = self.STATE.form_data.person
if not update_only:
self.initrec_form_person.load_state(state)
case EntityType.METADATA:
state.form_data = self.STATE.form_data.metadata
logger_initrecform.info(
"[Page -- InitRec] Load METADATA from Pydantic model...\n%s",
pformat(self.STATE.form_data.metadata),
)
self.initrec_form_metadata.load_state(state)
case _:
raise RuntimeError(f"Unspecified entity type {ent_type}")
@@ -3602,6 +3649,8 @@ class Page_InitRec(QWidget):
self.STATE.form_data.company = form_state.form_data
case EntityType.PERSON:
self.STATE.form_data.person = form_state.form_data
case EntityType.METADATA:
self.STATE.form_data.metadata = form_state.form_data
case _:
raise RuntimeError(f"Unspecified entity type {form_state.ent_type}")
@@ -3718,6 +3767,9 @@ class Page_InitRec(QWidget):
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 EntityType.METADATA:
val_data = state.cfg.pydantic_model(**self.STATE.form_data.metadata)
db_data.metadata = val_data.to_db(exclude=state.cfg.ignored_keys)
case _:
raise RuntimeError(f"Unspecified entity type {ent_type}")
src/wce_crm/types.py
+7
View File
@@ -106,6 +106,13 @@ class EntityIds:
for field_name in self._field_names
if field_name in RELEVANT_ATTRIBUTES
)
elif ent_type is EntityType.METADATA:
RELEVANT_ATTRIBUTES = set()
relevant_fields = (
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}")