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

Fix data model #4

Merged
foefl merged 13 commits from fix_data_model into dev 2026-07-28 17:04:13 +00:00
Conversation 0 Commits 13 Files Changed 13
+2741 -1535
13 changed files with 2741 additions and 1535 deletions
Download Patch File Download Diff File Expand all files Collapse all files
alembic/versions/5dfa1fc1dfdf_update_schema_to_meet_customer_.py
+48
View File
@@ -0,0 +1,48 @@
"""update schema to meet customer requirements
Revision ID: 5dfa1fc1dfdf
Revises: e290fe18132d
Create Date: 2026-07-28 08:39:44.168872
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "5dfa1fc1dfdf"
down_revision: Union[str, Sequence[str], None] = "e290fe18132d"
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! ###
with op.batch_alter_table("Grunderfassung_Personen", schema=None) as batch_op:
batch_op.add_column(
sa.Column("Stammdaten__kanal_aufmerksamkeit", sa.Text(), nullable=True)
)
with op.batch_alter_table("Personen_Unternehmen_Zuordnung", schema=None) as batch_op:
batch_op.add_column(sa.Column("Partnersuche__kontaktweg", sa.Text(), nullable=True))
batch_op.drop_column("Partnersuche__kanal_aufmerksamkeit")
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table("Personen_Unternehmen_Zuordnung", schema=None) as batch_op:
batch_op.add_column(
sa.Column("Partnersuche__kanal_aufmerksamkeit", sa.TEXT(), nullable=True)
)
batch_op.drop_column("Partnersuche__kontaktweg")
with op.batch_alter_table("Grunderfassung_Personen", schema=None) as batch_op:
batch_op.drop_column("Stammdaten__kanal_aufmerksamkeit")
# ### end Alembic commands ###
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 ###
alembic/versions/e290fe18132d_update_outstanding_table_changes.py
+38
View File
@@ -0,0 +1,38 @@
"""update outstanding table changes
Revision ID: e290fe18132d
Revises: 8000d15faf03
Create Date: 2026-07-27 09:05:06.301043
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "e290fe18132d"
down_revision: Union[str, Sequence[str], None] = "8000d15faf03"
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! ###
with op.batch_alter_table("Personen_Unternehmen_Zuordnung", schema=None) as batch_op:
batch_op.add_column(
sa.Column("Partnersuche__kanal_aufmerksamkeit", sa.Text(), nullable=True)
)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table("Personen_Unternehmen_Zuordnung", schema=None) as batch_op:
batch_op.drop_column("Partnersuche__kanal_aufmerksamkeit")
# ### end Alembic commands ###
data_model/project_db_schema.aml
+27 -25
View File
@@ -1,30 +1,30 @@
Ext_CRM.Master
Ext_Kontaktliste.Master
ma_id int pk
Ext_CRM.Ansprechpartner
Ext_Kontaktliste.Ansprechpartner
an_id int pk
ma_id int -> Ext_CRM.Master(ma_id)
ma_id int -> Ext_Kontaktliste.Master(ma_id)
Grunderfassung_Unternehmen
un_id int pk {autoIncrement}
Unternehmen | Es wird nur die ID für die Kontaktliste referenziert. Alle relevanten Infos werden live aus der Kontaktliste geladen via 'ATTACH'.
un_id int pk -> Ext_Kontaktliste.Master(ma_id)
erstellt "timestamp with time zone"
geloescht "timestamp with time zone" nullable
Metadaten_erstellung "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` nullable
Metadaten_aktualisierung "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` nullable {on_update: `datetime.datetime.now(datetime.UTC)`}
Metadaten_nutzer varchar(20) nullable
Grunderfassung_fallnummer text nullable
Grunderfassung_notiz text nullable
Partnersuche__un_suche int nullable -> Ext_CRM.Master(ma_id)
Partnersuche__person_suche int nullable -> Ext_CRM.Ansprechpartner(an_id)
Partnersuche__kanal_aufmerksamkeit text nullable
Personen_Unternehmen_Zuordnung
id int pk {autoIncrement}
pers_id int pk -> Grunderfassung_Personen(pers_id) unique=uq_aktive_zuordnung
un_id int -> Unternehmen(un_id) unique=uq_aktive_zuordnung | alias 'Partnersuche__un_suche'
an_id int -> Ext_Kontaktliste.Ansprechpartner(an_id) | alias 'Partnersuche__person_suche
Partnersuche__kontaktweg text nullable
erstellt "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)`
aktualisiert "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` {on_update: `datetime.datetime.now(datetime.UTC)`}
geloescht "timestamp with time zone" nullable | Soft-Delete Marker (partieller Unique-Index: WHERE geloescht IS NULL)
Grunderfassung_Personen
pers_id int pk {autoIncrement}
erstellt "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)`
aktualisiert "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` {on_update: `datetime.datetime.now(datetime.UTC)`}
geloescht "timestamp with time zone" nullable
Metadaten_erstellung "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` nullable
Metadaten_aktualisierung "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` nullable {on_update: `datetime.datetime.now(datetime.UTC)`}
Metadaten_wiedereintrittsdatum timestamp nullable
Metadaten_nutzer varchar(20) nullable
Grunderfassung_fallnummer text nullable
@@ -45,6 +45,7 @@ Grunderfassung_Personen
Projektrelevanz__relevanz text nullable
Projektrelevanz__foerderperiode text nullable
Schulbildung text nullable
Stammdaten__kanal_aufmerksamkeit text nullable
Stammdaten__PLZ text nullable
Stammdaten__anrede_anschrift text nullable
Stammdaten__anzahl_kinder__alter text nullable
@@ -71,19 +72,21 @@ Grunderfassung_Personen
WeitereInfos__WI_deutsch_sprache text nullable
WeitereInfos__WI_gueltigkeit_aufenthaltstitel text nullable
WeitereInfos__WI_meldung_institution text nullable
# Metadaten_erstellung "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` nullable
# Metadaten_aktualisierung "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` nullable {on_update: `datetime.datetime.now(datetime.UTC)`}
Zuordnung_Personen_Unternehmen | Eine Person kann zu mehreren Unternehmen gehören, ein Unternehmen kann mehrere Personen betreuen.
id int pk {autoIncrement}
un_id int -> Grunderfassung_Unternehmen(un_id)
pers_id int -> Grunderfassung_Personen(pers_id)
gueltig_ab "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)`
gueltig_bis "timestamp with time zone" nullable
# Zuordnung_Personen_Unternehmen | Eine Person kann zu mehreren Unternehmen gehören, ein Unternehmen kann mehrere Personen betreuen.
# id int pk {autoIncrement}
# un_id int -> Grunderfassung_Unternehmen(un_id)
# pers_id int -> Grunderfassung_Personen(pers_id)
# gueltig_ab "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)`
# gueltig_bis "timestamp with time zone" nullable
Beratung_Vorgang | Beratung kann an Unternehmen allein (pauschal), Person allein (pauschal) oder beide (individual) gebunden sein
vorgang_id int pk {autoIncrement}
un_id int nullable -> Grunderfassung_Unternehmen(un_id) | Pauschalberatungen sind sowohl für Unternehmen als auch Personen zulässig
un_id int nullable -> Unternehmen(un_id) | Pauschalberatungen sind sowohl für Unternehmen als auch Personen zulässig
pers_id int nullable -> Grunderfassung_Personen(pers_id) | sowohl Pauschal- als auch Individualberatung bei Personenbindung möglich
titel text
beratungs_typ berat_typ(pauschal, individual)
@@ -92,7 +95,6 @@ Beratung_Vorgang | Beratung kann an Unternehmen allein (pauschal), Person allein
geloescht "timestamp with time zone" nullable
# temporaer bool | Gibt an, ob der Vorgang losgelöst von der Grunderfassung gestartet wurde (Daten müssen nachträglich zugeordnet werden); theoretisch auch erfüllt, wenn `un_id` und `pers_id` beide NULL sind
Beratung_Einzelberatung | Änderungen müssen über einen DB-Trigger an Elterntabelle zurückgeschrieben werden
beratung_id int pk {autoIncrement}
vorgang_id int -> Beratung_Vorgang(vorgang_id) | Jede Beratung muss zu einem Beratungsvorgang gehören.
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
+256 -18
View File
@@ -8,20 +8,270 @@ import json
import re
from collections import defaultdict
from collections.abc import Sequence
from pprint import pprint
from pprint import pformat, pprint
from typing import Annotated, Any
import babel
import polars as pl
import sqlalchemy as sql
import sqlalchemy as sa
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator
from PySide6.QtCore import QDate, Qt
from wce_crm import constants, db
from wce_crm.backend import backend
from wce_crm.types import EntityIds, MainPageEntry, RecordingType
# %%
int("test")
# backend.page_consulting_linking_companies(None)
# backend.page_consulting_linking_persons(3)
ret = backend.main_page_get_entries()
pprint(ret)
# %%
with db.ENGINE.connect() as conn:
q_relevant_entries = (
sa.select(
db.t_grunderfassung_personen.c.pers_id,
db.t_grunderfassung_personen.c.aktualisiert,
db.t_zuordnung_personen_unternehmen.c.id.label("link_id"),
db.t_zuordnung_personen_unternehmen.c.un_id,
db.t_zuordnung_personen_unternehmen.c.an_id,
db.t_grunderfassung_personen.c.Stammdaten__name,
db.t_grunderfassung_personen.c.Stammdaten__vorname,
db.t_grunderfassung_personen.c.Stammdaten__ort,
)
.outerjoin(
db.t_zuordnung_personen_unternehmen,
db.t_grunderfassung_personen.c.pers_id
== db.t_zuordnung_personen_unternehmen.c.pers_id,
)
.order_by(db.t_grunderfassung_personen.c.Stammdaten__name)
).subquery("relevant_entries")
stmt = sa.select(
q_relevant_entries,
db.t_ext_crm_master_attach.c.ma_unternehmensname,
).outerjoin(
db.t_ext_crm_master_attach,
q_relevant_entries.c.un_id == db.t_ext_crm_master_attach.c.ma_id,
)
ret = conn.execute(stmt)
# results = list(ret.mappings())
# print(f"Results are:\n{pformat(results)}")
main_page_entries: list[MainPageEntry] = []
for entry in ret.mappings():
ids = EntityIds(
pers_id=entry["pers_id"],
link_id=entry["link_id"],
un_id=entry["un_id"],
an_id=entry["an_id"],
)
recording_type = RecordingType.WITHOUT_COMPANY
if ids.un_id or ids.link_id:
recording_type = RecordingType.WITH_COMPANY
first_name = entry["Stammdaten__vorname"]
last_name = entry["Stammdaten__name"]
location_city = entry["Stammdaten__ort"]
person_name = f"{first_name} {last_name}"
if location_city:
person_name += f" (Ort: {location_city})"
company_name = entry["ma_unternehmensname"]
display_name = person_name
if recording_type is RecordingType.WITH_COMPANY:
display_name = company_name
main_page_entries.append(
MainPageEntry(
ids=ids,
display_name=display_name,
dt_updated=entry["aktualisiert"],
recording_type=recording_type,
)
)
main_page_entries.sort(key=lambda x: x.dt_updated, reverse=True)
pprint(main_page_entries)
# %%
# %%
entry = results[0]
ids = EntityIds(
pers_id=entry["pers_id"],
link_id=entry["link_id"],
un_id=entry["un_id"],
an_id=entry["an_id"],
)
recording_type = RecordingType.WITHOUT_COMPANY
if ids.un_id or ids.link_id:
recording_type = RecordingType.WITH_COMPANY
first_name = entry["Stammdaten__vorname"]
last_name = entry["Stammdaten__name"]
location_city = entry["Stammdaten__ort"]
person_name = f"{first_name} {last_name}"
if location_city:
person_name += f" (Ort: {location_city})"
company_name = entry["ma_unternehmensname"]
display_name = person_name
if recording_type is RecordingType.WITH_COMPANY:
display_name = company_name
main_page_entry = MainPageEntry(
ids=ids,
display_name=display_name,
dt_updated=entry["aktualisiert"],
recording_type=recording_type,
)
# %%
pprint(main_page_entry)
# %%
# %%
@dc.dataclass(slots=True, kw_only=True)
class LinkingPerson:
pers_id: int
display_name: str
un_ids: set[int]
with db.ENGINE.connect() as conn:
# if un_id None: list all persons
# else: list just the persons associated with the company
stmt = (
sa.select(
db.t_grunderfassung_personen.c.pers_id,
db.t_zuordnung_personen_unternehmen.c.un_id,
db.t_grunderfassung_personen.c.Stammdaten__name,
db.t_grunderfassung_personen.c.Stammdaten__vorname,
db.t_grunderfassung_personen.c.Stammdaten__ort,
)
.outerjoin(
db.t_zuordnung_personen_unternehmen,
db.t_grunderfassung_personen.c.pers_id
== db.t_zuordnung_personen_unternehmen.c.pers_id,
)
.order_by(db.t_grunderfassung_personen.c.Stammdaten__name)
)
# if un_id is not None:
# person_select = person_select.select_from(
# db.t_grunderfassung_personen.join(
# db.t_zuordnung_personen_unternehmen,
# db.t_grunderfassung_personen.c.pers_id
# == db.t_zuordnung_personen_unternehmen.c.pers_id,
# )
# ).where(db.t_zuordnung_personen_unternehmen.c.un_id == un_id)
ret = conn.execute(stmt)
persons_map: dict[str, Any] = {}
for row in ret:
p_id = row.pers_id
if p_id not in persons_map:
first_name = row.Stammdaten__vorname
last_name = row.Stammdaten__name
location_city = row.Stammdaten__ort
person_name = f"{first_name} {last_name}"
if location_city:
person_name += f" (Ort: {location_city})"
persons_map[p_id] = {
"pers_id": p_id,
"display_name": person_name,
"un_ids": set(),
}
# Falls verknüpft, ID zur Menge hinzufügen
if row.un_id is not None:
persons_map[p_id]["un_ids"].add(row.un_id)
person_names = [LinkingPerson(**data) for data in persons_map.values()]
# person_names = ret.mappings().all()
# if not person_names:
# print(f"No persons found for 'un_id': {un_id}")
print(f"[Page Consulting -- Linking] Retrieved person names:\n{pformat(person_names)}")
# %%
dic = {
"t1": "test1",
"t2": "test2",
"t3": "test3",
"t4": "test4",
}
filter_keys = frozenset(("t3", "t2"))
dic_in, dic_out = backend.split_dict(dic, filter_keys)
print(f"{dic_in=}")
print(f"{dic_out=}")
# %%
dic_update = {
"t5": "new",
"t6": "new_2",
}
dic_updated = backend.dic_deep_update(dic, dic_update)
print(f"{dic=}")
print(f"{dic_update=}")
print(f"{dic_updated=}")
# %%
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()=}")
print(f"{raw_ent_ids.valid()=}")
# %%
raw_ent_ids.update(db_ent_ids)
print(f"{raw_ent_ids=}")
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(EntityIds())
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty()=}")
print(f"{raw_ent_ids.valid()=}")
# %%
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.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.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.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
@@ -29,22 +279,10 @@ crm_path = constants.Config.DB_PATH_CRM
engine = sql.create_engine(f"sqlite:///{str(db_path)}")
engine_crm = sql.create_engine(f"sqlite:///{str(crm_path)}")
# %%
stmt = sql.select(
db.grunderfassung.c.erfassung_id,
db.grunderfassung.c.Partnersuche__un_suche,
)
with engine.connect() as conn:
res = conn.execute(stmt)
for r in res:
print(r)
# %%
# %%
backend._main_page_get_company_list()
backend._main_page_get_company_list() # type: ignore
# %%
@@ -218,12 +456,12 @@ class FlatBaseModel(BaseModel):
return final_nested_data
def to_db(self) -> Dict[str, Any]:
def to_db(self) -> dict[str, Any]:
"""Ausgang für die DB: Flach, Listen sind JSON-Strings."""
nested = super().model_dump()
return self.__flatten_dict(nested, serialize_lists=True)
def to_gui(self) -> Dict[str, Any]:
def to_gui(self) -> dict[str, Any]:
"""Ausgang für die GUI: Flach, aber Listen bleiben Python-Listen."""
nested = super().model_dump()
return self.__flatten_dict(nested, serialize_lists=False)
src/wce_crm/backend/backend.py
+361 -422
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
@@ -13,9 +14,13 @@ 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,
FIELD_DB_SPLIT_METADATEN,
Beratungsgespraech_Einzelgespraech,
Beratungsgespraech_Vorgang,
InitRec,
Page_InitRec_Form_Data,
)
from wce_crm.logging import logger_back as logger
from wce_crm.types import (
@@ -29,11 +34,17 @@ from wce_crm.types import (
LinkingConsultationsEntryCompany,
LinkingConsultationsEntryPerson,
MainPageEntry,
RecordingType,
)
if TYPE_CHECKING:
from dopt_basics.datastructures import DualDict
from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId
K = TypeVar("K")
V = TypeVar("V")
def _transform_for_gui_output(
data: pl.DataFrame,
@@ -69,7 +80,8 @@ def initrec_comp_search_choices() -> tuple[tuple[str, int], ...]:
return tuple(zip(df["dedupl"], df["ma_id"]))
def initrec_comp_search_get_info(
# TODO change to direct SQL interaction
def external_company_get_info(
ma_id: ExtMaId,
) -> CompanyInfo:
logger.debug("[Call backend] comp_search_get_info")
@@ -122,139 +134,67 @@ def initrec_comp_contact_person_search_get_info(
# // internals
# TODO check removal
# def initrec_company_insert_initial_recording(
# data: dict[str, Any],
# ) -> RecId:
# logger.debug("[Call backend] insert_initial_recording")
# stmt = db.grunderfassung_unternehmen.insert()
# with db.ENGINE.begin() as conn:
# ret = conn.execute(stmt, data)
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)
# if ret.rowcount == 0:
# raise IOError("Entry was not inserted correctly")
for check_key in dic:
if check_key in split_keys:
keys_in[check_key] = keys_out.pop(check_key)
# prim_keys = ret.inserted_primary_key
# assert prim_keys
# return prim_keys[0]
return keys_in, keys_out
# def initrec_company_update_initial_recording(
# id_: RecId,
# data: dict[str, Any],
# ) -> None:
# logger.debug("[Call backend] update_initial_recording")
# stmt = db.grunderfassung_unternehmen.update().where(
# db.grunderfassung_unternehmen.c.un_id == id_
# )
# with db.ENGINE.begin() as conn:
# conn.execute(stmt, data)
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 initrec_company_to_db(
auto_form_data: InitRec,
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:
logger.debug("[AutoForm -- backend] Call database saving routine...")
dump_data = copy.deepcopy(auto_form_data.db_data)
dump_data["geloescht"] = auto_form_data.geloescht
with db.ENGINE.begin() as conn:
if auto_form_data.ids.un_id is None:
logger.debug("[AutoForm -- backend] Insert...")
# person
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
stmt = db.grunderfassung_unternehmen.insert().returning(
db.grunderfassung_unternehmen.c.un_id,
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
db.grunderfassung_unternehmen.c.geloescht,
)
ret = conn.execute(stmt, dump_data)
if data.ids.pers_id is None:
logger.debug("[InitRec -- backend] Insert...")
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0]
auto_form_data.ids.un_id = data_from_db["un_id"]
auto_form_data.geloescht = data_from_db["geloescht"]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
logger.debug("[AutoForm -- backend] Inserted InitRec Company successfully")
else:
logger.debug("[AutoForm -- backend] Update...")
stmt = (
db.grunderfassung_unternehmen.update()
.where(db.grunderfassung_unternehmen.c.un_id == auto_form_data.ids.un_id)
.returning(
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
db.grunderfassung_unternehmen.c.geloescht,
)
)
ret = conn.execute(stmt, dump_data)
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excatly one returned row"
data_from_db = from_db[0]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
logger.debug(
"[AutoForm -- backend] Updated InitRec Company with ID %d successfully",
auto_form_data.ids.un_id,
)
return auto_form_data
def initrec_company_from_db(
id_: RecId,
) -> InitRec:
logger.debug("[AutoForm -- backend] Call database loading routine...")
stmt = db.grunderfassung_unternehmen.select().where(
db.grunderfassung_unternehmen.c.un_id == id_
)
with db.ENGINE.connect() as conn:
ret = conn.execute(stmt)
from_db = ret.mappings().all()
assert len(from_db) == 1, "not excactly one company initial recording obtained"
data_from_db = dict(from_db[0])
geloescht = data_from_db["geloescht"]
del data_from_db["geloescht"]
return InitRec(
ids=EntityIds(un_id=id_),
# rec_id=id_, # TODO remove
geloescht=geloescht,
db_data=data_from_db,
)
def initrec_person_to_db(
auto_form_data: InitRec,
) -> InitRec:
logger.debug("[AutoForm -- backend] Call database saving routine...")
dump_data = copy.deepcopy(auto_form_data.db_data)
dump_data["geloescht"] = auto_form_data.geloescht
with db.ENGINE.begin() as conn:
if auto_form_data.ids.pers_id is None:
logger.debug("[AutoForm -- backend] Insert...")
stmt = sa.insert(db.grunderfassung_personen).returning(
db.grunderfassung_personen.c.pers_id,
db.grunderfassung_personen.c.Metadaten_aktualisierung,
db.grunderfassung_personen.c.geloescht,
stmt = sa.insert(db.t_grunderfassung_personen).returning(
db.t_grunderfassung_personen
)
stmt_compiled = str(stmt.compile(db.ENGINE))
logger.debug(
"[AutoForm -- backend] Data to insert:\nStatement: %s\n%s",
"[InitRec -- backend] Data to insert:\nStatement: %s\n%s",
stmt_compiled,
pformat(dump_data),
)
@@ -263,170 +203,189 @@ def initrec_person_to_db(
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0]
auto_form_data.ids.pers_id = data_from_db["pers_id"]
auto_form_data.geloescht = data_from_db["geloescht"]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
data_from_db = dict(from_db[0])
data_from_db = _rename_db_data(
data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
data.ids.pers_id = data_from_db["pers_id"]
logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully")
else:
logger.debug("[AutoForm -- backend] Update...")
stmt = (
db.grunderfassung_personen.update()
.where(db.grunderfassung_personen.c.pers_id == auto_form_data.ids.pers_id)
.returning(
db.grunderfassung_personen.c.Metadaten_aktualisierung,
db.grunderfassung_personen.c.geloescht,
)
db.t_grunderfassung_personen.update()
.where(db.t_grunderfassung_personen.c.pers_id == data.ids.pers_id)
.returning(db.t_grunderfassung_personen)
)
ret = conn.execute(stmt, dump_data)
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excatly one returned row"
data_from_db = from_db[0]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
logger.debug(
"[AutoForm -- backend] Updated InitRec Person with ID %d successfully",
auto_form_data.ids.pers_id,
data_from_db = dict(from_db[0])
data_from_db = _rename_db_data(
data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
return auto_form_data
# 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
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 --> linking table
dump_data = _rename_db_data(
data.form_data.company, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
)
logger.debug(
"[InitRec -- backend] Company - additional info: Dump Data = \n%s",
pformat(dump_data),
)
if "un_id" not in dump_data:
# only case when updated
dump_data["un_id"] = data.ids.un_id
dump_data["an_id"] = data.ids.an_id
stmt = sqlite_insert(db.t_zuordnung_personen_unternehmen).values(
**dump_data, pers_id=data.ids.pers_id
)
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__kontaktweg": stmt.excluded.Partnersuche__kontaktweg,
"aktualisiert": datetime.datetime.now(datetime.UTC),
"geloescht": data.geloescht,
},
).returning(db.t_zuordnung_personen_unternehmen.c.id)
ret = conn.execute(stmt)
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excatly one returned row"
data_from_db = dict(from_db[0])
data_from_db = _rename_db_data(
data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
)
data.ids.link_id = data_from_db["id"]
return data
def initrec_person_from_db(
id_: RecId,
def page_initrec_from_db(
ids: EntityIds,
) -> InitRec:
logger.debug("[AutoForm -- backend] Call database loading routine...")
stmt = db.grunderfassung_personen.select().where(
db.grunderfassung_personen.c.pers_id == id_
logger.debug("[InitRec -- backend] Call database loading routine...")
recording_type = RecordingType.WITHOUT_COMPANY
if ids.link_id is not None or ids.valid(EntityType.COMPANY):
recording_type = RecordingType.WITH_COMPANY
data_company_table: dict[str, Any] = {}
if recording_type is RecordingType.WITH_COMPANY:
logger.debug("[InitRec -- backend] With company. Link ID: %s", ids.link_id)
if ids.link_id:
stmt = sa.select(
db.t_zuordnung_personen_unternehmen.c.pers_id,
db.t_zuordnung_personen_unternehmen.c.un_id,
db.t_zuordnung_personen_unternehmen.c.an_id,
db.t_zuordnung_personen_unternehmen.c.Partnersuche__kontaktweg,
).where(
db.t_zuordnung_personen_unternehmen.c.id == ids.link_id,
)
else:
assert ids.un_id, "company ID not set"
assert ids.pers_id, "person ID not set"
stmt = sa.select(
db.t_zuordnung_personen_unternehmen.c.pers_id,
db.t_zuordnung_personen_unternehmen.c.un_id,
db.t_zuordnung_personen_unternehmen.c.an_id,
db.t_zuordnung_personen_unternehmen.c.Partnersuche__kontaktweg,
).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])
ids.pers_id = data_company_table.pop("pers_id")
ids.un_id = data_company_table["un_id"]
ids.an_id = data_company_table["an_id"]
data_company_table = _rename_db_data(
data_company_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN
)
# person
assert ids.pers_id, "person ID not set"
logger.debug("[InitRec -- backend] Person... Person ID: %s", ids.pers_id)
stmt = db.t_grunderfassung_personen.select().where(
db.t_grunderfassung_personen.c.pers_id == ids.pers_id
)
with db.ENGINE.connect() as conn:
ret = conn.execute(stmt)
from_db = ret.mappings().all()
assert len(from_db) == 1, "not excactly one company initial recording obtained"
data_from_db = dict(from_db[0])
data_person_table = dict(from_db[0])
geloescht = data_from_db["geloescht"]
del data_from_db["geloescht"]
data_person_table = _rename_db_data(
data_person_table, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
)
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=in_filter,
person=out_filter,
company=data_company_table,
)
if geloescht is not None:
logger.warning(
"[InitRec -- backend] Loaded entries which seem to be deleted - IDs: %s", ids
)
return InitRec(
ids=EntityIds(pers_id=id_),
# rec_id=id_, # TODO remove
ids=ids,
recording_type=recording_type,
geloescht=geloescht,
db_data=data_from_db,
form_data=form_data,
)
# TODO check removal
def initrec_company_get_initial_recording(
id_: RecId,
) -> dict[str, Any]:
logger.debug("[Call backend] get_initial_recording")
stmt = db.grunderfassung_unternehmen.select().where(
db.grunderfassung_unternehmen.c.un_id == id_
)
with db.ENGINE.connect() as conn:
ret = conn.execute(stmt)
results = ret.mappings().all()
if not results:
raise KeyError(f"Database ID {id_} not found")
assert len(results) == 1, "more than one company initial recording obtained"
row = results[0]
assert row, "row was not obtained"
return dict(row)
# def initrec_company_delete_initial_recording(
# id_: RecId,
# ) -> None:
# logger.debug("[Call backend] delete_initial_recording")
# 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],
# ) -> RecId:
# 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_: RecId,
# 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_: RecId,
# ) -> dict[str, Any]:
# logger.debug("[Call backend] get_initial_recording person")
# stmt = db.grunderfassung_personen.select().where(
# db.grunderfassung_personen.c.pers_id == id_
# )
# with db.ENGINE.connect() as conn:
# ret = conn.execute(stmt)
# results = ret.mappings().all()
# if not results:
# raise KeyError(f"Database ID {id_} not found")
# assert len(results) == 1, "more than one person initial recording obtained"
# row = results[0]
# assert row, "row was not obtained"
# return dict(row)
# def initrec_person_delete_initial_recording(
# id_: RecId,
# ) -> 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)
# if ret.rowcount == 0:
# raise KeyError(f"Database ID {id_} not found for deletion")
@wrap_result(10)
def page_consulting_to_db(
consultation_data: Beratungsgespraech_Vorgang,
@@ -444,7 +403,7 @@ def page_consulting_to_db(
pformat(insert_data_process),
)
stmt = db.beratung_vorgang.insert()
stmt = db.t_beratung_vorgang.insert()
ret = conn.execute(stmt, insert_data_process)
if ret.rowcount == 0:
raise IOError("Entry was not inserted correctly")
@@ -460,8 +419,8 @@ def page_consulting_to_db(
consultation_data.vorgang_id,
)
update_data_process = consultation_data.model_dump(exclude={"beratungen"})
stmt = db.beratung_vorgang.update().where(
db.beratung_vorgang.c.vorgang_id == consultation_data.vorgang_id
stmt = db.t_beratung_vorgang.update().where(
db.t_beratung_vorgang.c.vorgang_id == consultation_data.vorgang_id
)
conn.execute(stmt, update_data_process)
@@ -487,8 +446,8 @@ def page_consulting_to_db(
logger.debug(
"[Consulting Page] Call update for sessions:\n%s", pformat(rows_for_db_update)
)
stmt = db.beratung_einzelberatung.update().where(
db.beratung_einzelberatung.c.beratung_id == sa.bindparam("b_beratung_id")
stmt = db.t_beratung_einzelberatung.update().where(
db.t_beratung_einzelberatung.c.beratung_id == sa.bindparam("b_beratung_id")
)
conn.execute(stmt, rows_for_db_update)
@@ -497,8 +456,8 @@ def page_consulting_to_db(
logger.debug(
"[Consulting Page] Call insert for sessions:\n%s", pformat(rows_for_db_insert)
)
stmt = db.beratung_einzelberatung.insert().returning(
db.beratung_einzelberatung.c.beratung_id
stmt = db.t_beratung_einzelberatung.insert().returning(
db.t_beratung_einzelberatung.c.beratung_id
)
res = conn.execute(stmt, rows_for_db_insert)
new_cons_session_ids = cast(list[int], [row[0] for row in res.fetchall()])
@@ -518,12 +477,14 @@ def page_consulting_delete(
dump_data: dict[str, Any] = {"geloescht": timestamp_deleted}
with db.ENGINE.begin() as conn:
stmt = db.beratung_vorgang.update().where(db.beratung_vorgang.c.vorgang_id == cons_id)
stmt = db.t_beratung_vorgang.update().where(
db.t_beratung_vorgang.c.vorgang_id == cons_id
)
conn.execute(stmt, dump_data)
stmt = db.beratung_einzelberatung.update().where(
db.beratung_einzelberatung.c.vorgang_id == cons_id,
db.beratung_einzelberatung.c.geloescht.is_(None),
stmt = db.t_beratung_einzelberatung.update().where(
db.t_beratung_einzelberatung.c.vorgang_id == cons_id,
db.t_beratung_einzelberatung.c.geloescht.is_(None),
)
conn.execute(stmt, dump_data)
@@ -537,7 +498,9 @@ def page_consulting_from_db(
with db.ENGINE.connect() as conn:
# get the consultation process
stmt = db.beratung_vorgang.select().where(db.beratung_vorgang.c.vorgang_id == cons_id)
stmt = db.t_beratung_vorgang.select().where(
db.t_beratung_vorgang.c.vorgang_id == cons_id
)
ret = conn.execute(stmt)
results = ret.mappings().all()
@@ -570,9 +533,9 @@ def page_consulting_from_db(
# get all consultation sessions of this process, only the ones
# which are not marked as deleted
stmt = db.beratung_einzelberatung.select().where(
db.beratung_einzelberatung.c.vorgang_id == cons_id,
db.beratung_einzelberatung.c.geloescht.is_(None),
stmt = db.t_beratung_einzelberatung.select().where(
db.t_beratung_einzelberatung.c.vorgang_id == cons_id,
db.t_beratung_einzelberatung.c.geloescht.is_(None),
)
ret = conn.execute(stmt)
cons_sessions = ret.mappings().all()
@@ -609,27 +572,24 @@ def page_consulting_linking_companies(
) -> list[LinkingConsultationsEntryCompany]:
with db.ENGINE.connect() as conn:
q_relevant_companies = sa.select(
db.grunderfassung_unternehmen.c.un_id,
db.grunderfassung_unternehmen.c.Partnersuche__un_suche,
db.t_unternehmen.c.un_id,
).where(
db.grunderfassung_unternehmen.c.geloescht.is_(None),
db.grunderfassung_unternehmen.c.Partnersuche__un_suche.is_not(None),
db.t_unternehmen.c.geloescht.is_(None),
)
if un_id is not None:
q_relevant_companies = q_relevant_companies.where(
db.grunderfassung_unternehmen.c.un_id == un_id
db.t_unternehmen.c.un_id == un_id
)
q_relevant_companies = q_relevant_companies.subquery("relevant_companies")
stmt = sa.select(
q_relevant_companies.c.un_id,
db.ext_crm_master_attach.c.ma_unternehmensname,
db.t_ext_crm_master_attach.c.ma_unternehmensname,
).select_from(
q_relevant_companies.join(
db.ext_crm_master_attach,
q_relevant_companies.c.Partnersuche__un_suche
== db.ext_crm_master_attach.c.ma_id,
db.t_ext_crm_master_attach,
q_relevant_companies.c.un_id == db.t_ext_crm_master_attach.c.ma_id,
)
)
@@ -652,55 +612,49 @@ def page_consulting_linking_companies(
return linking_entries
def page_consulting_linking_persons(
un_id: RecId | None,
) -> list[LinkingConsultationsEntryPerson]:
def page_consulting_linking_persons() -> list[LinkingConsultationsEntryPerson]:
with db.ENGINE.connect() as conn:
# if un_id None: list all persons
# else: list just the persons associated with the company
person_select = sa.select(
db.grunderfassung_personen.c.pers_id,
db.grunderfassung_personen.c.Stammdaten__name,
db.grunderfassung_personen.c.Stammdaten__vorname,
db.grunderfassung_personen.c.Stammdaten__ort,
)
if un_id is not None:
person_select = person_select.select_from(
db.grunderfassung_personen.join(
db.zuordnung_personen_unternehmen,
db.grunderfassung_personen.c.pers_id
== db.zuordnung_personen_unternehmen.c.pers_id,
)
).where(db.zuordnung_personen_unternehmen.c.un_id == un_id)
ret = conn.execute(person_select)
person_names = ret.mappings().all()
if not person_names:
logger.warning(f"No persons found for 'un_id': {un_id}")
return []
logger.debug("[Page Consulting -- Linking] Retrieved person names:\n%s", person_names)
linking_entries: list[LinkingConsultationsEntryPerson] = []
for entry in person_names:
first_name = entry["Stammdaten__vorname"]
last_name = entry["Stammdaten__name"]
location_city = entry["Stammdaten__ort"]
display_name = f"{first_name} {last_name}"
if location_city:
display_name += f" (Ort: {location_city})"
linking_entries.append(
LinkingConsultationsEntryPerson(
pers_id=entry["pers_id"],
display_name=display_name,
stmt = (
sa.select(
db.t_grunderfassung_personen.c.pers_id,
db.t_zuordnung_personen_unternehmen.c.un_id,
db.t_grunderfassung_personen.c.Stammdaten__name,
db.t_grunderfassung_personen.c.Stammdaten__vorname,
db.t_grunderfassung_personen.c.Stammdaten__ort,
)
.outerjoin(
db.t_zuordnung_personen_unternehmen,
db.t_grunderfassung_personen.c.pers_id
== db.t_zuordnung_personen_unternehmen.c.pers_id,
)
.order_by(db.t_grunderfassung_personen.c.Stammdaten__name)
)
return linking_entries
ret = conn.execute(stmt)
persons_map: dict[str, LinkingConsultationsEntryPerson] = {}
for row in ret:
p_id = row.pers_id
if p_id not in persons_map:
first_name = row.Stammdaten__vorname
last_name = row.Stammdaten__name
location_city = row.Stammdaten__ort
display_name = f"{first_name} {last_name}"
if location_city:
display_name += f" (Ort: {location_city})"
persons_map[p_id] = LinkingConsultationsEntryPerson(
pers_id=p_id,
un_ids=set(),
display_name=display_name,
)
if row.un_id is not None:
persons_map[p_id].un_ids.add(row.un_id)
return list(persons_map.values())
# TODO wrap result
@@ -709,10 +663,9 @@ def page_consulting_eastablish_link(
un_id: RecId | None,
pers_id: RecId | None,
) -> None:
stmt = (
sa.update(db.beratung_vorgang)
.where(db.beratung_vorgang.c.vorgang_id == vorgang_id)
sa.update(db.t_beratung_vorgang)
.where(db.t_beratung_vorgang.c.vorgang_id == vorgang_id)
.values(un_id=un_id, pers_id=pers_id)
)
@@ -720,16 +673,19 @@ def page_consulting_eastablish_link(
conn.execute(stmt)
# // company profile interaction
def companyprofile_page_get_consultations(
un_id: RecId,
) -> CompanyProfileConsultations:
logger.debug("[Call backend] _companyprofile_page_get_consultations")
stmt = sa.select(
db.beratung_vorgang.c.vorgang_id,
db.beratung_vorgang.c.aktualisiert,
db.beratung_vorgang.c.titel,
db.beratung_vorgang.c.beratungs_typ,
).where(db.beratung_vorgang.c.un_id == un_id, db.beratung_vorgang.c.geloescht.is_(None))
db.t_beratung_vorgang.c.vorgang_id,
db.t_beratung_vorgang.c.aktualisiert,
db.t_beratung_vorgang.c.titel,
db.t_beratung_vorgang.c.beratungs_typ,
).where(
db.t_beratung_vorgang.c.un_id == un_id, db.t_beratung_vorgang.c.geloescht.is_(None)
)
with db.ENGINE.connect() as conn:
res = conn.execute(stmt)
@@ -743,11 +699,8 @@ def companyprofile_page_get_consultations(
datetime_updated = cast(datetime.datetime, entry["aktualisiert"])
datetime_updated = datetime_updated.astimezone(TIMEZONE_CEST)
base_title = entry["titel"]
_cons_type = entry["beratungs_typ"]
cons_type = ConsultingType(_cons_type)
cons_type = ConsultingType(entry["beratungs_typ"])
con_entry = CompanyProfileConsultationEntry(
cons_id=cons_id,
@@ -761,7 +714,7 @@ def companyprofile_page_get_consultations(
elif cons_type is ConsultingType.INDIVIDUAL:
cons_entries_individual.append(con_entry)
else:
raise TypeError(f"Unknown consulting type: {_cons_type}")
raise TypeError(f"Unknown consulting type: {cons_type}")
cons_entries_pauschal.sort(key=lambda x: x.date_updated, reverse=True)
cons_entries_individual.sort(key=lambda x: x.date_updated, reverse=True)
@@ -773,7 +726,7 @@ def companyprofile_page_get_consultations(
# // initial recording interaction
# TODO wrap result
# TODO wrap result, add to corresponding component
def initrec_link_person_company(
un_id: RecId,
pers_id: RecId,
@@ -782,17 +735,17 @@ def initrec_link_person_company(
if unlink:
deleted_ts = datetime.datetime.now(tz=datetime.UTC)
stmt = (
sa.update(db.zuordnung_personen_unternehmen)
sa.update(db.t_zuordnung_personen_unternehmen)
.where(
db.zuordnung_personen_unternehmen.c.un_id == un_id,
db.zuordnung_personen_unternehmen.c.pers_id == pers_id,
db.zuordnung_personen_unternehmen.c.gueltig_bis.is_(None),
db.t_zuordnung_personen_unternehmen.c.un_id == un_id,
db.t_zuordnung_personen_unternehmen.c.pers_id == pers_id,
db.t_zuordnung_personen_unternehmen.c.gueltig_bis.is_(None),
)
.values(gueltig_bis=deleted_ts)
)
else:
stmt = (
sqlite_insert(db.zuordnung_personen_unternehmen)
sqlite_insert(db.t_zuordnung_personen_unternehmen)
.values(un_id=un_id, pers_id=pers_id)
.on_conflict_do_nothing(
index_elements=["un_id", "pers_id"],
@@ -805,85 +758,71 @@ def initrec_link_person_company(
# // main page interaction
def _main_page_get_company_list() -> list[MainPageEntry]:
logger.debug("[Call backend] get_company_list")
stmt = sa.select(
db.grunderfassung_unternehmen.c.un_id,
db.grunderfassung_unternehmen.c.Partnersuche__un_suche,
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
).where(db.grunderfassung_unternehmen.c.geloescht.is_(None))
with db.ENGINE.connect() as conn:
res = conn.execute(stmt)
main_page_companies: list[MainPageEntry] = []
for entry in res.mappings():
rec_id = entry["un_id"]
assert rec_id, "no RecID defined"
ma_id_external = entry["Partnersuche__un_suche"]
assert ma_id_external is not None, "external MA ID is NULL"
datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"])
datetime_akt = datetime_akt.astimezone(TIMEZONE_CEST)
comp_info = initrec_comp_search_get_info(ma_id_external)
display_name = comp_info["ma_unternehmensname"]
main_page_companies.append(
MainPageEntry(
rec_id=rec_id,
display_name=display_name,
Metadaten_aktualisierung=datetime_akt,
type=EntityType.COMPANY,
)
)
return main_page_companies
def _main_page_get_person_list() -> list[MainPageEntry]:
logger.debug("[Call backend] get_company_list")
stmt = sa.select(
db.grunderfassung_personen.c.pers_id,
db.grunderfassung_personen.c.Stammdaten__vorname,
db.grunderfassung_personen.c.Stammdaten__name,
db.grunderfassung_personen.c.Metadaten_aktualisierung,
).where(db.grunderfassung_personen.c.geloescht.is_(None))
with db.ENGINE.connect() as conn:
res = conn.execute(stmt)
main_page_persons: list[MainPageEntry] = []
for entry in res.mappings():
rec_id = entry["pers_id"]
assert rec_id, "no RecID defined"
datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"])
datetime_akt = datetime_akt.astimezone()
surname: str = entry["Stammdaten__name"]
first_name: str = entry["Stammdaten__vorname"]
names_to_join = [n for n in (first_name, surname) if n]
display_name = " ".join(names_to_join)
main_page_persons.append(
MainPageEntry(
rec_id=rec_id,
display_name=display_name,
Metadaten_aktualisierung=datetime_akt,
type=EntityType.PERSON,
)
)
return main_page_persons
def main_page_get_entries() -> list[MainPageEntry]:
companies = _main_page_get_company_list()
persons = _main_page_get_person_list()
with db.ENGINE.connect() as conn:
q_relevant_entries = (
sa.select(
db.t_grunderfassung_personen.c.pers_id,
db.t_grunderfassung_personen.c.aktualisiert,
db.t_zuordnung_personen_unternehmen.c.id.label("link_id"),
db.t_zuordnung_personen_unternehmen.c.un_id,
db.t_zuordnung_personen_unternehmen.c.an_id,
db.t_grunderfassung_personen.c.Stammdaten__name,
db.t_grunderfassung_personen.c.Stammdaten__vorname,
db.t_grunderfassung_personen.c.Stammdaten__ort,
)
.outerjoin(
db.t_zuordnung_personen_unternehmen,
db.t_grunderfassung_personen.c.pers_id
== db.t_zuordnung_personen_unternehmen.c.pers_id,
)
.where(db.t_grunderfassung_personen.c.geloescht.is_(None))
.order_by(db.t_grunderfassung_personen.c.Stammdaten__name)
).subquery("relevant_entries")
all_entries = companies + persons
stmt = sa.select(
q_relevant_entries,
db.t_ext_crm_master_attach.c.ma_unternehmensname,
).outerjoin(
db.t_ext_crm_master_attach,
q_relevant_entries.c.un_id == db.t_ext_crm_master_attach.c.ma_id,
)
all_entries.sort(key=lambda x: x.Metadaten_aktualisierung, reverse=True)
ret = conn.execute(stmt)
return all_entries
main_page_entries: list[MainPageEntry] = []
for entry in ret.mappings():
ids = EntityIds(
pers_id=entry["pers_id"],
link_id=entry["link_id"],
un_id=entry["un_id"],
an_id=entry["an_id"],
)
recording_type = RecordingType.WITHOUT_COMPANY
if ids.un_id or ids.link_id:
recording_type = RecordingType.WITH_COMPANY
first_name = entry["Stammdaten__vorname"]
last_name = entry["Stammdaten__name"]
location_city = entry["Stammdaten__ort"]
person_name = f"{first_name} {last_name}"
if location_city:
person_name += f" (Ort: {location_city})"
company_name = entry["ma_unternehmensname"]
display_name = person_name
if recording_type is RecordingType.WITH_COMPANY:
display_name = company_name
main_page_entries.append(
MainPageEntry(
ids=ids,
display_name=display_name,
dt_updated=entry["aktualisiert"],
recording_type=recording_type,
)
)
main_page_entries.sort(key=lambda x: x.dt_updated, reverse=True)
return main_page_entries
src/wce_crm/data_models.py
+154 -80
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 EntityIds
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:
from wce_crm.gui import Page_Consulting_ConsultingSession # noqa: F401
from wce_crm.types import ConsId, ConsultingType, EntityType, RecId, UserId
from wce_crm.gui import InitRecForm, Page_Consulting_ConsultingSession # noqa: F401
from wce_crm.types import ConsId, EntityType, RecId, UserId
ValidAge = Annotated[int, Field(ge=0, le=99)]
COLUMN_SEP: Final[str] = "__"
@@ -123,36 +124,6 @@ class Module(WrapperModule, Protocol):
def save_data(self) -> None: ...
# TODO check removal
# class AutoFormInsert(Protocol):
# def __call__(
# self,
# data: dict[str, Any],
# ) -> RecId: ...
# class AutoFormUpdate(Protocol):
# def __call__(
# self,
# id_: int,
# data: dict[str, Any],
# ) -> None: ...
# class AutoFormGet(Protocol):
# def __call__(
# self,
# id_: int,
# ) -> dict[str, Any]: ...
# class AutoFormDelete(Protocol):
# def __call__(
# self,
# id_: int,
# ) -> None: ...
class InitRecFormToDb(Protocol):
def __call__(
self,
@@ -163,7 +134,7 @@ class InitRecFormToDb(Protocol):
class InitRecFormFromDb(Protocol):
def __call__(
self,
id_: int,
ids: EntityIds,
) -> InitRec: ...
@@ -171,19 +142,24 @@ class InitRecFormFromDb(Protocol):
def set_page_state(
target: S,
source: S,
ignore_attr: set[str] | None = None,
deep: bool = False,
) -> None:
"""
transfer all field values generically from 'source' to 'target'
"""
if not (dc.is_dataclass(target) and dc.is_dataclass(source)):
raise TypeError("Both source and atrget must be dataclasses")
raise TypeError("Both source and target must be dataclasses")
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)
@@ -212,50 +188,83 @@ class Page_MainPage_State(PageState[Module]):
@dc.dataclass(slots=True, kw_only=True)
class Page_NewInitRec_State(PageState[Module]):
class Page_NewEntry_State(PageState[Module]):
session: Session
@dc.dataclass(slots=True, kw_only=True)
class InitRecFormConfig:
model: type[FlatBaseModel]
to_db: InitRecFormToDb
from_db: InitRecFormFromDb
class Page_InitRec_Form_Data:
metadata: dict[str, Any]
person: dict[str, Any]
company: dict[str, Any]
def _create_empty_initrec_form_data() -> Page_InitRec_Form_Data:
return Page_InitRec_Form_Data(
metadata={},
person={},
company={},
)
# !! 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: EntityIds = dc.field(default_factory=EntityIds)
recording_type: RecordingType
# un_id: RecId | None = None
locked: bool
geloescht: bool = False
form_data: Page_InitRec_Form_Data = dc.field(
default_factory=_create_empty_initrec_form_data
)
initrec_form_states: dict[EntityType, Page_InitRec_Form_State] = dc.field(
init=False, default_factory=dict
)
def __post_init__(self) -> None:
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,
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]
form_fields: Sequence[FormField]
ignored_keys: Iterable[str] = tuple()
add_buttons: bool = True
add_buttons: bool = False
id_mapping: dict[str, str]
@dc.dataclass(slots=True, kw_only=True)
class InitRecForm_State(PageState[Module]):
class Page_InitRec_Form_State(PageState[Module]):
session: Session
cfg: InitRecFormConfig
cfg: Page_InitRec_Form_Config
ids: EntityIds
ent_type: EntityType
# rec_id: RecId | None = None
form_data: dict[str, Any] | None = None
form_data: dict[str, Any] = dc.field(default_factory=dict)
locked: bool
geloescht: bool = False
@dc.dataclass(slots=True, kw_only=True)
class Page_InitRecCompany_State(PageState[Module]):
session: Session
ids: EntityIds
# un_id: RecId | None = None
locked: bool
@dc.dataclass(slots=True, kw_only=True)
class Page_InitRecPerson_State(PageState[Module]):
session: Session
ids: EntityIds
# pers_id: RecId | None = None
# un_id: RecId | None = None
locked: bool
# TODO change to entity IDs
@dc.dataclass(slots=True, kw_only=True)
class Page_Consulting_State(PageState[Module]):
session: Session
@@ -271,13 +280,13 @@ 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
un_id: RecId | None = None
pers_id: RecId | None = None
beratungs_typ: ConsultingType
filter_un_ids: set[RecId] | None = None
@dc.dataclass(slots=True, kw_only=True)
@@ -308,8 +317,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: EntityIds
# rec_id: RecId | None = None
un_id: RecId | None = None
# // Pydantic models
@@ -428,24 +436,36 @@ class FlatBaseModel(BaseModel):
# ** InitRec
class InitRec(BaseModel):
ids: EntityIds
# rec_id: RecId | None
recording_type: RecordingType
geloescht: AwareDatetime | None = None
db_data: dict[str, Any]
form_data: Page_InitRec_Form_Data
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
# 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",
)
Partnersuche: Grunderfassung_PartnerSuche
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_Personen(FlatBaseModel):
# default in SQLAlchemy with lambda and timezone-aware datetime
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
@@ -453,6 +473,24 @@ class Grunderfassung_Personen(FlatBaseModel):
Grunderfassung_fallnummer: str
Grunderfassung_notiz: str | None
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
# Metadaten_nutzer: str | None
Projektrelevanz: Grunderfassung_Projektrelevanz
Kontaktperson: Grunderfassung_Kontaktperson
Stammdaten: Grunderfassung_Stammdaten
@@ -468,7 +506,7 @@ class Grunderfassung_PartnerSuche(BaseModel):
un_suche: int | None
person_suche: int | None
kanal_aufmerksamkeit: str | None
kontaktweg: str | None
class Grunderfassung_Projektrelevanz(BaseModel):
@@ -496,6 +534,7 @@ class Grunderfassung_Kontaktperson(BaseModel):
class Grunderfassung_Stammdaten(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True)
kanal_aufmerksamkeit: str | None
titel: str | None
anrede_anschrift: str
name: str
@@ -635,3 +674,38 @@ 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.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={
"Partnersuche__un_suche": "un_id",
"Partnersuche__person_suche": "an_id",
},
),
EntityType.PERSON: Page_InitRec_Form_Config(
pydantic_model=Grunderfassung_Personen,
ignored_keys=tuple(),
# ignored_keys=(
# "Metadaten_erstellung",
# "Metadaten_aktualisierung",
# ),
form_fields=INITREC_PERSON,
add_buttons=False,
id_mapping={},
),
}
src/wce_crm/db.py
+111 -54
View File
@@ -255,7 +255,7 @@ 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)
ext_crm_master_attach: Table = Table(
t_ext_crm_master_attach: Table = Table(
"Master",
MD_CRM,
Column("ma_id", sa.Integer, nullable=False, unique=True),
@@ -285,9 +285,71 @@ ext_crm_master_attach: Table = Table(
schema="kontaktliste",
)
t_ext_crm_contact_person_attach: Table = Table(
"Ansprechpartner",
MD_CRM,
Column("an_id", sa.Integer, nullable=False, unique=True),
Column("ma_id", sa.ForeignKey("Master.ma_id")),
Column("wce_id", sa.ForeignKey("Nutzer.wce_id")),
Column("st_id", sa.Integer, nullable=False),
Column("an_sachgebiet", sa.Text, nullable=True),
Column("an_anrede", sa.Text, nullable=True),
Column("an_titel", sa.Text, nullable=True),
Column("an_nachname", sa.Text, nullable=True),
Column("an_vorname", sa.Text, nullable=True),
Column("an_position", sa.Text, nullable=True),
Column("an_mail", sa.Text, nullable=True),
Column("an_festnetz", sa.Text, nullable=True),
Column("an_mobil", sa.Text, nullable=True),
Column("an_faxnummer", sa.Text, nullable=True),
Column("an_hauptansprechpartner", sa.Text, nullable=True),
Column("an_anrede_anschrift", sa.Text, nullable=True),
Column("an_bemerkung", sa.Text, nullable=True),
Column("an_aktualisierung_datum", SafeDateTime, nullable=True),
Column("an_aktualisierung_nutzer", sa.Text, nullable=True),
Column("an_letztes_kontaktdatum", SafeDateTime, nullable=True),
Column("an_ersteintrag_datum", SafeDateTime, nullable=True),
Column("an_archiviert", sa.Boolean, nullable=True, default=0),
schema="kontaktliste",
)
grunderfassung_unternehmen: Table = Table(
"grunderfassung_unternehmen",
# grunderfassung_unternehmen: Table = Table(
# "grunderfassung_unternehmen",
# MD_MAIN,
# Column(
# "un_id",
# sa.Integer,
# primary_key=True,
# autoincrement=True,
# ),
# Column(
# "geloescht",
# UTCDateTime,
# nullable=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", sa.String(20), nullable=True),
# Column("Grunderfassung_fallnummer", sa.Text, nullable=True),
# Column("Grunderfassung_notiz", sa.Text, nullable=True),
# Column("Partnersuche__kanal_aufmerksamkeit", sa.Text, nullable=True, default=None),
# Column("Partnersuche__person_suche", sa.Integer, nullable=True, default=None),
# Column("Partnersuche__un_suche", sa.Integer, nullable=True, default=None),
# )
t_unternehmen: Table = Table(
"Unternehmen",
MD_MAIN,
Column(
"un_id",
@@ -295,34 +357,21 @@ grunderfassung_unternehmen: Table = Table(
primary_key=True,
autoincrement=True,
),
Column(
"erstellt",
UTCDateTime,
nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC),
),
Column(
"geloescht",
UTCDateTime,
nullable=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", sa.String(20), nullable=True),
Column("Grunderfassung_fallnummer", sa.Text, nullable=True),
Column("Grunderfassung_notiz", sa.Text, nullable=True),
Column("Partnersuche__kanal_aufmerksamkeit", sa.Text, nullable=True, default=None),
Column("Partnersuche__person_suche", sa.Integer, nullable=True, default=None),
Column("Partnersuche__un_suche", sa.Integer, nullable=True, default=None),
)
grunderfassung_personen: Table = Table(
"grunderfassung_personen",
t_grunderfassung_personen: Table = Table(
"Grunderfassung_Personen",
MD_MAIN,
Column(
"pers_id",
@@ -330,24 +379,24 @@ grunderfassung_personen: Table = Table(
primary_key=True,
autoincrement=True,
),
Column(
"erstellt",
UTCDateTime,
nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC),
),
Column(
"aktualisiert",
UTCDateTime,
nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC),
onupdate=lambda: datetime.datetime.now(datetime.UTC),
),
Column(
"geloescht",
UTCDateTime,
nullable=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", sa.String(20), nullable=True),
Column("Metadaten_wiedereintrittsdatum", sa.Date, nullable=True, default=None),
Column("Arbeitserfahrung", sa.Text, nullable=True),
@@ -361,15 +410,14 @@ grunderfassung_personen: Table = Table(
Column("Kontaktperson__KP_funktion_beziehung", sa.Text, nullable=True),
Column("Kontaktperson__KP_mobilfunknummer", sa.Text, nullable=True),
Column("Kontaktperson__KP_name", sa.Text, nullable=True),
Column(
"Kontaktperson__KP_name_partner", sa.Text, nullable=True
), # TODO: check if needed when set by trigger
Column("Kontaktperson__KP_name_partner", sa.Text, nullable=True),
Column("Kontaktperson__KP_titel", sa.Text, nullable=True),
Column("Kontaktperson__KP_vorname", sa.Text, nullable=True),
Column("Projektrelevanz__relevanz", sa.Text, nullable=True),
Column("Projektrelevanz__foerderperiode", sa.Text, nullable=True),
Column("Schulbildung", sa.Text, nullable=True),
Column("Sprachkenntnisse", sa.Text, nullable=True),
Column("Stammdaten__kanal_aufmerksamkeit", sa.Text, nullable=True),
Column("Stammdaten__PLZ", sa.Text, nullable=True),
Column("Stammdaten__anrede_anschrift", sa.Text, nullable=True),
Column("Stammdaten__anzahl_kinder__alter", sa.Text, nullable=True),
@@ -398,8 +446,8 @@ grunderfassung_personen: Table = Table(
Column("WeitereInfos__WI_meldung_institution", sa.Text, nullable=True),
)
zuordnung_personen_unternehmen: Table = Table(
"zuordnung_personen_unternehmen",
t_zuordnung_personen_unternehmen: Table = Table(
"Personen_Unternehmen_Zuordnung",
MD_MAIN,
Column(
"id",
@@ -407,31 +455,40 @@ zuordnung_personen_unternehmen: Table = Table(
primary_key=True,
autoincrement=True,
),
Column("un_id", sa.ForeignKey("grunderfassung_unternehmen.un_id"), nullable=False),
Column("pers_id", sa.ForeignKey("grunderfassung_personen.pers_id"), nullable=False),
Column("un_id", sa.ForeignKey("Unternehmen.un_id"), nullable=False),
Column("an_id", sa.Integer, nullable=False),
Column("pers_id", sa.ForeignKey("Grunderfassung_Personen.pers_id"), nullable=False),
Column(
"gueltig_ab",
"erstellt",
UTCDateTime,
nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC),
),
Column(
"gueltig_bis",
"aktualisiert",
UTCDateTime,
nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC),
onupdate=lambda: datetime.datetime.now(datetime.UTC),
),
Column(
"geloescht",
UTCDateTime,
nullable=True,
),
Column("Partnersuche__kontaktweg", sa.Text, nullable=True),
Index(
"uq_aktive_zuordnung",
"un_id",
"pers_id",
unique=True,
sqlite_where=sql.text("gueltig_bis IS NULL"),
sqlite_where=sql.text("geloescht IS NULL"),
),
)
beratung_vorgang: Table = Table(
"beratung_vorgang",
t_beratung_vorgang: Table = Table(
"Beratung_Vorgang",
MD_MAIN,
Column(
"vorgang_id",
@@ -439,8 +496,8 @@ 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(
@@ -464,8 +521,8 @@ beratung_vorgang: Table = Table(
)
beratung_einzelberatung: Table = Table(
"beratung_einzelberatung",
t_beratung_einzelberatung: Table = Table(
"Beratung_Einzelberatung",
MD_MAIN,
Column(
"beratung_id",
@@ -473,7 +530,7 @@ 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
+59 -111
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,
@@ -307,53 +307,38 @@ INITREC_COMP_SEARCH_HEAD = [
info="an_position",
),
FormField(
"Wie sind Sie auf uns aufmerksam geworden?",
"Kontaktweg",
FormFieldType.DROPDOWN,
required=False,
key="kanal_aufmerksamkeit",
key="kontaktweg",
options=[
("Agentur für Arbeit", None),
("Ausländerbehörde", None),
("Jobcenter", None),
("Freunde/Familie", None),
("Anerkennungsstelle", None),
("Beratungsstelle", None),
("Internet", None),
("Arbeitgeber", None),
("Bildungsdienstleister", None),
("Welcome-Mappe", None),
("Newsletter WFE", None),
("Newsletter RM", None),
("Sonstiges", None),
("Telefon", None),
("Mail", None),
("Persönliches Gespräch", None),
],
),
]
INITREC_CONTACT_PERSON = [
# TODO check removal
_INITREC_CONTACT_PERSON = [
FormField(
"Name Unternehmen/Netzwerkpartner (vorausgefüllt von Suche)",
"Name Unternehmen/Netzwerkpartner",
FormFieldType.TEXT,
key="KP_name_partner",
required=False,
placeholder="[TODO: Logik in Verbindung mit Unternehmen unklar]",
readonly=True,
readonly=False,
),
FormField(
"Titel",
FormFieldType.TEXT,
key="KP_titel",
required=False,
tooltip=(
"* nur wenn anrufende Person oder kontaktaufnehmende Person "
"nicht die zu beratende Person ist"
),
),
FormField(
"Anrede_Anschrift",
FormFieldType.TEXT,
"Anrede",
FormFieldType.DROPDOWN,
key="KP_anrede_anschrift",
required=True,
options=[("Sehr geehrte Frau", None), ("Sehr geehrter Herr", None)],
),
FormField(
"Name",
@@ -400,22 +385,40 @@ INITREC_CONTACT_PERSON = [
]
INITREC_MASTER_DATA = [
_INITREC_MASTER_DATA = [
FormField(
"Wie sind Sie auf uns aufmerksam geworden?",
FormFieldType.DROPDOWN,
required=False,
key="kanal_aufmerksamkeit",
options=[
("Agentur für Arbeit", None),
("Ausländerbehörde", None),
("Jobcenter", None),
("Freunde/Familie", None),
("Anerkennungsstelle", None),
("Beratungsstelle", None),
("Internet", None),
("Arbeitgeber", None),
("Bildungsdienstleister", None),
("Welcome-Mappe", None),
("Newsletter WFE", None),
("Newsletter RM", None),
("Sonstiges", None),
],
),
FormField(
"Titel",
FormFieldType.TEXT,
key="titel",
required=False,
tooltip=(
"* nur wenn anrufende Person oder kontaktaufnehmende Person "
"nicht die zu beratende Person ist"
),
),
FormField(
"Anrede",
FormFieldType.TEXT,
FormFieldType.DROPDOWN,
key="anrede_anschrift",
required=True,
options=[("Sehr geehrte Frau", None), ("Sehr geehrter Herr", None)],
),
FormField(
"Name",
@@ -427,17 +430,13 @@ INITREC_MASTER_DATA = [
"Vorname",
FormFieldType.TEXT,
key="vorname",
required=False,
required=True,
),
FormField(
"Geburtsdatum",
FormFieldType.DATE,
FormFieldType.DATE_FLEXIBLE,
key="geburtsdatum",
required=False,
tooltip=(
"* Wichtig zu erfragen, da u.a. Mindestgehaltsschwelle davon abhängt "
"(Regelung bei Ü45 Jahre)"
),
),
FormField(
"Herkunftsland",
@@ -446,7 +445,6 @@ INITREC_MASTER_DATA = [
required=True,
placeholder="Suche...",
options=COUNTRY_LIST.for_dropdown,
tooltip=("* Wichtig zu erfragen aufgrund eventueller EU-Freizügigkeitsregelung"),
),
FormField(
"Staatsangehörigkeit",
@@ -455,7 +453,6 @@ INITREC_MASTER_DATA = [
required=False,
placeholder="Suche...",
options=COUNTRY_LIST.for_dropdown,
tooltip=("* Wichtig zu erfragen aufgrund eventueller EU-Freizügigkeitsregelung"),
),
FormField(
"Rückkehrer",
@@ -463,7 +460,6 @@ INITREC_MASTER_DATA = [
key="rueckkehrer",
required=False,
options=[("ja", None), ("nein", None)],
tooltip=("* Wichtig zu erfragen aufgrund eventueller EU-Freizügigkeitsregelung"),
),
FormField(
"Wo befindet sich die Person?",
@@ -502,7 +498,6 @@ INITREC_MASTER_DATA = [
key="bundesland",
required=False,
options=GERMAN_STATE_LIST.for_dropdown,
tooltip=("nur wenn Inland angegeben"),
),
FormField(
"Land",
@@ -535,13 +530,11 @@ INITREC_MASTER_DATA = [
FormFieldType.TEXT,
key="familienstand",
required=False,
tooltip="* Wichtig zu erfragen aufgrund Lebensunterhaltssicherung",
),
FormField(
"Anzahl Kinder",
FormFieldType.DYNAMIC_DROPDOWN_NUMERIC,
required=False,
tooltip="* Wichtig zu erfragen aufgrund Lebensunterhaltssicherung",
key="anzahl_kinder",
children=[
FormField(
@@ -549,7 +542,6 @@ INITREC_MASTER_DATA = [
FormFieldType.DROPDOWN,
required=False,
options=[(str(x), None) for x in range(11)],
tooltip="* Wichtig zu erfragen aufgrund Lebensunterhaltssicherung",
key="anzahl",
children=[
FormField("Alter Kind", FormFieldType.TEXT, key="alter"),
@@ -560,7 +552,7 @@ INITREC_MASTER_DATA = [
]
INITREC_ADDITIONAL_DATA = [
_INITREC_ADDITIONAL_DATA = [
FormField(
"Deutsch als Kommunikationssprache",
FormFieldType.DROPDOWN,
@@ -593,11 +585,10 @@ INITREC_ADDITIONAL_DATA = [
("Chancenaufenthaltsrecht §104c AufenthG", None),
("Sonstiges", None),
],
tooltip="sofern nicht bekannt, unbedingt einfordern",
),
FormField(
"Gültigkeit Aufenthaltsstatus",
FormFieldType.DATE,
FormFieldType.DATE_FLEXIBLE,
required=False,
key="WI_gueltigkeit_aufenthaltstitel",
),
@@ -635,7 +626,7 @@ INITREC_ADDITIONAL_DATA = [
]
INITREC_SCHOOL = [
_INITREC_SCHOOL = [
FormField("Abschluss", FormFieldType.TEXT, required=False, key="SB_abschluss"),
FormField(
"Abschlussgrad laut Dokument",
@@ -673,7 +664,7 @@ INITREC_SCHOOL = [
]
INITREC_HIGHER_EDUCATION = [
_INITREC_HIGHER_EDUCATION = [
FormField(
"Anerkennung",
FormFieldType.TEXT,
@@ -685,10 +676,6 @@ INITREC_HIGHER_EDUCATION = [
FormFieldType.TEXT,
required=False,
key="HB_abschlussgrad",
tooltip=(
"bitte den Titel eingeben z.B. Doktor, Diplom oder "
"Betriebswirt (Fachschulabschluss)"
),
),
FormField(
"Abschlussgrad laut Dokument",
@@ -707,10 +694,6 @@ INITREC_HIGHER_EDUCATION = [
FormFieldType.TEXT,
required=False,
key="HB_beruf",
tooltip=(
"bitte spezifizieren z.B. Allgemeinmedizin, Ingenieur Maschinenbau, "
"technischer Betriebswirt Datenverarbeitung"
),
),
FormField(
"Land",
@@ -737,12 +720,11 @@ INITREC_HIGHER_EDUCATION = [
FormFieldType.TEXT,
required=False,
key="HB_bemerkungsfeld",
tooltip="z.B. Promotionen oder den Studiengang angeben",
),
]
INITREC_WORK_EXPERIENCE = [
_INITREC_WORK_EXPERIENCE = [
FormField(
"Branche",
FormFieldType.DROPDOWN,
@@ -824,7 +806,6 @@ INITREC_WORK_EXPERIENCE = [
("Teilzeit", None),
("Sonstiges", None),
],
tooltip="Minijob, Praktikum, Wehrdienst, soziale Dienste",
),
FormField(
"Bemerkungsfeld",
@@ -835,7 +816,7 @@ INITREC_WORK_EXPERIENCE = [
]
INITREC_LANGUAGES = [
_INITREC_LANGUAGES = [
FormField(
"Sprache",
FormFieldType.TEXT,
@@ -877,7 +858,7 @@ INITREC_LANGUAGES = [
),
FormField(
"Datum des Nachweises",
FormFieldType.DATE,
FormFieldType.DATE_FLEXIBLE,
required=False,
key="SP_datum_nachweis",
),
@@ -889,52 +870,16 @@ 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(
"Suche",
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,
@@ -957,11 +902,11 @@ INITREC_PERSON = [
required=False,
key="Metadaten_nutzer",
readonly=True,
ignore_get_data=True,
ignore_get_data=False,
),
FormField(
"Wiedereintrittsdatum",
FormFieldType.DATE,
FormFieldType.DATE_FLEXIBLE,
required=False,
key="Metadaten_wiedereintrittsdatum",
readonly=False,
@@ -978,6 +923,9 @@ INITREC_PERSON = [
required=False,
key="Grunderfassung_notiz",
),
]
INITREC_PERSON = [
FormField(
"Status && Projektrelevanz",
FormFieldType.GROUP,
@@ -1009,42 +957,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
+1148 -774
View File
File diff suppressed because it is too large Load Diff
src/wce_crm/logging.py
+14 -5
View File
@@ -30,19 +30,28 @@ logger_base.setLevel(logging.DEBUG)
# ** GUI
logger_gui = logger_base.getChild("gui")
logger_gui.setLevel(logging.DEBUG)
logger_search_widget = logger_gui.getChild("search_widget")
logger_search_widget.setLevel(logging.DEBUG)
logger_search_widget.setLevel(logging.WARNING)
logger_get_data = logger_gui.getChild("get_data")
logger_get_data.setLevel(logging.DEBUG)
logger_auto_form = logger_gui.getChild("get_data_auto_form")
logger_auto_form.setLevel(logging.DEBUG)
logger_get_data.setLevel(logging.WARNING)
logger_custom_fields = logger_gui.getChild("custom_fields")
logger_custom_fields.setLevel(logging.DEBUG)
logger_custom_fields.setLevel(logging.WARNING)
logger_page_consulting = logger_gui.getChild("page_consulting")
logger_page_consulting.setLevel(logging.DEBUG)
logger_page_initrec = logger_gui.getChild("page_initrec")
logger_page_initrec.setLevel(logging.DEBUG)
logger_initrecform = logger_gui.getChild("initrecform")
logger_initrecform.setLevel(logging.DEBUG)
logger_prototyping = logger_gui.getChild("prototyping")
logger_prototyping.setLevel(logging.DEBUG)
# ** Backend
logger_back = logger_base.getChild("backend")
logger_back.setLevel(logging.DEBUG)
src/wce_crm/types.py
+152 -44
View File
@@ -1,9 +1,11 @@
from __future__ import annotations
import copy
import dataclasses as dc
import datetime
import enum
from typing import TYPE_CHECKING, TypeAlias, TypedDict
from collections.abc import Generator
from typing import TYPE_CHECKING, Final, TypeAlias, TypedDict
if TYPE_CHECKING:
import polars as pl
@@ -12,13 +14,22 @@ if TYPE_CHECKING:
UserId: TypeAlias = int
RecId: TypeAlias = int
ConsId: TypeAlias = int
IntLinkId: TypeAlias = int
ExtMaId: TypeAlias = int
ExtAnId: TypeAlias = int
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()
@@ -30,54 +41,150 @@ class ConsultingType(enum.StrEnum):
@dc.dataclass(slots=True, kw_only=True)
class EntityIds:
un_id: RecId | None = None
un_id: ExtMaId | None = None
an_id: ExtAnId | None = None
pers_id: RecId | None = None
type: dc.InitVar[EntityType | None] = None
_ent_type: EntityType | None = dc.field(init=False)
link_id: IntLinkId | None = None
_field_names: frozenset[str] = dc.field(init=False, repr=False, compare=False)
def __post_init__(self, type: EntityType | None) -> None:
self._ent_type = type
def __post_init__(self) -> None:
field_names = set(f.name for f in dc.fields(EntityIds))
field_names.remove("_field_names")
self._field_names = frozenset(field_names)
@property
def ent_type(self) -> EntityType | None:
return self._ent_type
def any(self) -> bool:
return self.un_id is not None or self.pers_id is not None
def all(self) -> bool:
return self.un_id is not None and self.pers_id is not None
def valid(self) -> bool:
if self.ent_type is None:
raise TypeError("Initrec type not defined. Can not determine valid state.")
elif self.ent_type is EntityType.COMPANY:
return self.all()
elif self.ent_type is EntityType.PERSON:
return self.any()
else:
raise RuntimeError(
f"Unknown initrec type {self.ent_type}. This is not supported."
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}"
)
def set_id(
self,
id_: RecId | None,
ent_type: EntityType,
) -> None:
if ent_type is EntityType.COMPANY:
self.un_id = id_
elif ent_type is EntityType.PERSON:
self.pers_id = id_
setattr(self, attr_name, value)
def get_id(
def update(
self,
ent_type: EntityType,
) -> RecId | None:
if ent_type is EntityType.COMPANY:
return self.un_id
ids: EntityIds,
ignore_none: bool = False,
deep: bool = False,
) -> None:
if not isinstance(ids, EntityIds):
raise TypeError("New IDs must be have the same type as this object")
for field in dc.fields(ids):
new_value = getattr(ids, field.name)
if ignore_none and new_value is None:
continue
if deep and new_value is not None:
new_value = copy.deepcopy(new_value)
setattr(self, field.name, new_value)
def _get_relevant_fields(
self,
ent_type: EntityType | None = None,
) -> Generator[int | None, None, None]:
if ent_type is None:
relevant_fields = (getattr(self, field_name) for field_name in self._field_names)
elif ent_type is EntityType.COMPANY:
# RELEVANT_ATTRIBUTES = {"un_id", "an_id", "link_id"}
RELEVANT_ATTRIBUTES = {"un_id", "an_id"}
relevant_fields = (
getattr(self, field_name)
for field_name in self._field_names
if field_name in RELEVANT_ATTRIBUTES
)
elif ent_type is EntityType.PERSON:
return self.pers_id
RELEVANT_ATTRIBUTES = {"pers_id"}
relevant_fields = (
getattr(self, field_name)
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}")
return relevant_fields
def empty(
self,
ent_type: EntityType | None = None,
) -> bool:
relevant_fields = self._get_relevant_fields(ent_type=ent_type)
return not any(relevant_fields)
def valid(
self,
ent_type: EntityType | None = None,
) -> bool:
relevant_fields = self._get_relevant_fields(ent_type=ent_type)
return all(relevant_fields)
# @dc.dataclass(slots=True, kw_only=True)
# class EntityIds:
# un_id: RecId | None = None
# pers_id: RecId | None = None
# type: dc.InitVar[RecordingType | None] = None
# _rec_type: RecordingType | None = dc.field(init=False)
# def __post_init__(self, type: RecordingType | None) -> None:
# self._rec_type = type
# @property
# def rec_type(self) -> RecordingType | None:
# return self._rec_type
# def any(self) -> bool:
# return self.un_id is not None or self.pers_id is not None
# def all(self) -> bool:
# return self.un_id is not None and self.pers_id is not None
# def valid(self) -> bool:
# if self.rec_type is None:
# raise TypeError("Initrec type not defined. Can not determine valid state.")
# elif self.rec_type is RecordingType.COMPANY:
# return self.all()
# elif self.rec_type is RecordingType.PERSON:
# return self.any()
# else:
# raise RuntimeError(
# f"Unknown initrec type {self.rec_type}. This is not supported."
# )
# def set_id(
# self,
# id_: RecId | None,
# ent_type: RecordingType,
# ) -> None:
# if ent_type is RecordingType.COMPANY:
# self.un_id = id_
# elif ent_type is RecordingType.PERSON:
# self.pers_id = id_
# def get_id(
# self,
# ent_type: RecordingType,
# ) -> RecId | None:
# if ent_type is RecordingType.COMPANY:
# return self.un_id
# elif ent_type is RecordingType.PERSON:
# return self.pers_id
# // database interaction data structures
@@ -137,10 +244,10 @@ class ContactPersonInfo(TypedDict):
# ** internals
@dc.dataclass(slots=True)
class MainPageEntry:
rec_id: RecId
ids: EntityIds
display_name: str
Metadaten_aktualisierung: datetime.datetime
type: EntityType
dt_updated: datetime.datetime
recording_type: RecordingType
@dc.dataclass(slots=True)
@@ -166,4 +273,5 @@ class LinkingConsultationsEntryCompany:
@dc.dataclass(slots=True)
class LinkingConsultationsEntryPerson:
pers_id: RecId
un_ids: set[RecId]
display_name: str