switch to new data model with distinct company and person data

This commit is contained in:
2026-06-26 14:14:45 +02:00
parent b9427bc756
commit df93914942
9 changed files with 436 additions and 265 deletions
+84 -15
View File
@@ -154,11 +154,11 @@ def initrec_comp_contact_person_search_get_info(
# // internals
def initrec_insert_initial_recording(
def initrec_company_insert_initial_recording(
data: dict[str, Any],
) -> InitRecId:
logger.debug("[Call backend] insert_initial_recording")
stmt = db.grunderfassung.insert()
stmt = db.grunderfassung_unternehmen.insert()
with db.ENGINE.begin() as conn:
ret = conn.execute(stmt, data)
@@ -171,21 +171,25 @@ def initrec_insert_initial_recording(
return prim_keys[0]
def initrec_update_initial_recording(
def initrec_company_update_initial_recording(
id_: InitRecId,
data: dict[str, Any],
) -> None:
logger.debug("[Call backend] update_initial_recording")
stmt = db.grunderfassung.update().where(db.grunderfassung.c.erfassung_id == id_)
stmt = db.grunderfassung_unternehmen.update().where(
db.grunderfassung_unternehmen.c.un_id == id_
)
with db.ENGINE.begin() as conn:
conn.execute(stmt, data)
def initrec_get_initial_recording(
def initrec_company_get_initial_recording(
id_: InitRecId,
) -> dict[str, Any]:
logger.debug("[Call backend] get_initial_recording")
stmt = db.grunderfassung.select().where(db.grunderfassung.c.erfassung_id == id_)
stmt = db.grunderfassung_unternehmen.select().where(
db.grunderfassung_unternehmen.c.un_id == id_
)
with db.ENGINE.begin() as conn:
ret = conn.execute(stmt)
@@ -198,11 +202,75 @@ def initrec_get_initial_recording(
return row._asdict()
def initrec_delete_initial_recording(
def initrec_company_delete_initial_recording(
id_: InitRecId,
) -> None:
logger.debug("[Call backend] delete_initial_recording")
stmt = db.grunderfassung.delete().where(db.grunderfassung.c.erfassung_id == id_)
stmt = db.grunderfassung_unternehmen.delete().where(
db.grunderfassung_unternehmen.c.un_id == id_
)
with db.ENGINE.begin() as conn:
ret = conn.execute(stmt)
if ret.rowcount == 0:
raise KeyError(f"Database ID {id_} not found for deletion")
def initrec_person_insert_initial_recording(
data: dict[str, Any],
) -> InitRecId:
logger.debug("[Call backend] insert_initial_recording")
stmt = db.grunderfassung_personen.insert()
with db.ENGINE.begin() as conn:
ret = conn.execute(stmt, data)
if ret.rowcount == 0:
raise IOError("Entry was not inserted correctly")
prim_keys = ret.inserted_primary_key
assert prim_keys
return prim_keys[0]
def initrec_person_update_initial_recording(
id_: InitRecId,
data: dict[str, Any],
) -> None:
logger.debug("[Call backend] update_initial_recording")
stmt = db.grunderfassung_personen.update().where(
db.grunderfassung_personen.c.pers_id == id_
)
with db.ENGINE.begin() as conn:
conn.execute(stmt, data)
def initrec_person_get_initial_recording(
id_: InitRecId,
) -> dict[str, Any]:
logger.debug("[Call backend] get_initial_recording")
stmt = db.grunderfassung_personen.select().where(
db.grunderfassung_personen.c.pers_id == id_
)
with db.ENGINE.begin() as conn:
ret = conn.execute(stmt)
if ret.rowcount == 0:
raise KeyError(f"Database ID {id_} not found")
row = ret.fetchone()
assert row, "row was not obtained"
return row._asdict()
def initrec_person_delete_initial_recording(
id_: InitRecId,
) -> None:
logger.debug("[Call backend] delete_initial_recording")
stmt = db.grunderfassung_personen.delete().where(
db.grunderfassung_personen.c.pers_id == id_
)
with db.ENGINE.begin() as conn:
ret = conn.execute(stmt)
@@ -218,15 +286,16 @@ class FrontpageCompany:
is_company: bool
# TODO separate functions for companies and person: focus on companies first
def front_get_company_list() -> list[FrontpageCompany]:
logger.debug("[Call backend] get_company_list")
stmt = sql.select(
db.grunderfassung.c.erfassung_id,
db.grunderfassung.c.Partnersuche__un_suche,
db.grunderfassung.c.Stammdaten__vorname,
db.grunderfassung.c.Stammdaten__name,
db.grunderfassung.c.Metadaten_aktualisierung,
).order_by(db.grunderfassung.c.Metadaten_aktualisierung.desc())
db.grunderfassung_unternehmen.c.un_id,
db.grunderfassung_unternehmen.c.Partnersuche__un_suche,
# db.grunderfassung_unternehmen.c.Stammdaten__vorname,
# db.grunderfassung_unternehmen.c.Stammdaten__name,
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
).order_by(db.grunderfassung_unternehmen.c.Metadaten_aktualisierung.desc())
with db.ENGINE.connect() as conn:
res = conn.execute(stmt)
@@ -234,7 +303,7 @@ def front_get_company_list() -> list[FrontpageCompany]:
front_page_companies: list[FrontpageCompany] = []
for entry in res.mappings():
erfassung_id = entry["erfassung_id"]
erfassung_id = entry["un_id"]
ma_id = entry["Partnersuche__un_suche"]
datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"])
datetime_akt = datetime_akt.astimezone()
+1 -1
View File
@@ -4,6 +4,6 @@ from typing import Final
CUSTOM_WIDGET_NAMES: Final[frozenset] = frozenset(
[
"grunderfassung_suche",
"grunderfassung_unternehmen_suche",
]
)
+13 -3
View File
@@ -83,7 +83,7 @@ class FlatBaseModel(BaseModel):
@model_validator(mode="before")
@classmethod
def __unflatten_input(
def _unflatten_input(
cls,
data: Any,
) -> Any: # type: ignore
@@ -141,7 +141,18 @@ class FlatBaseModel(BaseModel):
return dict(items)
class Grunderfassung(FlatBaseModel):
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
Partnersuche: Grunderfassung_PartnerSuche
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
@@ -150,7 +161,6 @@ class Grunderfassung(FlatBaseModel):
Grunderfassung_fallnummer: str
Grunderfassung_notiz: str | None
Partnersuche: Grunderfassung_PartnerSuche | None = None
Projektrelevanz: Grunderfassung_Projektrelevanz
Kontaktperson: Grunderfassung_Kontaktperson
Stammdaten: Grunderfassung_Stammdaten
+59 -83
View File
@@ -62,83 +62,6 @@ MD_CRM = sql.MetaData()
MD_MAIN = sql.MetaData()
ENGINE = sql.create_engine(f"sqlite:///{constants.Config.DB_PATH_MAIN}")
# ---------- OLD "Kontaktliste" ----------
# md_kontaktliste = sql.MetaData()
# ext_kl_unternehmen: sql.Table = Table(
# "Unternehmen",
# md_kontaktliste,
# Column("u_id", sql.Integer, nullable=False, unique=True),
# Column("u_zeitstempel_eintrag", sql.DateTime, nullable=False),
# Column("u_rechtsform", sql.Text, nullable=False),
# Column("u_firmenname", sql.Text, nullable=False),
# Column("u_strasse", sql.Text, nullable=False),
# Column("u_hausnummer", sql.Text, nullable=False),
# Column("u_adresszusatz", sql.Text, nullable=True),
# Column("u_plz", sql.Text, nullable=False),
# Column("u_ort", sql.Text, nullable=False),
# Column("u_postfach", sql.Text, nullable=True),
# Column("u_website", sql.Text, nullable=True),
# Column("u_anrede", sql.Text, nullable=False),
# Column("u_titel", sql.Text, nullable=True),
# Column("u_vorname", sql.Text, nullable=False),
# Column("u_nachname", sql.Text, nullable=False),
# Column("u_funktion", sql.Text, nullable=False),
# Column("u_mail", sql.Text, nullable=False),
# Column("u_telefon", sql.Text, nullable=False),
# Column("u_plz_postfach", sql.Text, nullable=True),
# Column("u_einwilligung_inhaber", sql.Boolean, nullable=True),
# Column("u_einwilligung_ansprechpartner", sql.Boolean, nullable=True),
# Column("u_aktiv", sql.Boolean, nullable=False, default=1),
# )
# ext_kl_unternehmen_schema: t.PolarsSchema = {
# "u_id": pl.UInt64,
# "u_zeitstempel_eintrag": pl.Datetime,
# "u_rechtsform": pl.String,
# "u_firmenname": pl.String,
# "u_strasse": pl.String,
# "u_hausnummer": pl.String,
# "u_adresszusatz": pl.String,
# "u_plz": pl.String,
# "u_ort": pl.String,
# "u_postfach": pl.String,
# "u_website": pl.String,
# "u_anrede": pl.String,
# "u_titel": pl.String,
# "u_vorname": pl.String,
# "u_nachname": pl.String,
# "u_funktion": pl.String,
# "u_mail": pl.String,
# "u_telefon": pl.String,
# "u_plz_postfach": pl.String,
# "u_einwilligung_inhaber": pl.Boolean,
# "u_einwilligung_ansprechpartner": pl.Boolean,
# "u_aktiv": pl.Boolean,
# }
# def get_ext_kontaktliste(
# db_path: Path | None,
# ) -> pl.DataFrame:
# if db_path is None:
# ENV_PTH = os.environ.get("DOPT_DB_KONTAKTLISTE", None)
# if ENV_PTH is None:
# raise ValueError("No database path provided or found as ENV var.")
# db_path = Path(ENV_PTH)
# if not db_path.exists():
# raise FileNotFoundError(f"Database not found under >{db_path}<")
# engine = sql.create_engine(f"sqlite:///{db_path}")
# stmt = sql.select(ext_kl_unternehmen)
# return pl.read_database(stmt, engine, schema_overrides=ext_kl_unternehmen_schema)
# df_kontaktliste = get_ext_kontaktliste(None)
# ----------------------------------------------------
ext_crm_master: sql.Table = Table(
"Master",
@@ -324,11 +247,41 @@ def get_ext_crm_contact_person(
# df_contact_person = get_ext_crm_contact_person(None)
DF_CONTACT_PERSON = get_ext_crm_contact_person(constants.Config.DB_PATH_CRM)
grunderfassung: sql.Table = Table(
"grunderfassung",
grunderfassung_unternehmen: sql.Table = Table(
"grunderfassung_unternehmen",
MD_MAIN,
Column(
"erfassung_id",
"un_id",
sql.Integer,
primary_key=True,
autoincrement=True,
),
Column(
"Metadaten_erstellung",
UTCDateTime,
nullable=True,
default=lambda: datetime.datetime.now(datetime.UTC),
),
Column(
"Metadaten_aktualisierung",
UTCDateTime,
nullable=True,
default=lambda: datetime.datetime.now(datetime.UTC),
onupdate=lambda: datetime.datetime.now(datetime.UTC),
),
Column("Metadaten_nutzer", sql.String(20), nullable=True),
Column("Grunderfassung_fallnummer", sql.Text, nullable=True),
Column("Grunderfassung_notiz", sql.Text, nullable=True),
Column("Partnersuche__kanal_aufmerksamkeit", sql.Text, nullable=True, default=None),
Column("Partnersuche__person_suche", sql.Integer, nullable=True, default=None),
Column("Partnersuche__un_suche", sql.Integer, nullable=True, default=None),
)
grunderfassung_personen: sql.Table = Table(
"grunderfassung_personen",
MD_MAIN,
Column(
"pers_id",
sql.Integer,
primary_key=True,
autoincrement=True,
@@ -364,9 +317,6 @@ grunderfassung: sql.Table = Table(
), # TODO: check if needed when set by trigger
Column("Kontaktperson__KP_titel", sql.Text, nullable=True),
Column("Kontaktperson__KP_vorname", sql.Text, nullable=True),
Column("Partnersuche__kanal_aufmerksamkeit", sql.Text, nullable=True, default=None),
Column("Partnersuche__person_suche", sql.Integer, nullable=True, default=None),
Column("Partnersuche__un_suche", sql.Integer, nullable=True, default=None),
Column("Projektrelevanz__relevanz", sql.Text, nullable=True),
Column("Projektrelevanz__foerderperiode", sql.Text, nullable=True),
Column("Schulbildung", sql.Text, nullable=True),
@@ -398,3 +348,29 @@ grunderfassung: sql.Table = Table(
Column("WeitereInfos__WI_gueltigkeit_aufenthaltstitel", sql.Date, nullable=True),
Column("WeitereInfos__WI_meldung_institution", sql.Text, nullable=True),
)
zuordnung_personen_unternehmen: sql.Table = Table(
"zuordnung_personen_unternehmen",
MD_MAIN,
Column(
"id",
sql.Integer,
primary_key=True,
autoincrement=True,
),
Column("un_id", sql.ForeignKey("grunderfassung_unternehmen.un_id"), nullable=False),
Column("pers_id", sql.ForeignKey("grunderfassung_personen.pers_id"), nullable=False),
Column(
"gueltig_ab",
UTCDateTime,
nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC),
),
Column(
"gueltig_bis",
UTCDateTime,
nullable=True,
default=lambda: datetime.datetime.now(datetime.UTC),
onupdate=lambda: datetime.datetime.now(datetime.UTC),
),
)
+1 -70
View File
@@ -920,79 +920,10 @@ INITREC_COMP = [
FormField(
"Suche",
FormFieldType.CUSTOM,
custom_widget="grunderfassung_suche",
custom_widget="grunderfassung_unternehmen_suche",
key="Partnersuche",
children=INITREC_COMP_SEARCH_HEAD,
),
FormField(
"Status && Projektrelevanz",
FormFieldType.GROUP,
key="Projektrelevanz",
children=[
FormField(
"Projektrelevanz",
FormFieldType.DYNAMIC_DROPDOWN_OPTION,
key="",
trigger_value="ja",
children=[
FormField(
"Relevanz",
FormFieldType.DROPDOWN,
required=True,
options=[("ja", None), ("nein", None)],
key="relevanz",
children=[
FormField(
"Förderperiode", FormFieldType.TEXT, key="foerderperiode"
),
],
),
],
),
],
),
FormField(
"Daten Kontaktperson",
FormFieldType.GROUP,
key="Kontaktperson",
children=INITREC_CONTACT_PERSON,
),
FormField(
"Stammdaten",
FormFieldType.GROUP,
key="Stammdaten",
children=INITREC_MASTER_DATA,
),
FormField(
"Weitere Informationen",
FormFieldType.GROUP,
key="WeitereInfos",
children=INITREC_ADDITIONAL_DATA,
),
FormField(
"Schulbildung",
FormFieldType.DYNAMIC_LIST,
children=INITREC_SCHOOL,
key="Schulbildung",
),
FormField(
"Studium/Ausbildung",
FormFieldType.DYNAMIC_LIST,
children=INITREC_HIGHER_EDUCATION,
key="HoehereBildung",
),
FormField(
"Arbeitserfahrung",
FormFieldType.DYNAMIC_LIST,
children=INITREC_WORK_EXPERIENCE,
key="Arbeitserfahrung",
),
FormField(
"Sprachkenntnisse",
FormFieldType.DYNAMIC_LIST,
children=INITREC_LANGUAGES,
key="Sprachkenntnisse",
),
]
INITREC_PERSON = [
+62 -56
View File
@@ -58,7 +58,12 @@ from PySide6.QtWidgets import (
import wce_crm.constants
from wce_crm.backend import backend
from wce_crm.data_models import COLUMN_SEP, FlatBaseModel, Grunderfassung
from wce_crm.data_models import (
COLUMN_SEP,
FlatBaseModel,
Grunderfassung_Personen,
Grunderfassung_Unternehmen,
)
from wce_crm.form_defs import (
INITREC_COMP,
INITREC_PERSON,
@@ -1351,7 +1356,7 @@ class AutoForm(QWidget):
logger_auto_form.debug(
"Loaded data dict:\n%s Passing to Pydantic...", pformat(loaded_data)
)
model = Grunderfassung(**loaded_data)
model = self.cfg.model(**loaded_data)
logger_auto_form.debug("Loaded to Pydantic.")
logger_auto_form.debug("Convert to GUI structure...")
form_data = model.to_gui()
@@ -2081,35 +2086,33 @@ class NewEntrySelect_view(QWidget):
CONFIG_GRUNDERFASSUNG_UNTERNEHMEN: Final[AutoFormConfig] = AutoFormConfig(
model=Grunderfassung,
data_insert=backend.initrec_insert_initial_recording,
data_update=backend.initrec_update_initial_recording,
data_get=backend.initrec_get_initial_recording,
data_delete=backend.initrec_delete_initial_recording,
model=Grunderfassung_Unternehmen,
data_insert=backend.initrec_company_insert_initial_recording,
data_update=backend.initrec_company_update_initial_recording,
data_get=backend.initrec_company_get_initial_recording,
data_delete=backend.initrec_company_delete_initial_recording,
ignored_keys=(
"Metadaten_erstellung",
"Metadaten_aktualisierung",
"Metadaten_wiedereintrittsdatum",
),
form_fields=INITREC_COMP,
)
CONFIG_GRUNDERFASSUNG_PERSONEN: Final[AutoFormConfig] = AutoFormConfig(
model=Grunderfassung,
data_insert=backend.initrec_insert_initial_recording,
data_update=backend.initrec_update_initial_recording,
data_get=backend.initrec_get_initial_recording,
data_delete=backend.initrec_delete_initial_recording,
model=Grunderfassung_Personen,
data_insert=backend.initrec_person_insert_initial_recording,
data_update=backend.initrec_person_update_initial_recording,
data_get=backend.initrec_person_get_initial_recording,
data_delete=backend.initrec_person_delete_initial_recording,
ignored_keys=(
"Metadaten_erstellung",
"Metadaten_aktualisierung",
"Partnersuche",
),
form_fields=INITREC_PERSON,
)
CUSTOM_WIDGETS: Final[dict[str, type[CustomWidget]]] = {
"grunderfassung_suche": Grunderfassung_SuchWidget,
"grunderfassung_unternehmen_suche": Grunderfassung_SuchWidget,
}
@@ -2203,51 +2206,53 @@ class Page_InitRecCompany(QWidget):
self.search_widget_trigger = cast(
QLineEdit, search_widget.company_widgets["ma_unternehmensname"]
)
text_widget_set = search_widgets_by_key(
self.auto_form.widget_registry, f"Kontaktperson{COLUMN_SEP}KP_name_partner"
)
assert len(text_widget_set) == 1
self.text_widget_set = cast(QLineEdit, text_widget_set[0]["widget"])
self.search_widget_trigger.textChanged.connect(
self._custom_set_company_name_contact_person
)
# TODO check how this should be done with new architecture
# text_widget_set = search_widgets_by_key(
# self.auto_form.widget_registry, f"Kontaktperson{COLUMN_SEP}KP_name_partner"
# )
# assert len(text_widget_set) == 1
# self.text_widget_set = cast(QLineEdit, text_widget_set[0]["widget"])
# self.search_widget_trigger.textChanged.connect(
# self._custom_set_company_name_contact_person
# )
# ** 'Bundesland' only if 'Inland' selected in 'Stammdaten'
person_location = search_widgets_by_key(
self.auto_form.widget_registry, f"Stammdaten{COLUMN_SEP}aufenthaltsort"
)
assert len(person_location) == 1
self.person_location = cast(QComboBox, person_location[0]["widget"])
assert isinstance(self.person_location, QComboBox)
self.person_location.currentIndexChanged.connect(self._custom_county_selection)
# TODO check removal
# person_location = search_widgets_by_key(
# self.auto_form.widget_registry, f"Stammdaten{COLUMN_SEP}aufenthaltsort"
# )
# assert len(person_location) == 1
# self.person_location = cast(QComboBox, person_location[0]["widget"])
# assert isinstance(self.person_location, QComboBox)
# self.person_location.currentIndexChanged.connect(self._custom_county_selection)
selection_county = search_widgets_by_key(
self.auto_form.widget_registry, f"Stammdaten{COLUMN_SEP}bundesland"
)
assert len(selection_county) == 1
self.selection_county = cast(QComboBox, selection_county[0]["widget"])
assert isinstance(self.selection_county, QComboBox)
self.selection_county.setProperty("styleClass", "stempel")
self.selection_county.setEnabled(False)
# selection_county = search_widgets_by_key(
# self.auto_form.widget_registry, f"Stammdaten{COLUMN_SEP}bundesland"
# )
# assert len(selection_county) == 1
# self.selection_county = cast(QComboBox, selection_county[0]["widget"])
# assert isinstance(self.selection_county, QComboBox)
# self.selection_county.setProperty("styleClass", "stempel")
# self.selection_county.setEnabled(False)
def _custom_set_company_name_contact_person(self, value: str) -> None:
self.text_widget_set.setText(str(value))
# def _custom_set_company_name_contact_person(self, value: str) -> None:
# self.text_widget_set.setText(str(value))
def _custom_county_selection(self, idx: int) -> None:
value = self.person_location.itemData(idx)
if value == "Inland":
self.selection_county.setEnabled(True)
self.selection_county.setProperty("styleClass", "")
self.selection_county.style().unpolish(self.selection_county)
self.selection_county.style().polish(self.selection_county)
self.selection_county.update()
else:
self.selection_county.setCurrentIndex(0)
self.selection_county.setEnabled(False)
self.selection_county.setProperty("styleClass", "stempel")
self.selection_county.style().unpolish(self.selection_county)
self.selection_county.style().polish(self.selection_county)
self.selection_county.update()
# def _custom_county_selection(self, idx: int) -> None:
# value = self.person_location.itemData(idx)
# if value == "Inland":
# self.selection_county.setEnabled(True)
# self.selection_county.setProperty("styleClass", "")
# self.selection_county.style().unpolish(self.selection_county)
# self.selection_county.style().polish(self.selection_county)
# self.selection_county.update()
# else:
# self.selection_county.setCurrentIndex(0)
# self.selection_county.setEnabled(False)
# self.selection_county.setProperty("styleClass", "stempel")
# self.selection_county.style().unpolish(self.selection_county)
# self.selection_county.style().polish(self.selection_county)
# self.selection_county.update()
def reset_form(self) -> None:
self.auto_form.reset_form()
@@ -2330,6 +2335,7 @@ class Page_InitRecPerson(QWidget):
# --- CUSTOM LOGIC ---
# ** fill 'Kontaktperson -> Namen Unternehmen'
# TODO check how partners are selected with new architecture
# search_res = search_widgets_by_key(self.auto_form.widget_registry, "Partnersuche")
# assert len(search_res) == 1
# search_widget = cast(Grunderfassung_SuchWidget, search_res[0]["widget"])
@@ -2628,7 +2634,7 @@ class Page_CompanyProfile(QWidget):
self.current_id = id_
return
init_rec = backend.initrec_get_initial_recording(id_)
init_rec = backend.initrec_company_get_initial_recording(id_)
ma_id = cast(ExtMaId, init_rec["Partnersuche__un_suche"])
assert ma_id is not None, "no ma_id obtained"
comp_info = backend.initrec_comp_search_get_info(ma_id)