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
+1054 -192
9 changed files with 1054 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit acfc4a824a - Show all commits
alembic/versions/8000d15faf03_major_change_in_data_model.py
+317
View File
@@ -0,0 +1,317 @@
"""major change in data model
Revision ID: 8000d15faf03
Revises: 665259cdfdc0
Create Date: 2026-07-24 15:13:57.443684
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
import wce_crm.db
# revision identifiers, used by Alembic.
revision: str = "8000d15faf03"
down_revision: Union[str, Sequence[str], None] = "665259cdfdc0"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("grunderfassung_personen")
op.drop_table("beratung_vorgang")
op.drop_table("grunderfassung_unternehmen")
with op.batch_alter_table("zuordnung_personen_unternehmen", schema=None) as batch_op:
batch_op.drop_index(
batch_op.f("uq_aktive_zuordnung"), sqlite_where=sa.text("gueltig_bis IS NULL")
)
op.drop_table("zuordnung_personen_unternehmen")
op.drop_table("beratung_einzelberatung")
op.create_table(
"Grunderfassung_Personen",
sa.Column("pers_id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("aktualisiert", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True),
sa.Column("Metadaten_nutzer", sa.String(length=20), nullable=True),
sa.Column("Metadaten_wiedereintrittsdatum", sa.Date(), nullable=True),
sa.Column("Arbeitserfahrung", sa.Text(), nullable=True),
sa.Column("Grunderfassung_fallnummer", sa.Text(), nullable=True),
sa.Column("Grunderfassung_notiz", sa.Text(), nullable=True),
sa.Column("HoehereBildung", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_adresse", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_anrede_anschrift", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_email", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_festnetznummer", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_funktion_beziehung", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_mobilfunknummer", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_name", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_name_partner", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_titel", sa.Text(), nullable=True),
sa.Column("Kontaktperson__KP_vorname", sa.Text(), nullable=True),
sa.Column("Projektrelevanz__relevanz", sa.Text(), nullable=True),
sa.Column("Projektrelevanz__foerderperiode", sa.Text(), nullable=True),
sa.Column("Schulbildung", sa.Text(), nullable=True),
sa.Column("Sprachkenntnisse", sa.Text(), nullable=True),
sa.Column("Stammdaten__PLZ", sa.Text(), nullable=True),
sa.Column("Stammdaten__anrede_anschrift", sa.Text(), nullable=True),
sa.Column("Stammdaten__anzahl_kinder__alter", sa.Text(), nullable=True),
sa.Column("Stammdaten__anzahl_kinder__anzahl", sa.Integer(), nullable=True),
sa.Column("Stammdaten__aufenthaltsort", sa.Text(), nullable=True),
sa.Column("Stammdaten__bundesland", sa.Text(), nullable=True),
sa.Column("Stammdaten__land", sa.Text(), nullable=True),
sa.Column("Stammdaten__email", sa.Text(), nullable=True),
sa.Column("Stammdaten__familienstand", sa.Text(), nullable=True),
sa.Column("Stammdaten__festnetznummer", sa.Text(), nullable=True),
sa.Column("Stammdaten__geburtsdatum", sa.Date(), nullable=True),
sa.Column("Stammdaten__hausnummer", sa.Text(), nullable=True),
sa.Column("Stammdaten__herkunftsland", sa.Text(), nullable=True),
sa.Column("Stammdaten__mobilfunknummer", sa.Text(), nullable=True),
sa.Column("Stammdaten__name", sa.Text(), nullable=True),
sa.Column("Stammdaten__ort", sa.Text(), nullable=True),
sa.Column("Stammdaten__rueckkehrer", sa.Boolean(), nullable=True),
sa.Column("Stammdaten__staatsangehoerigkeit", sa.Text(), nullable=True),
sa.Column("Stammdaten__strasse", sa.Text(), nullable=True),
sa.Column("Stammdaten__titel", sa.Text(), nullable=True),
sa.Column("Stammdaten__vorname", sa.Text(), nullable=True),
sa.Column("WeitereInfos__WI_arbeitsstatus", sa.Text(), nullable=True),
sa.Column("WeitereInfos__WI_aufenthaltstitel", sa.Text(), nullable=True),
sa.Column("WeitereInfos__WI_deutsch_sprache", sa.Text(), nullable=True),
sa.Column("WeitereInfos__WI_gueltigkeit_aufenthaltstitel", sa.Date(), nullable=True),
sa.Column("WeitereInfos__WI_meldung_institution", sa.Text(), nullable=True),
sa.PrimaryKeyConstraint("pers_id"),
)
op.create_table(
"Unternehmen",
sa.Column("un_id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True),
sa.PrimaryKeyConstraint("un_id"),
)
op.create_table(
"Beratung_Vorgang",
sa.Column("vorgang_id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("un_id", sa.Integer(), nullable=True),
sa.Column("pers_id", sa.Integer(), nullable=True),
sa.Column("titel", sa.Text(), nullable=False),
sa.Column("beratungs_typ", sa.String(length=20), nullable=False),
sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("aktualisiert", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True),
sa.ForeignKeyConstraint(
["pers_id"],
["Grunderfassung_Personen.pers_id"],
),
sa.ForeignKeyConstraint(
["un_id"],
["Unternehmen.un_id"],
),
sa.PrimaryKeyConstraint("vorgang_id"),
)
op.create_table(
"Personen_Unternehmen_Zuordnung",
sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("un_id", sa.Integer(), nullable=False),
sa.Column("an_id", sa.Integer(), nullable=False),
sa.Column("pers_id", sa.Integer(), nullable=False),
sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("aktualisiert", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True),
sa.ForeignKeyConstraint(
["pers_id"],
["Grunderfassung_Personen.pers_id"],
),
sa.ForeignKeyConstraint(
["un_id"],
["Unternehmen.un_id"],
),
sa.PrimaryKeyConstraint("id"),
)
with op.batch_alter_table("Personen_Unternehmen_Zuordnung", schema=None) as batch_op:
batch_op.create_index(
"uq_aktive_zuordnung",
["un_id", "pers_id"],
unique=True,
sqlite_where=sa.text("geloescht IS NULL"),
)
op.create_table(
"Beratung_Einzelberatung",
sa.Column("beratung_id", sa.Integer(), autoincrement=True, nullable=False),
sa.Column("vorgang_id", sa.Integer(), nullable=False),
sa.Column("nutzer_id", sa.Integer(), nullable=False),
sa.Column("nutzer_name", sa.String(length=20), nullable=False),
sa.Column("zeitstempel", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("ansprechpartner", sa.Text(), nullable=False),
sa.Column("kommunikationsweg", sa.Text(), nullable=False),
sa.Column("thema_crm_matrix", sa.Text(), nullable=True),
sa.Column("anmerkungen", sa.Text(), nullable=True),
sa.Column("rueckmeldung", sa.Text(), nullable=True),
sa.Column("erstellt", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("aktualisiert", wce_crm.db.UTCDateTime(), nullable=False),
sa.Column("geloescht", wce_crm.db.UTCDateTime(), nullable=True),
sa.ForeignKeyConstraint(
["vorgang_id"],
["Beratung_Vorgang.vorgang_id"],
),
sa.PrimaryKeyConstraint("beratung_id"),
)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("Beratung_Einzelberatung")
with op.batch_alter_table("Personen_Unternehmen_Zuordnung", schema=None) as batch_op:
batch_op.drop_index("uq_aktive_zuordnung", sqlite_where=sa.text("geloescht IS NULL"))
op.drop_table("Personen_Unternehmen_Zuordnung")
op.drop_table("Beratung_Vorgang")
op.drop_table("Unternehmen")
op.drop_table("Grunderfassung_Personen")
op.create_table(
"beratung_einzelberatung",
sa.Column("beratung_id", sa.INTEGER(), nullable=False),
sa.Column("vorgang_id", sa.INTEGER(), nullable=False),
sa.Column("nutzer_id", sa.INTEGER(), nullable=False),
sa.Column("nutzer_name", sa.VARCHAR(length=20), nullable=False),
sa.Column("zeitstempel", sa.DATETIME(), nullable=False),
sa.Column("ansprechpartner", sa.TEXT(), nullable=False),
sa.Column("kommunikationsweg", sa.TEXT(), nullable=False),
sa.Column("thema_crm_matrix", sa.TEXT(), nullable=True),
sa.Column("anmerkungen", sa.TEXT(), nullable=True),
sa.Column("rueckmeldung", sa.TEXT(), nullable=True),
sa.Column("erstellt", sa.DATETIME(), nullable=False),
sa.Column("aktualisiert", sa.DATETIME(), nullable=False),
sa.Column("geloescht", sa.DATETIME(), nullable=True),
sa.ForeignKeyConstraint(
["vorgang_id"],
["beratung_vorgang.vorgang_id"],
),
sa.PrimaryKeyConstraint("beratung_id"),
)
op.create_table(
"zuordnung_personen_unternehmen",
sa.Column("id", sa.INTEGER(), nullable=False),
sa.Column("un_id", sa.INTEGER(), nullable=False),
sa.Column("pers_id", sa.INTEGER(), nullable=False),
sa.Column("gueltig_ab", sa.DATETIME(), nullable=False),
sa.Column("gueltig_bis", sa.DATETIME(), nullable=True),
sa.ForeignKeyConstraint(
["pers_id"],
["grunderfassung_personen.pers_id"],
),
sa.ForeignKeyConstraint(
["un_id"],
["grunderfassung_unternehmen.un_id"],
),
sa.PrimaryKeyConstraint("id"),
)
with op.batch_alter_table("zuordnung_personen_unternehmen", schema=None) as batch_op:
batch_op.create_index(
batch_op.f("uq_aktive_zuordnung"),
["un_id", "pers_id"],
unique=1,
sqlite_where=sa.text("gueltig_bis IS NULL"),
)
op.create_table(
"grunderfassung_unternehmen",
sa.Column("un_id", sa.INTEGER(), nullable=False),
sa.Column("Metadaten_erstellung", sa.DATETIME(), nullable=True),
sa.Column("Metadaten_aktualisierung", sa.DATETIME(), nullable=True),
sa.Column("Metadaten_nutzer", sa.VARCHAR(length=20), nullable=True),
sa.Column("Grunderfassung_fallnummer", sa.TEXT(), nullable=True),
sa.Column("Grunderfassung_notiz", sa.TEXT(), nullable=True),
sa.Column("Partnersuche__kanal_aufmerksamkeit", sa.TEXT(), nullable=True),
sa.Column("Partnersuche__person_suche", sa.INTEGER(), nullable=True),
sa.Column("Partnersuche__un_suche", sa.INTEGER(), nullable=True),
sa.Column("geloescht", sa.DATETIME(), nullable=True),
sa.PrimaryKeyConstraint("un_id"),
)
op.create_table(
"beratung_vorgang",
sa.Column("vorgang_id", sa.INTEGER(), nullable=False),
sa.Column("un_id", sa.INTEGER(), nullable=True),
sa.Column("pers_id", sa.INTEGER(), nullable=True),
sa.Column("titel", sa.TEXT(), nullable=False),
sa.Column("beratungs_typ", sa.VARCHAR(length=20), nullable=False),
sa.Column("erstellt", sa.DATETIME(), nullable=False),
sa.Column("aktualisiert", sa.DATETIME(), nullable=False),
sa.Column("geloescht", sa.DATETIME(), nullable=True),
sa.ForeignKeyConstraint(
["pers_id"],
["grunderfassung_personen.pers_id"],
),
sa.ForeignKeyConstraint(
["un_id"],
["grunderfassung_unternehmen.un_id"],
),
sa.PrimaryKeyConstraint("vorgang_id"),
)
op.create_table(
"grunderfassung_personen",
sa.Column("pers_id", sa.INTEGER(), nullable=False),
sa.Column("Metadaten_erstellung", sa.DATETIME(), nullable=True),
sa.Column("Metadaten_aktualisierung", sa.DATETIME(), nullable=True),
sa.Column("Metadaten_nutzer", sa.VARCHAR(length=20), nullable=True),
sa.Column("Metadaten_wiedereintrittsdatum", sa.DATE(), nullable=True),
sa.Column("Arbeitserfahrung", sa.TEXT(), nullable=True),
sa.Column("Grunderfassung_fallnummer", sa.TEXT(), nullable=True),
sa.Column("Grunderfassung_notiz", sa.TEXT(), nullable=True),
sa.Column("HoehereBildung", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_adresse", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_anrede_anschrift", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_email", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_festnetznummer", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_funktion_beziehung", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_mobilfunknummer", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_name", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_name_partner", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_titel", sa.TEXT(), nullable=True),
sa.Column("Kontaktperson__KP_vorname", sa.TEXT(), nullable=True),
sa.Column("Projektrelevanz__relevanz", sa.TEXT(), nullable=True),
sa.Column("Projektrelevanz__foerderperiode", sa.TEXT(), nullable=True),
sa.Column("Schulbildung", sa.TEXT(), nullable=True),
sa.Column("Sprachkenntnisse", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__PLZ", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__anrede_anschrift", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__anzahl_kinder__alter", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__anzahl_kinder__anzahl", sa.INTEGER(), nullable=True),
sa.Column("Stammdaten__aufenthaltsort", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__bundesland", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__land", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__email", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__familienstand", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__festnetznummer", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__geburtsdatum", sa.DATE(), nullable=True),
sa.Column("Stammdaten__hausnummer", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__herkunftsland", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__mobilfunknummer", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__name", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__ort", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__rueckkehrer", sa.BOOLEAN(), nullable=True),
sa.Column("Stammdaten__staatsangehoerigkeit", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__strasse", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__titel", sa.TEXT(), nullable=True),
sa.Column("Stammdaten__vorname", sa.TEXT(), nullable=True),
sa.Column("WeitereInfos__WI_arbeitsstatus", sa.TEXT(), nullable=True),
sa.Column("WeitereInfos__WI_aufenthaltstitel", sa.TEXT(), nullable=True),
sa.Column("WeitereInfos__WI_deutsch_sprache", sa.TEXT(), nullable=True),
sa.Column("WeitereInfos__WI_gueltigkeit_aufenthaltstitel", sa.DATE(), nullable=True),
sa.Column("WeitereInfos__WI_meldung_institution", sa.TEXT(), nullable=True),
sa.Column("geloescht", sa.DATETIME(), nullable=True),
sa.PrimaryKeyConstraint("pers_id"),
)
# ### end Alembic commands ###
prototypes/dual_dict.py
+54
View File
@@ -0,0 +1,54 @@
# %%
from __future__ import annotations
import copy
from typing import Any
from dopt_basics.datastructures import DualDict
from wce_crm.data_models import FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
# %%
dd: DualDict[str, str] = DualDict(peter="test")
# %%
def rename_db_data(
db_data: dict[str, Any],
mapping: DualDict,
) -> dict[str, Any]:
new_db_data = copy.deepcopy(db_data)
for k, v in db_data.items():
if k in mapping:
new_db_data[mapping[k]] = v
del new_db_data[k]
if k in mapping.inverted:
new_db_data[mapping.inverted[k]] = v
del new_db_data[k]
return new_db_data
form_data = {
"Metadaten_erstellung": "t1",
"Metadaten_aktualisierung": "t2",
"Metadaten_nutzer": "t3",
}
db_data = rename_db_data(form_data, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN)
# %%
db_data
# %%
cvt_form_data = rename_db_data(db_data, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN)
cvt_form_data
# %%
dd["test"] = "peter"
# %%
dd["test"]
# %%
dd.inverted["peter"]
# %%
dd
# %%
prototypes/tests.py
+20 -20
View File
@@ -19,11 +19,11 @@ from PySide6.QtCore import QDate, Qt
from wce_crm import constants, db
from wce_crm.backend import backend
from wce_crm.types import NewEntityIds, RecordingType
from wce_crm.types import EntityIds, RecordingType
# %%
raw_ent_ids = NewEntityIds()
db_ent_ids = NewEntityIds(un_id=3, an_id=2)
raw_ent_ids = EntityIds()
db_ent_ids = EntityIds(un_id=3, an_id=2)
print(f"{raw_ent_ids=}\n{db_ent_ids=}")
print(f"{raw_ent_ids.empty()=}")
@@ -31,36 +31,36 @@ print(f"{raw_ent_ids.valid()=}")
# %%
raw_ent_ids.update(db_ent_ids)
print(f"{raw_ent_ids=}")
db_ent_ids = NewEntityIds(un_id=None, an_id=2, link_id=1)
db_ent_ids = EntityIds(un_id=None, an_id=2, link_id=1)
raw_ent_ids.update(db_ent_ids)
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty()=}")
print(f"{raw_ent_ids.valid()=}")
raw_ent_ids.update(NewEntityIds())
raw_ent_ids.update(EntityIds())
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty()=}")
print(f"{raw_ent_ids.valid()=}")
# %%
raw_ent_ids.update(NewEntityIds(un_id=None, pers_id=2))
raw_ent_ids.update(EntityIds(un_id=None, pers_id=2))
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty()=}")
print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}")
print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}")
raw_ent_ids.update(NewEntityIds(un_id=None, an_id=2, link_id=3, pers_id=2))
print(f"{raw_ent_ids.empty(RecordingType.WITH_COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.WITHOUT_COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.WITH_COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.WITHOUT_COMPANY)=}")
raw_ent_ids.update(EntityIds(un_id=None, an_id=2, link_id=3, pers_id=2))
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}")
print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}")
raw_ent_ids.update(NewEntityIds(un_id=2, pers_id=None, an_id=2, link_id=3))
print(f"{raw_ent_ids.empty(RecordingType.WITH_COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.WITHOUT_COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.WITH_COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.WITHOUT_COMPANY)=}")
raw_ent_ids.update(EntityIds(un_id=2, pers_id=None, an_id=2, link_id=3))
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}")
print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}")
print(f"{raw_ent_ids.empty(RecordingType.WITH_COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.WITHOUT_COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.WITH_COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.WITHOUT_COMPANY)=}")
# %%
db_path = constants.Config.DB_PATH_MAIN
src/wce_crm/backend/backend.py
+199 -6
View File
@@ -13,9 +13,12 @@ from sqlalchemy.dialects.sqlite import insert as sqlite_insert
from wce_crm import db
from wce_crm.constants import TIMEZONE_CEST
from wce_crm.data_models import (
FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN,
FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN,
Beratungsgespraech_Einzelgespraech,
Beratungsgespraech_Vorgang,
InitRec,
Page_InitRec_Form_Data,
)
from wce_crm.logging import logger_back as logger
from wce_crm.types import (
@@ -24,14 +27,17 @@ from wce_crm.types import (
CompanyProfileConsultations,
ConsultingType,
ContactPersonInfo,
EntityIds,
EntityType,
LinkingConsultationsEntryCompany,
LinkingConsultationsEntryPerson,
MainPageEntry,
NewEntityIds,
RecordingType,
)
if TYPE_CHECKING:
from dopt_basics.datastructures import DualDict
from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId
@@ -348,14 +354,201 @@ def initrec_comp_contact_person_search_get_info(
# return dict(row)
def _rename_db_data(
db_data: dict[str, Any],
mapping: DualDict,
) -> dict[str, Any]:
new_db_data = copy.deepcopy(db_data)
for k, v in db_data.items():
if k in mapping:
new_db_data[mapping[k]] = v
del new_db_data[k]
if k in mapping.inverted:
new_db_data[mapping.inverted[k]] = v
del new_db_data[k]
return new_db_data
def page_initrec_to_db(
data: InitRec,
) -> InitRec: ...
) -> InitRec:
with db.ENGINE.begin() as conn:
# person
dump_data = _rename_db_data(
data.form_data.person, 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,
)
stmt_compiled = str(stmt.compile(db.ENGINE))
logger.debug(
"[InitRec -- backend] Data to insert:\nStatement: %s\n%s",
stmt_compiled,
pformat(dump_data),
)
ret = conn.execute(stmt, dump_data)
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0]
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"]
logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully")
else:
logger.debug("[AutoForm -- backend] Update...")
stmt = (
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.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]
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,
)
# company
if data.recording_type is RecordingType.WITH_COMPANY:
# insert into company table and add additional information
assert data.ids.un_id, "company ID not set"
assert data.ids.an_id, "contact person (an) ID not set"
assert data.ids.pers_id, "person ID not set"
stmt = (
sqlite_insert(db.t_unternehmen)
.values(un_id=data.ids.un_id, geloescht=None)
.on_conflict_do_nothing(index_elements=["un_id"])
)
conn.execute(stmt)
# additional information
dump_data = _rename_db_data(
data.form_data.company, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
)
dump_data["geloescht"] = data.geloescht
stmt = sqlite_insert(db.t_zuordnung_personen_unternehmen).values(
**dump_data, pers_id=data.ids.pers_id
)
stmt = stmt.on_conflict_do_update(
index_elements=["un_id", "pers_id"],
index_where=sa.text("geloescht IS NULL"),
set_={
"an_id": stmt.excluded.an_id,
"Partnersuche__kanal_aufmerksamkeit": stmt.excluded.Partnersuche__kanal_aufmerksamkeit,
"aktualisiert": datetime.datetime.now(datetime.UTC),
},
).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.ids.link_id = data_from_db["id"]
return data
def page_initrec_from_db(
ids: NewEntityIds,
) -> InitRec: ...
ids: EntityIds,
) -> InitRec:
recording_type = RecordingType.WITHOUT_COMPANY
if 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"
if ids.link_id:
stmt = sa.select(
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,
).where(
db.t_zuordnung_personen_unternehmen.c.id == ids.link_id,
)
else:
stmt = sa.select(
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,
).where(
db.t_zuordnung_personen_unternehmen.c.pers_id == ids.pers_id,
db.t_zuordnung_personen_unternehmen.c.un_id == ids.un_id,
db.t_zuordnung_personen_unternehmen.c.geloescht.is_(None),
)
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_company_table = dict(from_db[0])
data_company_table = _rename_db_data(
data_person_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
)
form_data = Page_InitRec_Form_Data(
metadata={},
person=data_person_table,
company=data_company_table,
)
return InitRec(
ids=ids,
recording_type=recording_type,
geloescht=geloescht,
form_data=form_data,
)
# def initrec_company_delete_initial_recording(
@@ -857,7 +1050,7 @@ def _main_page_get_company_list() -> list[MainPageEntry]:
rec_id=rec_id,
display_name=display_name,
Metadaten_aktualisierung=datetime_akt,
type=RecordingType.COMPANY,
type=RecordingType.WITH_COMPANY,
)
)
@@ -894,7 +1087,7 @@ def _main_page_get_person_list() -> list[MainPageEntry]:
rec_id=rec_id,
display_name=display_name,
Metadaten_aktualisierung=datetime_akt,
type=RecordingType.PERSON,
type=RecordingType.WITHOUT_COMPANY,
)
)
src/wce_crm/data_models.py
+93 -19
View File
@@ -7,6 +7,7 @@ import json
from collections.abc import Iterable, Sequence
from typing import TYPE_CHECKING, Annotated, Any, Final, Generic, Protocol, TypeVar
from dopt_basics.datastructures import DualDict
from pydantic import (
AwareDatetime,
BaseModel,
@@ -18,12 +19,12 @@ from pydantic import (
)
from pydantic_core import ErrorDetails
from wce_crm.form_defs import FormField
from wce_crm.types import NewEntityIds
from wce_crm.form_defs import INITREC_COMP, INITREC_PERSON, FormField
from wce_crm.types import EntityIds, EntityType, RecordingType
if TYPE_CHECKING:
from wce_crm.gui import InitRecForm, Page_Consulting_ConsultingSession # noqa: F401
from wce_crm.types import ConsId, ConsultingType, RecId, RecordingType, UserId
from wce_crm.types import ConsId, ConsultingType, EntityType, RecId, UserId
ValidAge = Annotated[int, Field(ge=0, le=99)]
COLUMN_SEP: Final[str] = "__"
@@ -163,7 +164,7 @@ class InitRecFormToDb(Protocol):
class InitRecFormFromDb(Protocol):
def __call__(
self,
ids: NewEntityIds,
ids: EntityIds,
) -> InitRec: ...
@@ -171,6 +172,7 @@ class InitRecFormFromDb(Protocol):
def set_page_state(
target: S,
source: S,
ignore_attr: set[str] | None = None,
deep: bool = False,
) -> None:
"""
@@ -182,8 +184,12 @@ def set_page_state(
if type(target) is not type(source):
raise TypeError("Both source and target must have the same type")
if ignore_attr is None:
ignore_attr = set()
ignore_attr.add("child_modules")
for field in dc.fields(source):
if field.name == "child_modules":
if field.name in ignore_attr:
continue
new_value = getattr(source, field.name)
@@ -216,35 +222,63 @@ 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,)),
}
@dc.dataclass(slots=True, kw_only=True)
class Page_InitRec_Form_Data:
metadata: dict[str, Any]
person: dict[str, Any]
company: dict[str, Any]
# !! there must be exactly one InitRecForm for each recording type
@dc.dataclass(slots=True, kw_only=True)
class Page_InitRec_State(PageState["InitRecForm"]):
session: Session
ids: NewEntityIds = dc.field(default_factory=NewEntityIds)
record_type: RecordingType
ids: EntityIds = dc.field(default_factory=EntityIds)
recording_type: RecordingType
# un_id: RecId | None = None
locked: bool
geloescht: bool = False
form_data: dict[RecordingType, dict[str, Any]] = dc.field(default_factory=dict)
pydantic_models: dict[RecordingType, type[FlatBaseModel]] = dc.field(default_factory=dict)
form_data: Page_InitRec_Form_Data
initrec_form_states: dict[EntityType, Page_InitRec_Form_State] = dc.field(
init=False, default_factory=dict
)
# 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]
for ent_type in relevant_entities:
initrec_form_state = Page_InitRec_Form_State(
session=self.session,
cfg=INITREC_CONFIGS[ent_type],
ids=self.ids,
ent_type=ent_type,
locked=self.locked,
geloescht=self.geloescht,
)
self.initrec_form_states[ent_type] = initrec_form_state
@dc.dataclass(slots=True, kw_only=True)
class Page_InitRec_Form_Config:
pydantic_model: type[FlatBaseModel]
# to_db: InitRecFormToDb # TODO remove
# from_db: InitRecFormFromDb # TODO remove
form_fields: Sequence[FormField]
ignored_keys: Iterable[str] = tuple()
add_buttons: bool = False
id_mapping: dict[str, str]
@dc.dataclass(slots=True, kw_only=True)
class Page_InitRec_Form_State(PageState[Module]):
session: Session
cfg: Page_InitRec_Form_Config
ids: NewEntityIds
recording_type: RecordingType
ids: EntityIds
ent_type: EntityType
# rec_id: RecId | None = None
form_data: dict[str, Any] = dc.field(default_factory=dict)
locked: bool
@@ -269,7 +303,6 @@ class Page_InitRec_Form_State(PageState[Module]):
# locked: bool
# TODO change to entity IDs
@dc.dataclass(slots=True, kw_only=True)
class Page_Consulting_State(PageState[Module]):
session: Session
@@ -285,7 +318,6 @@ class Page_Consulting_State(PageState[Module]):
locked: bool
# TODO change to entity IDs
@dc.dataclass(slots=True, kw_only=True)
class Page_Consulting_Linking_State(PageState[Module]):
session: Session
@@ -322,7 +354,7 @@ class Page_Consulting_Table_State(PageState["Page_Consulting_ConsultingSession"]
@dc.dataclass(slots=True, kw_only=True)
class Page_CompanyProfile_State(PageState[Module]):
session: Session
ids: NewEntityIds
ids: EntityIds
# rec_id: RecId | None = None
@@ -441,10 +473,21 @@ class FlatBaseModel(BaseModel):
# ** InitRec
class InitRec(BaseModel):
ids: NewEntityIds
# rec_id: RecId | None
ids: EntityIds
recording_type: RecordingType
geloescht: AwareDatetime | None = None
db_data: dict[RecordingType, dict[str, Any]]
form_data: Page_InitRec_Form_Data
# mapping between initrec form field keys and database properties
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",
)
class Grunderfassung_Unternehmen(FlatBaseModel):
@@ -649,3 +692,34 @@ class Beratungsgespraech_Einzelgespraech(BaseModel):
erstellt: AwareDatetime | None = Field(default=None, exclude=True)
aktualisiert: AwareDatetime | None = Field(default=None, exclude=True) # see above
geloescht: AwareDatetime | None = None
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",
),
form_fields=INITREC_COMP,
add_buttons=False,
id_mapping={
"Partnersuche__un_suche": "un_id",
"Partnersuche__person_suche": "an_id",
},
),
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",
),
form_fields=INITREC_PERSON,
add_buttons=False,
id_mapping={},
),
}
src/wce_crm/db.py
+4 -3
View File
@@ -490,6 +490,7 @@ t_zuordnung_personen_unternehmen: Table = Table(
UTCDateTime,
nullable=True,
),
Column("Partnersuche__kanal_aufmerksamkeit", sa.Text, nullable=True),
# Column(
# "gueltig_ab",
# UTCDateTime,
@@ -520,8 +521,8 @@ t_beratung_vorgang: Table = Table(
primary_key=True,
autoincrement=True,
),
Column("un_id", sa.ForeignKey("grunderfassung_unternehmen.un_id"), nullable=True),
Column("pers_id", sa.ForeignKey("grunderfassung_personen.pers_id"), nullable=True),
Column("un_id", sa.ForeignKey("Unternehmen.un_id"), nullable=True),
Column("pers_id", sa.ForeignKey("Grunderfassung_Personen.pers_id"), nullable=True),
Column("titel", sa.Text, nullable=False),
Column("beratungs_typ", sa.String(20), nullable=False),
Column(
@@ -554,7 +555,7 @@ t_beratung_einzelberatung: Table = Table(
primary_key=True,
autoincrement=True,
),
Column("vorgang_id", sa.ForeignKey("beratung_vorgang.vorgang_id"), nullable=False),
Column("vorgang_id", sa.ForeignKey("Beratung_Vorgang.vorgang_id"), nullable=False),
Column("nutzer_id", sa.Integer, nullable=False),
Column("nutzer_name", sa.String(20), nullable=False),
Column(
src/wce_crm/form_defs.py
+36 -36
View File
@@ -889,42 +889,42 @@ INITREC_LANGUAGES = [
INITREC_COMP = [
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=True,
),
FormField(
"Fallnummer",
FormFieldType.TEXT,
required=True,
key="Grunderfassung_fallnummer",
),
FormField(
"Notizen",
FormFieldType.LONGTEXT,
required=False,
key="Grunderfassung_notiz",
),
# 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=True,
# ),
# FormField(
# "Fallnummer",
# FormFieldType.TEXT,
# required=True,
# key="Grunderfassung_fallnummer",
# ),
# FormField(
# "Notizen",
# FormFieldType.LONGTEXT,
# required=False,
# key="Grunderfassung_notiz",
# ),
FormField(
"Suche",
FormFieldType.CUSTOM,
src/wce_crm/gui.py
+295 -95
View File
@@ -85,17 +85,16 @@ from wce_crm.backend import backend
from wce_crm.constants import TIMEZONE_CEST
from wce_crm.data_models import (
COLUMN_SEP,
INITREC_CONFIGS,
Beratungsgespraech_Einzelgespraech,
Beratungsgespraech_Vorgang,
Grunderfassung_Personen,
Grunderfassung_Unternehmen,
InitRec,
Page_CompanyProfile_State,
Page_Consulting_ConsultingSession_State,
Page_Consulting_Linking_State,
Page_Consulting_State,
Page_Consulting_Table_State,
Page_InitRec_Form_Config,
Page_InitRec_Form_Data,
Page_InitRec_Form_State,
Page_InitRec_State,
Page_MainPage_State,
@@ -105,8 +104,6 @@ from wce_crm.data_models import (
translate_pydantic_errors,
)
from wce_crm.form_defs import (
INITREC_COMP,
INITREC_PERSON,
FormField,
FormFieldType,
)
@@ -121,7 +118,8 @@ from wce_crm.logging import (
from wce_crm.types import (
CompanyProfileConsultationEntry,
ConsultingType,
NewEntityIds,
EntityIds,
EntityType,
RecId,
RecordingType,
)
@@ -1547,7 +1545,7 @@ class InitRecForm(QWidget):
# self.save_btn.setText(self.edit_buttons.save_btn_txt_enabled)
def relevant_ids_set(self) -> bool:
return self.STATE.ids.valid(self.STATE.recording_type)
return self.STATE.ids.valid(self.STATE.ent_type)
# def _activate_delete(self) -> None:
# if self.STATE.ids.valid():
@@ -1613,12 +1611,23 @@ class InitRecForm(QWidget):
# pformat(self.STATE.form_data),
# )
def _load_state_form_data(self) -> None:
logger_initrecform.info("[InitRecForm] Load data method...")
logger_initrecform.debug(
"[InitRecForm] Loaded data dict:\n%s Passing to Pydantic...",
pformat(self.STATE.form_data),
)
# TODO error catching?
model = self.cfg.pydantic_model(**self.STATE.form_data)
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:
reset_form(self.widget_registry)
# self.STATE.ids.set_id(None, self.STATE.ent_type)
self.STATE.ids.update(NewEntityIds())
# TODO remove
# self.STATE.rec_id = None
self.STATE.ids.update(EntityIds())
self.STATE.form_data.clear()
self.STATE.geloescht = False
@@ -1658,7 +1667,7 @@ class InitRecForm(QWidget):
# !! loading from IDs is only performed on top-level modules aka pages
if section is None or section == "form":
if not self.STATE.geloescht:
self.load_from_ids(self.STATE.ids)
self._load_state_form_data()
else:
self.reset_form()
@@ -1678,6 +1687,10 @@ class InitRecForm(QWidget):
# TODO change usage of user ID (needs changes in model and backend)
if self.SESSION_USER_KEY in form_data:
form_data[self.SESSION_USER_KEY] = self.STATE.session.user_name
for form_key, id_key in self.STATE.cfg.id_mapping.items():
self.STATE.ids.set_by_name(id_key, form_data[form_key])
self.STATE.form_data = form_data
def validate(self) -> list[str]:
@@ -1695,18 +1708,18 @@ class InitRecForm(QWidget):
return self.STATE
def _load_from_pydantic(
self,
data: InitRec,
) -> None:
# ent_type = self.STATE.recording_type
# self.STATE.ids.set_id(data.ids.get_id(ent_type), ent_type)
self.STATE.ids.update(data.ids)
# self.STATE.rec_id = data.rec_id
self.STATE.form_data.clear()
self.STATE.geloescht = True if data.geloescht else False
# def _load_from_pydantic(
# self,
# data: InitRec,
# ) -> None:
# # ent_type = self.STATE.recording_type
# # self.STATE.ids.set_id(data.ids.get_id(ent_type), ent_type)
# self.STATE.ids.update(data.ids)
# # self.STATE.rec_id = data.rec_id
# self.STATE.form_data.clear()
# self.STATE.geloescht = True if data.geloescht else False
self._sync_state_to_GUI(None) # TODO change to correct loading
# self._sync_state_to_GUI(None) # TODO change to correct loading
# !! must be part of the managing page
def save_data(self) -> None:
@@ -2979,7 +2992,8 @@ class Page_NewEntry(QWidget):
def _request_new_initrec(self) -> None:
req_state = Page_InitRec_State(
session=self.STATE.session,
record_type=RecordingType.COMPANY,
recording_type=RecordingType.WITH_COMPANY,
form_data=Page_InitRec_Form_Data(metadata={}, person={}, company={}),
locked=False,
)
logger_gui.debug("[Page -- NewInitRec] State to call: %s", req_state)
@@ -3009,29 +3023,29 @@ class Page_NewEntry(QWidget):
pass
CONFIG_GRUNDERFASSUNG_UNTERNEHMEN: Final[Page_InitRec_Form_Config] = 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",
),
form_fields=INITREC_COMP,
add_buttons=False,
)
# CONFIG_GRUNDERFASSUNG_UNTERNEHMEN: Final[Page_InitRec_Form_Config] = 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",
# ),
# form_fields=INITREC_COMP,
# add_buttons=False,
# )
CONFIG_GRUNDERFASSUNG_PERSONEN: Final[Page_InitRec_Form_Config] = 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",
),
form_fields=INITREC_PERSON,
add_buttons=False,
)
# CONFIG_GRUNDERFASSUNG_PERSONEN: Final[Page_InitRec_Form_Config] = 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",
# ),
# form_fields=INITREC_PERSON,
# add_buttons=False,
# )
CUSTOM_WIDGETS: Final[dict[str, type[CustomFormWidget]]] = {
"grunderfassung_unternehmen_suche": Grunderfassung_SuchWidget,
@@ -3128,12 +3142,12 @@ class Page_InitRec(QWidget):
back_btn_step.setMinimumWidth(200)
back_btn_step.setMaximumWidth(200)
if self.STATE.record_type is RecordingType.COMPANY:
if self.STATE.recording_type is RecordingType.WITH_COMPANY:
title = QLabel("Grunderfassung Unternehmen")
elif self.STATE.record_type is RecordingType.PERSON:
elif self.STATE.recording_type is RecordingType.WITHOUT_COMPANY:
title = QLabel("Grunderfassung Individualperson")
else:
raise RuntimeError(f"Unspecified recording type {self.STATE.record_type}")
raise RuntimeError(f"Unspecified recording type {self.STATE.recording_type}")
title.setStyleSheet("font-size: 20px; font-weight: bold;")
header_layout.setSpacing(5)
@@ -3142,6 +3156,64 @@ class Page_InitRec(QWidget):
header_layout.addWidget(title)
vert_layout.addWidget(header_container)
if DEBUG:
separator1 = QFrame()
separator1.setFrameShape(QFrame.Shape.HLine)
separator1.setFrameShadow(QFrame.Shadow.Sunken)
header_layout.addWidget(separator1)
btn_gui2state = QPushButton("Sync GUI to state")
btn_gui2state.clicked.connect(self._sync_GUI_to_state)
btn_gui2state.setFixedHeight(35)
btn_gui2state.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
)
header_layout.addWidget(btn_gui2state)
btn_state2gui = QPushButton("Sync state to GUI")
btn_state2gui.clicked.connect(self._sync_state_to_GUI)
btn_state2gui.setFixedHeight(35)
btn_state2gui.setSizePolicy(
QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
)
header_layout.addWidget(btn_state2gui)
# self.test_button = QPushButton("Initialisiere Laden")
# self.test_button.clicked.connect(self._load_from_id)
# self.test_button.setFixedHeight(35)
# self.test_button.setSizePolicy(
# QSizePolicy.Policy.Expanding, QSizePolicy.Policy.Fixed
# )
# header_layout.addWidget(self.test_button)
# button_get = QPushButton("GET DATA")
# button_get.setFixedHeight(35)
# button_get.clicked.connect(self._get_form_data)
# header_layout.addWidget(button_get)
# button_set = QPushButton("SET DATA")
# button_set.setFixedHeight(35)
# button_set.clicked.connect(self._set_form_data)
# header_layout.addWidget(button_set)
# id_field_layout = QHBoxLayout()
# id_field_layout.setContentsMargins(0, 0, 0, 0)
# id_field_layout.setSpacing(5)
# id_field_label = QLabel("ID Datenbank:")
# self.id_field_input = QLineEdit()
# id_field_layout.addWidget(id_field_label)
# id_field_layout.addWidget(self.id_field_input)
# header_layout.addLayout(id_field_layout)
# button_db_index = QPushButton("Setze DB Index")
# button_db_index.setFixedHeight(35)
# button_db_index.clicked.connect(self._set_db_index)
# header_layout.addWidget(button_db_index)
separator2 = QFrame()
separator2.setFrameShape(QFrame.Shape.HLine)
separator2.setFrameShadow(QFrame.Shadow.Sunken)
header_layout.addWidget(separator2)
header_layout.addSpacing(5)
# --- BANNER ---
self.info_banners_container = QWidget()
self.info_banners: list[InfoBanner] = []
@@ -3183,35 +3255,29 @@ class Page_InitRec(QWidget):
vert_layout.addWidget(scroll_area)
container_layout = QVBoxLayout(container)
container_layout.setContentsMargins(0, 0, 0, 0)
self.container_layout = QVBoxLayout(container)
self.container_layout.setContentsMargins(0, 0, 0, 0)
# --- AUTO FORM LAYOUT 1 ---
# --- AUTO FORM LAYOUT ---
# now we try to render the layout within the collapsable box
container_layout.addSpacing(20)
self.container_layout.addSpacing(20)
self.initrec_form_company: InitRecForm | None = None
if self.STATE.record_type is RecordingType.COMPANY:
# add company
pass
# ** person
self.initrec_form_person_box = CollapsibleBox("Angaben zur Person")
ent_type = EntityType.PERSON
self.initrec_form_person = InitRecForm(self.STATE.initrec_form_states[ent_type])
assert self.STATE.initrec_form_states[ent_type] is self.initrec_form_person.STATE
box1 = CollapsibleBox("Angaben zur Person")
initrec_form_state_person = Page_InitRec_Form_State(
session=self.STATE.session,
cfg=CONFIG_GRUNDERFASSUNG_PERSONEN,
ids=self.STATE.ids,
recording_type=RecordingType.PERSON,
locked=False,
)
self.initrec_form_person = InitRecForm(initrec_form_state_person)
self.STATE.child_modules.append(self.initrec_form_person)
box1.setContentWidget(self.initrec_form_person)
container_layout.addWidget(box1)
self.initrec_form_person_box.setContentWidget(self.initrec_form_person)
self.container_layout.addWidget(self.initrec_form_person_box)
# container_layout.addSpacing(15)
self.initrec_form_person.update_triggered.connect(
lambda: self.update_triggered.emit()
)
self.initrec_form_person.update_triggered.connect(self._auto_form_updated)
# TODO trigger needed?
# self.initrec_form_person.update_triggered.connect(
# lambda: self.update_triggered.emit()
# )
# self.initrec_form_person.update_triggered.connect(self._auto_form_updated)
# --- CUSTOM LOGIC ---
# ** 'Bundesland' only if 'Inland' selected in 'Stammdaten'
@@ -3233,7 +3299,68 @@ class Page_InitRec(QWidget):
self.selection_county.setProperty("styleClass", "stempel")
self.selection_county.setEnabled(False)
container_layout.addStretch()
# ** company
ent_type = EntityType.COMPANY
self.initrec_form_company_box = CollapsibleBox("Angaben zum Unternehmen")
self.initrec_form_company = InitRecForm(self.STATE.initrec_form_states[ent_type])
self.STATE.child_modules.append(self.initrec_form_company)
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
self.initrec_form_company_box.setContentWidget(self.initrec_form_company)
self.container_layout.insertWidget(target_idx, self.initrec_form_company_box)
# container_layout.addSpacing(15)
# TODO trigger needed?
# self.initrec_form_person.update_triggered.connect(
# lambda: self.update_triggered.emit()
# )
# self.initrec_form_person.update_triggered.connect(self._auto_form_updated)
# --- CUSTOM LOGIC ---
# ** fill 'Kontaktperson -> Namen Unternehmen'
search_res = search_widgets_by_key(
self.initrec_form_company.widget_registry, "Partnersuche"
)
assert len(search_res) == 1
search_widget = cast(Grunderfassung_SuchWidget, search_res[0]["widget"])
self.search_widget_trigger = cast(
QLineEdit, search_widget.company_widgets["ma_unternehmensname"]
)
self.container_layout.addStretch()
self._update_layout()
def _update_layout(self) -> None:
if self.STATE.recording_type is RecordingType.WITH_COMPANY:
# add company
self.initrec_form_company_box.setVisible(True)
else:
self.initrec_form_company_box.setVisible(False)
# elif self.initrec_form_company_box is not None:
# self.container_layout.removeWidget(self.initrec_form_company_box)
# # ** state
# assert self.initrec_form_company
# self.STATE.child_modules.remove(self.initrec_form_company)
# del self.STATE.initrec_form_states[EntityType.COMPANY]
# self.initrec_form_company_box = None
# self.initrec_form_company = None
# def _set_new_initrec_state(
# self,
# ent_type: EntityType,
# ) -> None:
# initrec_form_state = Page_InitRec_Form_State(
# session=self.STATE.session,
# cfg=INITREC_CONFIGS[ent_type],
# ids=self.STATE.ids,
# ent_type=ent_type,
# locked=False,
# )
# self.STATE.initrec_form_states[ent_type] = initrec_form_state
# ** custom logic
def _custom_county_selection(self, idx: int) -> None:
@@ -3328,18 +3455,28 @@ class Page_InitRec(QWidget):
# TODO staes
for form in self.STATE.child_modules:
form.reset_form()
self.STATE.ids = NewEntityIds()
self.STATE.ids = EntityIds()
self.STATE.geloescht = False
def _load_database(
self,
ids: NewEntityIds,
ids: EntityIds,
) -> None:
logger_initrecform.info("[Page -- InitRec] Load data method...")
logger_initrecform.debug("[Page -- InitRec] Entity IDs: %s", ids)
self._reset()
# TODO add loading logic
initrec: InitRec = ...
# 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={},
),
)
self._load_from_pydantic(initrec, sync_to_GUI=False)
@@ -3349,7 +3486,23 @@ class Page_InitRec(QWidget):
update_only: bool = False,
sync_to_GUI: bool = True,
) -> None:
logger_initrecform.info("[Page -- InitRec] Load from Pydantic model...")
# TODO add logic to update internal state
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():
match ent_type:
case EntityType.COMPANY:
state.form_data = self.STATE.form_data.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)
case _:
raise RuntimeError(f"Unspecified entity type {ent_type}")
if update_only:
...
@@ -3368,19 +3521,20 @@ class Page_InitRec(QWidget):
)
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,
recording_type=RecordingType.PERSON,
locked=False,
)
self.initrec_form_person.load_state(initrec_form_state_person)
# 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)
if section is None or section == "banners":
self._clear_info_banners()
@@ -3401,22 +3555,65 @@ class Page_InitRec(QWidget):
form.unlock()
def _sync_GUI_to_state(self) -> None:
logger_page_initrec.debug("[Page -- InitRec] Syncing GUI to state...")
logger_page_initrec.debug(
"[Page -- InitRec] Syncing GUI to state: Child modules...\n%s",
pformat(self.STATE.child_modules),
)
logger_page_initrec.debug(
"[Page -- InitRec] Syncing GUI to state: Existing form states keys...\n%s",
pformat(self.STATE.initrec_form_states.keys()),
)
for form in self.STATE.child_modules:
# ?? is this the same reference to use a similar pattern as
# ?? the consulting page (initialise once with the same state
# ?? as reference which are then automatically updated)
# !! check if this is working as expected
# new entries' states are linked to this instance by the page state
form_state = form.get_state()
self.STATE.ids.update(form_state.ids, ignore_none=True)
self.STATE.form_data[form_state.recording_type] = form_state.form_data
self.STATE.pydantic_models[form_state.recording_type] = (
form_state.cfg.pydantic_model
logger_page_initrec.debug(
"[Page -- InitRec] Syncing GUI to state: Form state ent_type %s",
form_state.ent_type,
)
logger_page_initrec.debug(
"[Page -- InitRec] Syncing GUI to state: identity of form states: %s",
(self.STATE.initrec_form_states[form_state.ent_type] is form_state),
)
assert self.STATE.initrec_form_states[form_state.ent_type] is form_state, (
"passing by reference not working"
)
self.STATE.ids.update(form_state.ids, ignore_none=True)
# self.STATE.form_data[form_state.ent_type] = form_state.form_data
match form_state.ent_type:
case EntityType.COMPANY:
self.STATE.form_data.company = form_state.form_data
case EntityType.PERSON:
self.STATE.form_data.person = form_state.form_data
case _:
raise RuntimeError(f"Unspecified entity type {form_state.ent_type}")
logger_page_initrec.debug(
"[Page -- InitRec] Syncing GUI to state: IDs after...\n%s",
pformat(self.STATE.ids),
)
logger_page_initrec.debug(
"[Page -- InitRec] Syncing GUI to state: Form data after...\n%s",
pformat(self.STATE.form_data),
)
def load_state(
self,
new_state: Page_InitRec_State,
) -> None:
set_page_state(self.STATE, new_state)
set_page_state(
self.STATE,
new_state,
ignore_attr={"form_data", "initrec_form_states"},
)
self._sync_state_to_GUI(None)
def get_state(self) -> Page_InitRec_State:
@@ -3444,6 +3641,8 @@ class Page_InitRec(QWidget):
def save_data(self) -> None:
logger_page_initrec.debug("[Page -- InitRec] Save button clicked...")
return
if self.STATE.ids.empty() and self.STATE.geloescht:
logger_page_initrec.debug(
"[Page -- InitRec] This was never saved in the database "
@@ -3460,7 +3659,7 @@ class Page_InitRec(QWidget):
to_db = InitRec(
ids=self.STATE.ids,
geloescht=deleted_datetime,
db_data={},
form_data={},
)
returned = self.cfg.to_db(to_db)
self._load_from_pydantic(returned)
@@ -3520,7 +3719,7 @@ class Page_InitRec(QWidget):
ids=self.STATE.ids,
# rec_id=self.STATE.rec_id,
geloescht=deleted_datetime,
db_data=db_data,
form_data=db_data,
)
returned = self.cfg.to_db(to_db)
@@ -5939,8 +6138,9 @@ class MainWindow(QMainWindow):
# SITE: 'Grunderfassung'
newinitrec_state = Page_InitRec_State(
session=session,
ids=NewEntityIds(),
record_type=RecordingType.COMPANY,
ids=EntityIds(),
recording_type=RecordingType.WITH_COMPANY,
form_data=Page_InitRec_Form_Data(metadata={}, person={}, company={}),
locked=False,
)
self.newinitrec = Page_InitRec(newinitrec_state)
@@ -5973,7 +6173,7 @@ class MainWindow(QMainWindow):
# self.stack.addWidget(self.initrec_person)
# SITE: 'Unternehmensprofil'
comp_profile_state = Page_CompanyProfile_State(
session=self.STATE.session, ids=NewEntityIds()
session=self.STATE.session, ids=EntityIds()
)
self.company_profile = Page_CompanyProfile(comp_profile_state)
self.company_profile.back_main_requested.connect(self.show_main_page)
src/wce_crm/types.py
+36 -13
View File
@@ -5,7 +5,7 @@ import dataclasses as dc
import datetime
import enum
from collections.abc import Generator
from typing import TYPE_CHECKING, TypeAlias, TypedDict
from typing import TYPE_CHECKING, Final, TypeAlias, TypedDict
if TYPE_CHECKING:
import polars as pl
@@ -24,6 +24,12 @@ PolarsSchema: TypeAlias = dict[str, type["pl.DataType"]]
# TODO rename: these are not entity types or entities but different forms of
# initial recordings
class RecordingType(enum.IntEnum):
WITH_COMPANY = enum.auto()
WITHOUT_COMPANY = enum.auto()
class EntityType(enum.IntEnum):
METADATA = enum.auto()
COMPANY = enum.auto()
PERSON = enum.auto()
@@ -34,19 +40,36 @@ class ConsultingType(enum.StrEnum):
@dc.dataclass(slots=True, kw_only=True)
class NewEntityIds:
class EntityIds:
un_id: ExtMaId | None = None
an_id: ExtAnId | None = None
pers_id: RecId | None = None
link_id: IntLinkId | None = None
_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))
def set_by_name(
self,
attr_name: str,
value: int | None,
) -> None:
if attr_name not in self._field_names:
raise KeyError(
f"Attribute {attr_name} not member of this class. Valid "
f"options: {self._field_names}"
)
setattr(self, attr_name, value)
def update(
self,
ids: NewEntityIds,
ids: EntityIds,
ignore_none: bool = False,
deep: bool = False,
) -> None:
if not isinstance(ids, NewEntityIds):
if not isinstance(ids, EntityIds):
raise TypeError("New IDs must be have the same type as this object")
for field in dc.fields(ids):
@@ -62,18 +85,18 @@ class NewEntityIds:
def _get_relevant_fields(
self,
recording_type: RecordingType | None = None,
ent_type: EntityType | None = None,
) -> Generator[int | None, None, None]:
if recording_type is None:
if ent_type is None:
relevant_fields = (getattr(self, field.name) for field in dc.fields(self))
elif recording_type is RecordingType.COMPANY:
elif ent_type is EntityType.COMPANY:
RELEVANT_ATTRIBUTES = {"un_id", "an_id", "link_id"}
relevant_fields = (
getattr(self, field.name)
for field in dc.fields(self)
if field.name in RELEVANT_ATTRIBUTES
)
elif recording_type is RecordingType.PERSON:
elif ent_type is EntityType.PERSON:
RELEVANT_ATTRIBUTES = {"pers_id"}
relevant_fields = (
getattr(self, field.name)
@@ -81,23 +104,23 @@ class NewEntityIds:
if field.name in RELEVANT_ATTRIBUTES
)
else:
raise RuntimeError(f"Unspecified recording type {recording_type}")
raise RuntimeError(f"Unspecified entity type {ent_type}")
return relevant_fields
def empty(
self,
recording_type: RecordingType | None = None,
ent_type: EntityType | None = None,
) -> bool:
relevant_fields = self._get_relevant_fields(recording_type=recording_type)
relevant_fields = self._get_relevant_fields(ent_type=ent_type)
return not any(relevant_fields)
def valid(
self,
recording_type: RecordingType | None = None,
ent_type: EntityType | None = None,
) -> bool:
relevant_fields = self._get_relevant_fields(recording_type=recording_type)
relevant_fields = self._get_relevant_fields(ent_type=ent_type)
return all(relevant_fields)