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 ma_id int pk
Ext_Kontaktliste.Ansprechpartner
Ext_CRM.Ansprechpartner
an_id int pk an_id int pk
ma_id int -> Ext_CRM.Master(ma_id) ma_id int -> Ext_Kontaktliste.Master(ma_id)
Unternehmen | Es wird nur die ID für die Kontaktliste referenziert. Alle relevanten Infos werden live aus der Kontaktliste geladen via 'ATTACH'.
Grunderfassung_Unternehmen un_id int pk -> Ext_Kontaktliste.Master(ma_id)
un_id int pk {autoIncrement} erstellt "timestamp with time zone"
geloescht "timestamp with time zone" nullable 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 Grunderfassung_Personen
pers_id int pk {autoIncrement} 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 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_wiedereintrittsdatum timestamp nullable
Metadaten_nutzer varchar(20) nullable Metadaten_nutzer varchar(20) nullable
Grunderfassung_fallnummer text nullable Grunderfassung_fallnummer text nullable
@@ -45,6 +45,7 @@ Grunderfassung_Personen
Projektrelevanz__relevanz text nullable Projektrelevanz__relevanz text nullable
Projektrelevanz__foerderperiode text nullable Projektrelevanz__foerderperiode text nullable
Schulbildung text nullable Schulbildung text nullable
Stammdaten__kanal_aufmerksamkeit text nullable
Stammdaten__PLZ text nullable Stammdaten__PLZ text nullable
Stammdaten__anrede_anschrift text nullable Stammdaten__anrede_anschrift text nullable
Stammdaten__anzahl_kinder__alter text nullable Stammdaten__anzahl_kinder__alter text nullable
@@ -71,19 +72,21 @@ Grunderfassung_Personen
WeitereInfos__WI_deutsch_sprache text nullable WeitereInfos__WI_deutsch_sprache text nullable
WeitereInfos__WI_gueltigkeit_aufenthaltstitel text nullable WeitereInfos__WI_gueltigkeit_aufenthaltstitel text nullable
WeitereInfos__WI_meldung_institution 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. # Zuordnung_Personen_Unternehmen | Eine Person kann zu mehreren Unternehmen gehören, ein Unternehmen kann mehrere Personen betreuen.
id int pk {autoIncrement} # id int pk {autoIncrement}
un_id int -> Grunderfassung_Unternehmen(un_id) # un_id int -> Grunderfassung_Unternehmen(un_id)
pers_id int -> Grunderfassung_Personen(pers_id) # pers_id int -> Grunderfassung_Personen(pers_id)
gueltig_ab "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` # gueltig_ab "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)`
gueltig_bis "timestamp with time zone" nullable # gueltig_bis "timestamp with time zone" nullable
Beratung_Vorgang | Beratung kann an Unternehmen allein (pauschal), Person allein (pauschal) oder beide (individual) gebunden sein Beratung_Vorgang | Beratung kann an Unternehmen allein (pauschal), Person allein (pauschal) oder beide (individual) gebunden sein
vorgang_id int pk {autoIncrement} 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 pers_id int nullable -> Grunderfassung_Personen(pers_id) | sowohl Pauschal- als auch Individualberatung bei Personenbindung möglich
titel text titel text
beratungs_typ berat_typ(pauschal, individual) 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 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 # 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_Einzelberatung | Änderungen müssen über einen DB-Trigger an Elterntabelle zurückgeschrieben werden
beratung_id int pk {autoIncrement} beratung_id int pk {autoIncrement}
vorgang_id int -> Beratung_Vorgang(vorgang_id) | Jede Beratung muss zu einem Beratungsvorgang gehören. 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 import re
from collections import defaultdict from collections import defaultdict
from collections.abc import Sequence from collections.abc import Sequence
from pprint import pprint from pprint import pformat, pprint
from typing import Annotated, Any from typing import Annotated, Any
import babel import babel
import polars as pl import polars as pl
import sqlalchemy as sql import sqlalchemy as sa
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator, model_validator
from PySide6.QtCore import QDate, Qt from PySide6.QtCore import QDate, Qt
from wce_crm import constants, db from wce_crm import constants, db
from wce_crm.backend import backend 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 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 = sql.create_engine(f"sqlite:///{str(db_path)}")
engine_crm = sql.create_engine(f"sqlite:///{str(crm_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 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.""" """Ausgang für die DB: Flach, Listen sind JSON-Strings."""
nested = super().model_dump() nested = super().model_dump()
return self.__flatten_dict(nested, serialize_lists=True) 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.""" """Ausgang für die GUI: Flach, aber Listen bleiben Python-Listen."""
nested = super().model_dump() nested = super().model_dump()
return self.__flatten_dict(nested, serialize_lists=False) return self.__flatten_dict(nested, serialize_lists=False)
src/wce_crm/backend/backend.py
+352 -413
View File
@@ -2,8 +2,9 @@ from __future__ import annotations
import copy import copy
import datetime import datetime
from collections.abc import Container, Mapping
from pprint import pformat from pprint import pformat
from typing import TYPE_CHECKING, Any, cast from typing import TYPE_CHECKING, Any, TypeVar, cast
import polars as pl import polars as pl
import sqlalchemy as sa 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 import db
from wce_crm.constants import TIMEZONE_CEST from wce_crm.constants import TIMEZONE_CEST
from wce_crm.data_models import ( from wce_crm.data_models import (
FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN,
FIELD_DB_MAPPING_GRUNDERERFASSUNG_UNTERNEHMEN,
FIELD_DB_SPLIT_METADATEN,
Beratungsgespraech_Einzelgespraech, Beratungsgespraech_Einzelgespraech,
Beratungsgespraech_Vorgang, Beratungsgespraech_Vorgang,
InitRec, InitRec,
Page_InitRec_Form_Data,
) )
from wce_crm.logging import logger_back as logger from wce_crm.logging import logger_back as logger
from wce_crm.types import ( from wce_crm.types import (
@@ -29,11 +34,17 @@ from wce_crm.types import (
LinkingConsultationsEntryCompany, LinkingConsultationsEntryCompany,
LinkingConsultationsEntryPerson, LinkingConsultationsEntryPerson,
MainPageEntry, MainPageEntry,
RecordingType,
) )
if TYPE_CHECKING: if TYPE_CHECKING:
from dopt_basics.datastructures import DualDict
from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId
K = TypeVar("K")
V = TypeVar("V")
def _transform_for_gui_output( def _transform_for_gui_output(
data: pl.DataFrame, data: pl.DataFrame,
@@ -69,7 +80,8 @@ def initrec_comp_search_choices() -> tuple[tuple[str, int], ...]:
return tuple(zip(df["dedupl"], df["ma_id"])) 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, ma_id: ExtMaId,
) -> CompanyInfo: ) -> CompanyInfo:
logger.debug("[Call backend] comp_search_get_info") logger.debug("[Call backend] comp_search_get_info")
@@ -122,139 +134,67 @@ def initrec_comp_contact_person_search_get_info(
# // internals # // internals
# TODO check removal def split_dict(
# def initrec_company_insert_initial_recording( dic: dict[K, V],
# data: dict[str, Any], split_keys: Container[K],
# ) -> RecId: ) -> tuple[dict[K, V], dict[K, V]]:
# logger.debug("[Call backend] insert_initial_recording") keys_in: dict[K, V] = {}
# stmt = db.grunderfassung_unternehmen.insert() keys_out = copy.deepcopy(dic)
# with db.ENGINE.begin() as conn:
# ret = conn.execute(stmt, data)
# if ret.rowcount == 0: for check_key in dic:
# raise IOError("Entry was not inserted correctly") if check_key in split_keys:
keys_in[check_key] = keys_out.pop(check_key)
# prim_keys = ret.inserted_primary_key return keys_in, keys_out
# assert prim_keys
# return prim_keys[0]
# def initrec_company_update_initial_recording( def dic_deep_update(
# id_: RecId, dic: dict[K, V],
# data: dict[str, Any], data: Mapping[K, V],
# ) -> None: ) -> dict[K, V]:
# logger.debug("[Call backend] update_initial_recording") dic_copy = copy.deepcopy(dic)
# stmt = db.grunderfassung_unternehmen.update().where( dic_copy.update(data)
# db.grunderfassung_unternehmen.c.un_id == id_
# ) return dic_copy
# with db.ENGINE.begin() as conn:
# conn.execute(stmt, data)
def initrec_company_to_db( def _rename_db_data(
auto_form_data: InitRec, db_data: dict[str, Any],
mapping: DualDict,
) -> dict[str, Any]:
new_db_data = copy.deepcopy(db_data)
for k, v in db_data.items():
if k in mapping:
new_db_data[mapping[k]] = v
del new_db_data[k]
if k in mapping.inverted:
new_db_data[mapping.inverted[k]] = v
del new_db_data[k]
return new_db_data
def page_initrec_to_db(
data: InitRec,
) -> InitRec: ) -> InitRec:
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: with db.ENGINE.begin() as conn:
if auto_form_data.ids.un_id is None: # person
logger.debug("[AutoForm -- backend] Insert...") 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( if data.ids.pers_id is None:
db.grunderfassung_unternehmen.c.un_id, logger.debug("[InitRec -- backend] Insert...")
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
db.grunderfassung_unternehmen.c.geloescht,
)
ret = conn.execute(stmt, dump_data)
from_db = ret.mappings().fetchall() stmt = sa.insert(db.t_grunderfassung_personen).returning(
assert len(from_db) == 1, "expected excactly one returned row" db.t_grunderfassung_personen
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_compiled = str(stmt.compile(db.ENGINE)) stmt_compiled = str(stmt.compile(db.ENGINE))
logger.debug( logger.debug(
"[AutoForm -- backend] Data to insert:\nStatement: %s\n%s", "[InitRec -- backend] Data to insert:\nStatement: %s\n%s",
stmt_compiled, stmt_compiled,
pformat(dump_data), pformat(dump_data),
) )
@@ -263,170 +203,189 @@ def initrec_person_to_db(
from_db = ret.mappings().fetchall() from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excactly one returned row" assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0] data_from_db = dict(from_db[0])
auto_form_data.ids.pers_id = data_from_db["pers_id"] data_from_db = _rename_db_data(
auto_form_data.geloescht = data_from_db["geloescht"] data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ )
"Metadaten_aktualisierung"
] data.ids.pers_id = data_from_db["pers_id"]
logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully") logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully")
else: else:
logger.debug("[AutoForm -- backend] Update...") logger.debug("[AutoForm -- backend] Update...")
stmt = ( stmt = (
db.grunderfassung_personen.update() db.t_grunderfassung_personen.update()
.where(db.grunderfassung_personen.c.pers_id == auto_form_data.ids.pers_id) .where(db.t_grunderfassung_personen.c.pers_id == data.ids.pers_id)
.returning( .returning(db.t_grunderfassung_personen)
db.grunderfassung_personen.c.Metadaten_aktualisierung,
db.grunderfassung_personen.c.geloescht,
)
) )
ret = conn.execute(stmt, dump_data) ret = conn.execute(stmt, dump_data)
from_db = ret.mappings().fetchall() from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excatly one returned row" assert len(from_db) == 1, "expected excatly one returned row"
data_from_db = from_db[0] data_from_db = dict(from_db[0])
data_from_db = _rename_db_data(
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ data_from_db, FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN
"Metadaten_aktualisierung"
]
logger.debug(
"[AutoForm -- backend] Updated InitRec Person with ID %d successfully",
auto_form_data.ids.pers_id,
) )
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( def page_initrec_from_db(
id_: RecId, ids: EntityIds,
) -> InitRec: ) -> InitRec:
logger.debug("[AutoForm -- backend] Call database loading routine...") logger.debug("[InitRec -- backend] Call database loading routine...")
stmt = db.grunderfassung_personen.select().where(
db.grunderfassung_personen.c.pers_id == id_ 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: with db.ENGINE.connect() as conn:
ret = conn.execute(stmt) ret = conn.execute(stmt)
from_db = ret.mappings().all() from_db = ret.mappings().all()
assert len(from_db) == 1, "not excactly one company initial recording obtained" 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"] data_person_table = _rename_db_data(
del data_from_db["geloescht"] 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( return InitRec(
ids=EntityIds(pers_id=id_), ids=ids,
# rec_id=id_, # TODO remove recording_type=recording_type,
geloescht=geloescht, 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) @wrap_result(10)
def page_consulting_to_db( def page_consulting_to_db(
consultation_data: Beratungsgespraech_Vorgang, consultation_data: Beratungsgespraech_Vorgang,
@@ -444,7 +403,7 @@ def page_consulting_to_db(
pformat(insert_data_process), pformat(insert_data_process),
) )
stmt = db.beratung_vorgang.insert() stmt = db.t_beratung_vorgang.insert()
ret = conn.execute(stmt, insert_data_process) ret = conn.execute(stmt, insert_data_process)
if ret.rowcount == 0: if ret.rowcount == 0:
raise IOError("Entry was not inserted correctly") raise IOError("Entry was not inserted correctly")
@@ -460,8 +419,8 @@ def page_consulting_to_db(
consultation_data.vorgang_id, consultation_data.vorgang_id,
) )
update_data_process = consultation_data.model_dump(exclude={"beratungen"}) update_data_process = consultation_data.model_dump(exclude={"beratungen"})
stmt = db.beratung_vorgang.update().where( stmt = db.t_beratung_vorgang.update().where(
db.beratung_vorgang.c.vorgang_id == consultation_data.vorgang_id db.t_beratung_vorgang.c.vorgang_id == consultation_data.vorgang_id
) )
conn.execute(stmt, update_data_process) conn.execute(stmt, update_data_process)
@@ -487,8 +446,8 @@ def page_consulting_to_db(
logger.debug( logger.debug(
"[Consulting Page] Call update for sessions:\n%s", pformat(rows_for_db_update) "[Consulting Page] Call update for sessions:\n%s", pformat(rows_for_db_update)
) )
stmt = db.beratung_einzelberatung.update().where( stmt = db.t_beratung_einzelberatung.update().where(
db.beratung_einzelberatung.c.beratung_id == sa.bindparam("b_beratung_id") db.t_beratung_einzelberatung.c.beratung_id == sa.bindparam("b_beratung_id")
) )
conn.execute(stmt, rows_for_db_update) conn.execute(stmt, rows_for_db_update)
@@ -497,8 +456,8 @@ def page_consulting_to_db(
logger.debug( logger.debug(
"[Consulting Page] Call insert for sessions:\n%s", pformat(rows_for_db_insert) "[Consulting Page] Call insert for sessions:\n%s", pformat(rows_for_db_insert)
) )
stmt = db.beratung_einzelberatung.insert().returning( stmt = db.t_beratung_einzelberatung.insert().returning(
db.beratung_einzelberatung.c.beratung_id db.t_beratung_einzelberatung.c.beratung_id
) )
res = conn.execute(stmt, rows_for_db_insert) res = conn.execute(stmt, rows_for_db_insert)
new_cons_session_ids = cast(list[int], [row[0] for row in res.fetchall()]) 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} dump_data: dict[str, Any] = {"geloescht": timestamp_deleted}
with db.ENGINE.begin() as conn: 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) conn.execute(stmt, dump_data)
stmt = db.beratung_einzelberatung.update().where( stmt = db.t_beratung_einzelberatung.update().where(
db.beratung_einzelberatung.c.vorgang_id == cons_id, db.t_beratung_einzelberatung.c.vorgang_id == cons_id,
db.beratung_einzelberatung.c.geloescht.is_(None), db.t_beratung_einzelberatung.c.geloescht.is_(None),
) )
conn.execute(stmt, dump_data) conn.execute(stmt, dump_data)
@@ -537,7 +498,9 @@ def page_consulting_from_db(
with db.ENGINE.connect() as conn: with db.ENGINE.connect() as conn:
# get the consultation process # 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) ret = conn.execute(stmt)
results = ret.mappings().all() results = ret.mappings().all()
@@ -570,9 +533,9 @@ def page_consulting_from_db(
# get all consultation sessions of this process, only the ones # get all consultation sessions of this process, only the ones
# which are not marked as deleted # which are not marked as deleted
stmt = db.beratung_einzelberatung.select().where( stmt = db.t_beratung_einzelberatung.select().where(
db.beratung_einzelberatung.c.vorgang_id == cons_id, db.t_beratung_einzelberatung.c.vorgang_id == cons_id,
db.beratung_einzelberatung.c.geloescht.is_(None), db.t_beratung_einzelberatung.c.geloescht.is_(None),
) )
ret = conn.execute(stmt) ret = conn.execute(stmt)
cons_sessions = ret.mappings().all() cons_sessions = ret.mappings().all()
@@ -609,27 +572,24 @@ def page_consulting_linking_companies(
) -> list[LinkingConsultationsEntryCompany]: ) -> list[LinkingConsultationsEntryCompany]:
with db.ENGINE.connect() as conn: with db.ENGINE.connect() as conn:
q_relevant_companies = sa.select( q_relevant_companies = sa.select(
db.grunderfassung_unternehmen.c.un_id, db.t_unternehmen.c.un_id,
db.grunderfassung_unternehmen.c.Partnersuche__un_suche,
).where( ).where(
db.grunderfassung_unternehmen.c.geloescht.is_(None), db.t_unternehmen.c.geloescht.is_(None),
db.grunderfassung_unternehmen.c.Partnersuche__un_suche.is_not(None),
) )
if un_id is not None: if un_id is not None:
q_relevant_companies = q_relevant_companies.where( 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") q_relevant_companies = q_relevant_companies.subquery("relevant_companies")
stmt = sa.select( stmt = sa.select(
q_relevant_companies.c.un_id, 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( ).select_from(
q_relevant_companies.join( q_relevant_companies.join(
db.ext_crm_master_attach, db.t_ext_crm_master_attach,
q_relevant_companies.c.Partnersuche__un_suche q_relevant_companies.c.un_id == db.t_ext_crm_master_attach.c.ma_id,
== db.ext_crm_master_attach.c.ma_id,
) )
) )
@@ -652,55 +612,49 @@ def page_consulting_linking_companies(
return linking_entries return linking_entries
def page_consulting_linking_persons( def page_consulting_linking_persons() -> list[LinkingConsultationsEntryPerson]:
un_id: RecId | None,
) -> list[LinkingConsultationsEntryPerson]:
with db.ENGINE.connect() as conn: with db.ENGINE.connect() as conn:
# if un_id None: list all persons stmt = (
# else: list just the persons associated with the company sa.select(
db.t_grunderfassung_personen.c.pers_id,
person_select = sa.select( db.t_zuordnung_personen_unternehmen.c.un_id,
db.grunderfassung_personen.c.pers_id, db.t_grunderfassung_personen.c.Stammdaten__name,
db.grunderfassung_personen.c.Stammdaten__name, db.t_grunderfassung_personen.c.Stammdaten__vorname,
db.grunderfassung_personen.c.Stammdaten__vorname, db.t_grunderfassung_personen.c.Stammdaten__ort,
db.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: ret = conn.execute(stmt)
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) persons_map: dict[str, LinkingConsultationsEntryPerson] = {}
person_names = ret.mappings().all() for row in ret:
if not person_names: p_id = row.pers_id
logger.warning(f"No persons found for 'un_id': {un_id}")
return []
logger.debug("[Page Consulting -- Linking] Retrieved person names:\n%s", person_names) if p_id not in persons_map:
first_name = row.Stammdaten__vorname
linking_entries: list[LinkingConsultationsEntryPerson] = [] last_name = row.Stammdaten__name
for entry in person_names: location_city = row.Stammdaten__ort
first_name = entry["Stammdaten__vorname"]
last_name = entry["Stammdaten__name"]
location_city = entry["Stammdaten__ort"]
display_name = f"{first_name} {last_name}" display_name = f"{first_name} {last_name}"
if location_city: if location_city:
display_name += f" (Ort: {location_city})" display_name += f" (Ort: {location_city})"
linking_entries.append( persons_map[p_id] = LinkingConsultationsEntryPerson(
LinkingConsultationsEntryPerson( pers_id=p_id,
pers_id=entry["pers_id"], un_ids=set(),
display_name=display_name, display_name=display_name,
) )
)
return linking_entries 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 # TODO wrap result
@@ -709,10 +663,9 @@ def page_consulting_eastablish_link(
un_id: RecId | None, un_id: RecId | None,
pers_id: RecId | None, pers_id: RecId | None,
) -> None: ) -> None:
stmt = ( stmt = (
sa.update(db.beratung_vorgang) sa.update(db.t_beratung_vorgang)
.where(db.beratung_vorgang.c.vorgang_id == vorgang_id) .where(db.t_beratung_vorgang.c.vorgang_id == vorgang_id)
.values(un_id=un_id, pers_id=pers_id) .values(un_id=un_id, pers_id=pers_id)
) )
@@ -720,16 +673,19 @@ def page_consulting_eastablish_link(
conn.execute(stmt) conn.execute(stmt)
# // company profile interaction
def companyprofile_page_get_consultations( def companyprofile_page_get_consultations(
un_id: RecId, un_id: RecId,
) -> CompanyProfileConsultations: ) -> CompanyProfileConsultations:
logger.debug("[Call backend] _companyprofile_page_get_consultations") logger.debug("[Call backend] _companyprofile_page_get_consultations")
stmt = sa.select( stmt = sa.select(
db.beratung_vorgang.c.vorgang_id, db.t_beratung_vorgang.c.vorgang_id,
db.beratung_vorgang.c.aktualisiert, db.t_beratung_vorgang.c.aktualisiert,
db.beratung_vorgang.c.titel, db.t_beratung_vorgang.c.titel,
db.beratung_vorgang.c.beratungs_typ, db.t_beratung_vorgang.c.beratungs_typ,
).where(db.beratung_vorgang.c.un_id == un_id, db.beratung_vorgang.c.geloescht.is_(None)) ).where(
db.t_beratung_vorgang.c.un_id == un_id, db.t_beratung_vorgang.c.geloescht.is_(None)
)
with db.ENGINE.connect() as conn: with db.ENGINE.connect() as conn:
res = conn.execute(stmt) res = conn.execute(stmt)
@@ -743,11 +699,8 @@ def companyprofile_page_get_consultations(
datetime_updated = cast(datetime.datetime, entry["aktualisiert"]) datetime_updated = cast(datetime.datetime, entry["aktualisiert"])
datetime_updated = datetime_updated.astimezone(TIMEZONE_CEST) datetime_updated = datetime_updated.astimezone(TIMEZONE_CEST)
base_title = entry["titel"] base_title = entry["titel"]
cons_type = ConsultingType(entry["beratungs_typ"])
_cons_type = entry["beratungs_typ"]
cons_type = ConsultingType(_cons_type)
con_entry = CompanyProfileConsultationEntry( con_entry = CompanyProfileConsultationEntry(
cons_id=cons_id, cons_id=cons_id,
@@ -761,7 +714,7 @@ def companyprofile_page_get_consultations(
elif cons_type is ConsultingType.INDIVIDUAL: elif cons_type is ConsultingType.INDIVIDUAL:
cons_entries_individual.append(con_entry) cons_entries_individual.append(con_entry)
else: 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_pauschal.sort(key=lambda x: x.date_updated, reverse=True)
cons_entries_individual.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 # // initial recording interaction
# TODO wrap result # TODO wrap result, add to corresponding component
def initrec_link_person_company( def initrec_link_person_company(
un_id: RecId, un_id: RecId,
pers_id: RecId, pers_id: RecId,
@@ -782,17 +735,17 @@ def initrec_link_person_company(
if unlink: if unlink:
deleted_ts = datetime.datetime.now(tz=datetime.UTC) deleted_ts = datetime.datetime.now(tz=datetime.UTC)
stmt = ( stmt = (
sa.update(db.zuordnung_personen_unternehmen) sa.update(db.t_zuordnung_personen_unternehmen)
.where( .where(
db.zuordnung_personen_unternehmen.c.un_id == un_id, db.t_zuordnung_personen_unternehmen.c.un_id == un_id,
db.zuordnung_personen_unternehmen.c.pers_id == pers_id, db.t_zuordnung_personen_unternehmen.c.pers_id == pers_id,
db.zuordnung_personen_unternehmen.c.gueltig_bis.is_(None), db.t_zuordnung_personen_unternehmen.c.gueltig_bis.is_(None),
) )
.values(gueltig_bis=deleted_ts) .values(gueltig_bis=deleted_ts)
) )
else: else:
stmt = ( stmt = (
sqlite_insert(db.zuordnung_personen_unternehmen) sqlite_insert(db.t_zuordnung_personen_unternehmen)
.values(un_id=un_id, pers_id=pers_id) .values(un_id=un_id, pers_id=pers_id)
.on_conflict_do_nothing( .on_conflict_do_nothing(
index_elements=["un_id", "pers_id"], index_elements=["un_id", "pers_id"],
@@ -805,85 +758,71 @@ def initrec_link_person_company(
# // main page interaction # // 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]: def main_page_get_entries() -> list[MainPageEntry]:
companies = _main_page_get_company_list() with db.ENGINE.connect() as conn:
persons = _main_page_get_person_list() 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 collections.abc import Iterable, Sequence
from typing import TYPE_CHECKING, Annotated, Any, Final, Generic, Protocol, TypeVar from typing import TYPE_CHECKING, Annotated, Any, Final, Generic, Protocol, TypeVar
from dopt_basics.datastructures import DualDict
from pydantic import ( from pydantic import (
AwareDatetime, AwareDatetime,
BaseModel, BaseModel,
@@ -18,12 +19,12 @@ from pydantic import (
) )
from pydantic_core import ErrorDetails from pydantic_core import ErrorDetails
from wce_crm.form_defs import FormField from wce_crm.form_defs import INITREC_COMP, INITREC_METADATEN, INITREC_PERSON, FormField
from wce_crm.types import EntityIds from wce_crm.types import ConsultingType, EntityIds, EntityType, RecordingType
if TYPE_CHECKING: if TYPE_CHECKING:
from wce_crm.gui import Page_Consulting_ConsultingSession # noqa: F401 from wce_crm.gui import InitRecForm, Page_Consulting_ConsultingSession # noqa: F401
from wce_crm.types import ConsId, ConsultingType, EntityType, RecId, UserId from wce_crm.types import ConsId, EntityType, RecId, UserId
ValidAge = Annotated[int, Field(ge=0, le=99)] ValidAge = Annotated[int, Field(ge=0, le=99)]
COLUMN_SEP: Final[str] = "__" COLUMN_SEP: Final[str] = "__"
@@ -123,36 +124,6 @@ class Module(WrapperModule, Protocol):
def save_data(self) -> None: ... 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): class InitRecFormToDb(Protocol):
def __call__( def __call__(
self, self,
@@ -163,7 +134,7 @@ class InitRecFormToDb(Protocol):
class InitRecFormFromDb(Protocol): class InitRecFormFromDb(Protocol):
def __call__( def __call__(
self, self,
id_: int, ids: EntityIds,
) -> InitRec: ... ) -> InitRec: ...
@@ -171,19 +142,24 @@ class InitRecFormFromDb(Protocol):
def set_page_state( def set_page_state(
target: S, target: S,
source: S, source: S,
ignore_attr: set[str] | None = None,
deep: bool = False, deep: bool = False,
) -> None: ) -> None:
""" """
transfer all field values generically from 'source' to 'target' transfer all field values generically from 'source' to 'target'
""" """
if not (dc.is_dataclass(target) and dc.is_dataclass(source)): 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): if type(target) is not type(source):
raise TypeError("Both source and target must have the same type") 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): for field in dc.fields(source):
if field.name == "child_modules": if field.name in ignore_attr:
continue continue
new_value = getattr(source, field.name) new_value = getattr(source, field.name)
@@ -212,50 +188,83 @@ class Page_MainPage_State(PageState[Module]):
@dc.dataclass(slots=True, kw_only=True) @dc.dataclass(slots=True, kw_only=True)
class Page_NewInitRec_State(PageState[Module]): class Page_NewEntry_State(PageState[Module]):
session: Session session: Session
@dc.dataclass(slots=True, kw_only=True) @dc.dataclass(slots=True, kw_only=True)
class InitRecFormConfig: class Page_InitRec_Form_Data:
model: type[FlatBaseModel] metadata: dict[str, Any]
to_db: InitRecFormToDb person: dict[str, Any]
from_db: InitRecFormFromDb 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] form_fields: Sequence[FormField]
ignored_keys: Iterable[str] = tuple() 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) @dc.dataclass(slots=True, kw_only=True)
class InitRecForm_State(PageState[Module]): class Page_InitRec_Form_State(PageState[Module]):
session: Session session: Session
cfg: InitRecFormConfig cfg: Page_InitRec_Form_Config
ids: EntityIds ids: EntityIds
ent_type: EntityType ent_type: EntityType
# rec_id: RecId | None = None # rec_id: RecId | None = None
form_data: dict[str, Any] | None = None form_data: dict[str, Any] = dc.field(default_factory=dict)
locked: bool locked: bool
geloescht: bool = False 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) @dc.dataclass(slots=True, kw_only=True)
class Page_Consulting_State(PageState[Module]): class Page_Consulting_State(PageState[Module]):
session: Session session: Session
@@ -271,13 +280,13 @@ class Page_Consulting_State(PageState[Module]):
locked: bool locked: bool
# TODO change to entity IDs
@dc.dataclass(slots=True, kw_only=True) @dc.dataclass(slots=True, kw_only=True)
class Page_Consulting_Linking_State(PageState[Module]): class Page_Consulting_Linking_State(PageState[Module]):
session: Session session: Session
un_id: RecId | None = None un_id: RecId | None = None
pers_id: RecId | None = None pers_id: RecId | None = None
beratungs_typ: ConsultingType beratungs_typ: ConsultingType
filter_un_ids: set[RecId] | None = None
@dc.dataclass(slots=True, kw_only=True) @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) @dc.dataclass(slots=True, kw_only=True)
class Page_CompanyProfile_State(PageState[Module]): class Page_CompanyProfile_State(PageState[Module]):
session: Session session: Session
ids: EntityIds un_id: RecId | None = None
# rec_id: RecId | None = None
# // Pydantic models # // Pydantic models
@@ -428,24 +436,36 @@ class FlatBaseModel(BaseModel):
# ** InitRec # ** InitRec
class InitRec(BaseModel): class InitRec(BaseModel):
ids: EntityIds ids: EntityIds
# rec_id: RecId | None recording_type: RecordingType
geloescht: AwareDatetime | None = None geloescht: AwareDatetime | None = None
db_data: dict[str, Any] form_data: Page_InitRec_Form_Data
class Grunderfassung_Unternehmen(FlatBaseModel): # mapping between initrec form field keys and database properties
# default in SQLAlchemy with lambda and timezone-aware datetime FIELD_DB_MAPPING_GRUNDERERFASSUNG_PERSONEN: Final[DualDict[str, str]] = DualDict(
Metadaten_erstellung: AwareDatetime | None = None Metadaten_erstellung="erstellt",
Metadaten_aktualisierung: AwareDatetime | None = None # see above Metadaten_aktualisierung="aktualisiert",
Metadaten_nutzer: str | None )
Grunderfassung_fallnummer: str
Grunderfassung_notiz: str | None
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): class Grunderfassung_Metadaten(FlatBaseModel):
# default in SQLAlchemy with lambda and timezone-aware datetime # displayed and handled separately, on database level included in 'Person'
Metadaten_erstellung: AwareDatetime | None = None Metadaten_erstellung: AwareDatetime | None = None
Metadaten_aktualisierung: AwareDatetime | None = None # see above Metadaten_aktualisierung: AwareDatetime | None = None # see above
Metadaten_nutzer: str | None Metadaten_nutzer: str | None
@@ -453,6 +473,24 @@ class Grunderfassung_Personen(FlatBaseModel):
Grunderfassung_fallnummer: str Grunderfassung_fallnummer: str
Grunderfassung_notiz: str | None 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 Projektrelevanz: Grunderfassung_Projektrelevanz
Kontaktperson: Grunderfassung_Kontaktperson Kontaktperson: Grunderfassung_Kontaktperson
Stammdaten: Grunderfassung_Stammdaten Stammdaten: Grunderfassung_Stammdaten
@@ -468,7 +506,7 @@ class Grunderfassung_PartnerSuche(BaseModel):
un_suche: int | None un_suche: int | None
person_suche: int | None person_suche: int | None
kanal_aufmerksamkeit: str | None kontaktweg: str | None
class Grunderfassung_Projektrelevanz(BaseModel): class Grunderfassung_Projektrelevanz(BaseModel):
@@ -496,6 +534,7 @@ class Grunderfassung_Kontaktperson(BaseModel):
class Grunderfassung_Stammdaten(BaseModel): class Grunderfassung_Stammdaten(BaseModel):
model_config = ConfigDict(str_strip_whitespace=True) model_config = ConfigDict(str_strip_whitespace=True)
kanal_aufmerksamkeit: str | None
titel: str | None titel: str | None
anrede_anschrift: str anrede_anschrift: str
name: str name: str
@@ -635,3 +674,38 @@ class Beratungsgespraech_Einzelgespraech(BaseModel):
erstellt: AwareDatetime | None = Field(default=None, exclude=True) erstellt: AwareDatetime | None = Field(default=None, exclude=True)
aktualisiert: AwareDatetime | None = Field(default=None, exclude=True) # see above aktualisiert: AwareDatetime | None = Field(default=None, exclude=True) # see above
geloescht: AwareDatetime | None = None 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(None)
DF_CONTACT_PERSON = get_ext_crm_contact_person(constants.Config.DB_PATH_CRM) 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", "Master",
MD_CRM, MD_CRM,
Column("ma_id", sa.Integer, nullable=False, unique=True), Column("ma_id", sa.Integer, nullable=False, unique=True),
@@ -285,9 +285,71 @@ ext_crm_master_attach: Table = Table(
schema="kontaktliste", 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: Table = Table(
"grunderfassung_unternehmen", # "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, MD_MAIN,
Column( Column(
"un_id", "un_id",
@@ -295,34 +357,21 @@ grunderfassung_unternehmen: Table = Table(
primary_key=True, primary_key=True,
autoincrement=True, autoincrement=True,
), ),
Column(
"erstellt",
UTCDateTime,
nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC),
),
Column( Column(
"geloescht", "geloescht",
UTCDateTime, UTCDateTime,
nullable=True, 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( t_grunderfassung_personen: Table = Table(
"grunderfassung_personen", "Grunderfassung_Personen",
MD_MAIN, MD_MAIN,
Column( Column(
"pers_id", "pers_id",
@@ -330,24 +379,24 @@ grunderfassung_personen: Table = Table(
primary_key=True, primary_key=True,
autoincrement=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( Column(
"geloescht", "geloescht",
UTCDateTime, UTCDateTime,
nullable=True, 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_nutzer", sa.String(20), nullable=True),
Column("Metadaten_wiedereintrittsdatum", sa.Date, nullable=True, default=None), Column("Metadaten_wiedereintrittsdatum", sa.Date, nullable=True, default=None),
Column("Arbeitserfahrung", sa.Text, nullable=True), 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_funktion_beziehung", sa.Text, nullable=True),
Column("Kontaktperson__KP_mobilfunknummer", sa.Text, nullable=True), Column("Kontaktperson__KP_mobilfunknummer", sa.Text, nullable=True),
Column("Kontaktperson__KP_name", sa.Text, nullable=True), Column("Kontaktperson__KP_name", sa.Text, nullable=True),
Column( Column("Kontaktperson__KP_name_partner", sa.Text, nullable=True),
"Kontaktperson__KP_name_partner", sa.Text, nullable=True
), # TODO: check if needed when set by trigger
Column("Kontaktperson__KP_titel", sa.Text, nullable=True), Column("Kontaktperson__KP_titel", sa.Text, nullable=True),
Column("Kontaktperson__KP_vorname", sa.Text, nullable=True), Column("Kontaktperson__KP_vorname", sa.Text, nullable=True),
Column("Projektrelevanz__relevanz", sa.Text, nullable=True), Column("Projektrelevanz__relevanz", sa.Text, nullable=True),
Column("Projektrelevanz__foerderperiode", sa.Text, nullable=True), Column("Projektrelevanz__foerderperiode", sa.Text, nullable=True),
Column("Schulbildung", sa.Text, nullable=True), Column("Schulbildung", sa.Text, nullable=True),
Column("Sprachkenntnisse", 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__PLZ", sa.Text, nullable=True),
Column("Stammdaten__anrede_anschrift", sa.Text, nullable=True), Column("Stammdaten__anrede_anschrift", sa.Text, nullable=True),
Column("Stammdaten__anzahl_kinder__alter", 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), Column("WeitereInfos__WI_meldung_institution", sa.Text, nullable=True),
) )
zuordnung_personen_unternehmen: Table = Table( t_zuordnung_personen_unternehmen: Table = Table(
"zuordnung_personen_unternehmen", "Personen_Unternehmen_Zuordnung",
MD_MAIN, MD_MAIN,
Column( Column(
"id", "id",
@@ -407,31 +455,40 @@ zuordnung_personen_unternehmen: Table = Table(
primary_key=True, primary_key=True,
autoincrement=True, autoincrement=True,
), ),
Column("un_id", sa.ForeignKey("grunderfassung_unternehmen.un_id"), nullable=False), Column("un_id", sa.ForeignKey("Unternehmen.un_id"), nullable=False),
Column("pers_id", sa.ForeignKey("grunderfassung_personen.pers_id"), nullable=False), Column("an_id", sa.Integer, nullable=False),
Column("pers_id", sa.ForeignKey("Grunderfassung_Personen.pers_id"), nullable=False),
Column( Column(
"gueltig_ab", "erstellt",
UTCDateTime, UTCDateTime,
nullable=False, nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC), default=lambda: datetime.datetime.now(datetime.UTC),
), ),
Column( Column(
"gueltig_bis", "aktualisiert",
UTCDateTime,
nullable=False,
default=lambda: datetime.datetime.now(datetime.UTC),
onupdate=lambda: datetime.datetime.now(datetime.UTC),
),
Column(
"geloescht",
UTCDateTime, UTCDateTime,
nullable=True, nullable=True,
), ),
Column("Partnersuche__kontaktweg", sa.Text, nullable=True),
Index( Index(
"uq_aktive_zuordnung", "uq_aktive_zuordnung",
"un_id", "un_id",
"pers_id", "pers_id",
unique=True, unique=True,
sqlite_where=sql.text("gueltig_bis IS NULL"), sqlite_where=sql.text("geloescht IS NULL"),
), ),
) )
beratung_vorgang: Table = Table( t_beratung_vorgang: Table = Table(
"beratung_vorgang", "Beratung_Vorgang",
MD_MAIN, MD_MAIN,
Column( Column(
"vorgang_id", "vorgang_id",
@@ -439,8 +496,8 @@ beratung_vorgang: Table = Table(
primary_key=True, primary_key=True,
autoincrement=True, autoincrement=True,
), ),
Column("un_id", sa.ForeignKey("grunderfassung_unternehmen.un_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("pers_id", sa.ForeignKey("Grunderfassung_Personen.pers_id"), nullable=True),
Column("titel", sa.Text, nullable=False), Column("titel", sa.Text, nullable=False),
Column("beratungs_typ", sa.String(20), nullable=False), Column("beratungs_typ", sa.String(20), nullable=False),
Column( Column(
@@ -464,8 +521,8 @@ beratung_vorgang: Table = Table(
) )
beratung_einzelberatung: Table = Table( t_beratung_einzelberatung: Table = Table(
"beratung_einzelberatung", "Beratung_Einzelberatung",
MD_MAIN, MD_MAIN,
Column( Column(
"beratung_id", "beratung_id",
@@ -473,7 +530,7 @@ beratung_einzelberatung: Table = Table(
primary_key=True, primary_key=True,
autoincrement=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_id", sa.Integer, nullable=False),
Column("nutzer_name", sa.String(20), nullable=False), Column("nutzer_name", sa.String(20), nullable=False),
Column( 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() GERMAN_STATE_LIST: Final[CountryList] = get_list_germany_states()
INITREC_COMP_SEARCH_HEAD = [ _INITREC_COMP_SEARCH_HEAD = [
FormField( FormField(
"Suche", "Suche",
FormFieldType.EXTENDED_DROPDOWN, FormFieldType.EXTENDED_DROPDOWN,
@@ -307,53 +307,38 @@ INITREC_COMP_SEARCH_HEAD = [
info="an_position", info="an_position",
), ),
FormField( FormField(
"Wie sind Sie auf uns aufmerksam geworden?", "Kontaktweg",
FormFieldType.DROPDOWN, FormFieldType.DROPDOWN,
required=False, required=False,
key="kanal_aufmerksamkeit", key="kontaktweg",
options=[ options=[
("Agentur für Arbeit", None), ("Telefon", None),
("Ausländerbehörde", None), ("Mail", None),
("Jobcenter", None), ("Persönliches Gespräch", 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),
], ],
), ),
] ]
INITREC_CONTACT_PERSON = [ _INITREC_CONTACT_PERSON = [
# TODO check removal
FormField( FormField(
"Name Unternehmen/Netzwerkpartner (vorausgefüllt von Suche)", "Name Unternehmen/Netzwerkpartner",
FormFieldType.TEXT, FormFieldType.TEXT,
key="KP_name_partner", key="KP_name_partner",
required=False, required=False,
placeholder="[TODO: Logik in Verbindung mit Unternehmen unklar]", readonly=False,
readonly=True,
), ),
FormField( FormField(
"Titel", "Titel",
FormFieldType.TEXT, FormFieldType.TEXT,
key="KP_titel", key="KP_titel",
required=False, required=False,
tooltip=(
"* nur wenn anrufende Person oder kontaktaufnehmende Person "
"nicht die zu beratende Person ist"
),
), ),
FormField( FormField(
"Anrede_Anschrift", "Anrede",
FormFieldType.TEXT, FormFieldType.DROPDOWN,
key="KP_anrede_anschrift", key="KP_anrede_anschrift",
required=True, required=True,
options=[("Sehr geehrte Frau", None), ("Sehr geehrter Herr", None)],
), ),
FormField( FormField(
"Name", "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( FormField(
"Titel", "Titel",
FormFieldType.TEXT, FormFieldType.TEXT,
key="titel", key="titel",
required=False, required=False,
tooltip=(
"* nur wenn anrufende Person oder kontaktaufnehmende Person "
"nicht die zu beratende Person ist"
),
), ),
FormField( FormField(
"Anrede", "Anrede",
FormFieldType.TEXT, FormFieldType.DROPDOWN,
key="anrede_anschrift", key="anrede_anschrift",
required=True, required=True,
options=[("Sehr geehrte Frau", None), ("Sehr geehrter Herr", None)],
), ),
FormField( FormField(
"Name", "Name",
@@ -427,17 +430,13 @@ INITREC_MASTER_DATA = [
"Vorname", "Vorname",
FormFieldType.TEXT, FormFieldType.TEXT,
key="vorname", key="vorname",
required=False, required=True,
), ),
FormField( FormField(
"Geburtsdatum", "Geburtsdatum",
FormFieldType.DATE, FormFieldType.DATE_FLEXIBLE,
key="geburtsdatum", key="geburtsdatum",
required=False, required=False,
tooltip=(
"* Wichtig zu erfragen, da u.a. Mindestgehaltsschwelle davon abhängt "
"(Regelung bei Ü45 Jahre)"
),
), ),
FormField( FormField(
"Herkunftsland", "Herkunftsland",
@@ -446,7 +445,6 @@ INITREC_MASTER_DATA = [
required=True, required=True,
placeholder="Suche...", placeholder="Suche...",
options=COUNTRY_LIST.for_dropdown, options=COUNTRY_LIST.for_dropdown,
tooltip=("* Wichtig zu erfragen aufgrund eventueller EU-Freizügigkeitsregelung"),
), ),
FormField( FormField(
"Staatsangehörigkeit", "Staatsangehörigkeit",
@@ -455,7 +453,6 @@ INITREC_MASTER_DATA = [
required=False, required=False,
placeholder="Suche...", placeholder="Suche...",
options=COUNTRY_LIST.for_dropdown, options=COUNTRY_LIST.for_dropdown,
tooltip=("* Wichtig zu erfragen aufgrund eventueller EU-Freizügigkeitsregelung"),
), ),
FormField( FormField(
"Rückkehrer", "Rückkehrer",
@@ -463,7 +460,6 @@ INITREC_MASTER_DATA = [
key="rueckkehrer", key="rueckkehrer",
required=False, required=False,
options=[("ja", None), ("nein", None)], options=[("ja", None), ("nein", None)],
tooltip=("* Wichtig zu erfragen aufgrund eventueller EU-Freizügigkeitsregelung"),
), ),
FormField( FormField(
"Wo befindet sich die Person?", "Wo befindet sich die Person?",
@@ -502,7 +498,6 @@ INITREC_MASTER_DATA = [
key="bundesland", key="bundesland",
required=False, required=False,
options=GERMAN_STATE_LIST.for_dropdown, options=GERMAN_STATE_LIST.for_dropdown,
tooltip=("nur wenn Inland angegeben"),
), ),
FormField( FormField(
"Land", "Land",
@@ -535,13 +530,11 @@ INITREC_MASTER_DATA = [
FormFieldType.TEXT, FormFieldType.TEXT,
key="familienstand", key="familienstand",
required=False, required=False,
tooltip="* Wichtig zu erfragen aufgrund Lebensunterhaltssicherung",
), ),
FormField( FormField(
"Anzahl Kinder", "Anzahl Kinder",
FormFieldType.DYNAMIC_DROPDOWN_NUMERIC, FormFieldType.DYNAMIC_DROPDOWN_NUMERIC,
required=False, required=False,
tooltip="* Wichtig zu erfragen aufgrund Lebensunterhaltssicherung",
key="anzahl_kinder", key="anzahl_kinder",
children=[ children=[
FormField( FormField(
@@ -549,7 +542,6 @@ INITREC_MASTER_DATA = [
FormFieldType.DROPDOWN, FormFieldType.DROPDOWN,
required=False, required=False,
options=[(str(x), None) for x in range(11)], options=[(str(x), None) for x in range(11)],
tooltip="* Wichtig zu erfragen aufgrund Lebensunterhaltssicherung",
key="anzahl", key="anzahl",
children=[ children=[
FormField("Alter Kind", FormFieldType.TEXT, key="alter"), FormField("Alter Kind", FormFieldType.TEXT, key="alter"),
@@ -560,7 +552,7 @@ INITREC_MASTER_DATA = [
] ]
INITREC_ADDITIONAL_DATA = [ _INITREC_ADDITIONAL_DATA = [
FormField( FormField(
"Deutsch als Kommunikationssprache", "Deutsch als Kommunikationssprache",
FormFieldType.DROPDOWN, FormFieldType.DROPDOWN,
@@ -593,11 +585,10 @@ INITREC_ADDITIONAL_DATA = [
("Chancenaufenthaltsrecht §104c AufenthG", None), ("Chancenaufenthaltsrecht §104c AufenthG", None),
("Sonstiges", None), ("Sonstiges", None),
], ],
tooltip="sofern nicht bekannt, unbedingt einfordern",
), ),
FormField( FormField(
"Gültigkeit Aufenthaltsstatus", "Gültigkeit Aufenthaltsstatus",
FormFieldType.DATE, FormFieldType.DATE_FLEXIBLE,
required=False, required=False,
key="WI_gueltigkeit_aufenthaltstitel", 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("Abschluss", FormFieldType.TEXT, required=False, key="SB_abschluss"),
FormField( FormField(
"Abschlussgrad laut Dokument", "Abschlussgrad laut Dokument",
@@ -673,7 +664,7 @@ INITREC_SCHOOL = [
] ]
INITREC_HIGHER_EDUCATION = [ _INITREC_HIGHER_EDUCATION = [
FormField( FormField(
"Anerkennung", "Anerkennung",
FormFieldType.TEXT, FormFieldType.TEXT,
@@ -685,10 +676,6 @@ INITREC_HIGHER_EDUCATION = [
FormFieldType.TEXT, FormFieldType.TEXT,
required=False, required=False,
key="HB_abschlussgrad", key="HB_abschlussgrad",
tooltip=(
"bitte den Titel eingeben z.B. Doktor, Diplom oder "
"Betriebswirt (Fachschulabschluss)"
),
), ),
FormField( FormField(
"Abschlussgrad laut Dokument", "Abschlussgrad laut Dokument",
@@ -707,10 +694,6 @@ INITREC_HIGHER_EDUCATION = [
FormFieldType.TEXT, FormFieldType.TEXT,
required=False, required=False,
key="HB_beruf", key="HB_beruf",
tooltip=(
"bitte spezifizieren z.B. Allgemeinmedizin, Ingenieur Maschinenbau, "
"technischer Betriebswirt Datenverarbeitung"
),
), ),
FormField( FormField(
"Land", "Land",
@@ -737,12 +720,11 @@ INITREC_HIGHER_EDUCATION = [
FormFieldType.TEXT, FormFieldType.TEXT,
required=False, required=False,
key="HB_bemerkungsfeld", key="HB_bemerkungsfeld",
tooltip="z.B. Promotionen oder den Studiengang angeben",
), ),
] ]
INITREC_WORK_EXPERIENCE = [ _INITREC_WORK_EXPERIENCE = [
FormField( FormField(
"Branche", "Branche",
FormFieldType.DROPDOWN, FormFieldType.DROPDOWN,
@@ -824,7 +806,6 @@ INITREC_WORK_EXPERIENCE = [
("Teilzeit", None), ("Teilzeit", None),
("Sonstiges", None), ("Sonstiges", None),
], ],
tooltip="Minijob, Praktikum, Wehrdienst, soziale Dienste",
), ),
FormField( FormField(
"Bemerkungsfeld", "Bemerkungsfeld",
@@ -835,7 +816,7 @@ INITREC_WORK_EXPERIENCE = [
] ]
INITREC_LANGUAGES = [ _INITREC_LANGUAGES = [
FormField( FormField(
"Sprache", "Sprache",
FormFieldType.TEXT, FormFieldType.TEXT,
@@ -877,7 +858,7 @@ INITREC_LANGUAGES = [
), ),
FormField( FormField(
"Datum des Nachweises", "Datum des Nachweises",
FormFieldType.DATE, FormFieldType.DATE_FLEXIBLE,
required=False, required=False,
key="SP_datum_nachweis", key="SP_datum_nachweis",
), ),
@@ -889,52 +870,16 @@ INITREC_LANGUAGES = [
INITREC_COMP = [ 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( FormField(
"Suche", "Suche",
FormFieldType.CUSTOM, FormFieldType.CUSTOM,
custom_widget="grunderfassung_unternehmen_suche", custom_widget="grunderfassung_unternehmen_suche",
key="Partnersuche", key="Partnersuche",
children=INITREC_COMP_SEARCH_HEAD, children=_INITREC_COMP_SEARCH_HEAD,
), ),
] ]
INITREC_PERSON = [ INITREC_METADATEN = [
FormField( FormField(
"Ersteintrag Datum", "Ersteintrag Datum",
FormFieldType.TEXT_DATETIME, FormFieldType.TEXT_DATETIME,
@@ -957,11 +902,11 @@ INITREC_PERSON = [
required=False, required=False,
key="Metadaten_nutzer", key="Metadaten_nutzer",
readonly=True, readonly=True,
ignore_get_data=True, ignore_get_data=False,
), ),
FormField( FormField(
"Wiedereintrittsdatum", "Wiedereintrittsdatum",
FormFieldType.DATE, FormFieldType.DATE_FLEXIBLE,
required=False, required=False,
key="Metadaten_wiedereintrittsdatum", key="Metadaten_wiedereintrittsdatum",
readonly=False, readonly=False,
@@ -978,6 +923,9 @@ INITREC_PERSON = [
required=False, required=False,
key="Grunderfassung_notiz", key="Grunderfassung_notiz",
), ),
]
INITREC_PERSON = [
FormField( FormField(
"Status && Projektrelevanz", "Status && Projektrelevanz",
FormFieldType.GROUP, FormFieldType.GROUP,
@@ -1009,42 +957,42 @@ INITREC_PERSON = [
"Daten Kontaktperson", "Daten Kontaktperson",
FormFieldType.GROUP, FormFieldType.GROUP,
key="Kontaktperson", key="Kontaktperson",
children=INITREC_CONTACT_PERSON, children=_INITREC_CONTACT_PERSON,
), ),
FormField( FormField(
"Stammdaten", "Stammdaten",
FormFieldType.GROUP, FormFieldType.GROUP,
key="Stammdaten", key="Stammdaten",
children=INITREC_MASTER_DATA, children=_INITREC_MASTER_DATA,
), ),
FormField( FormField(
"Weitere Informationen", "Weitere Informationen",
FormFieldType.GROUP, FormFieldType.GROUP,
key="WeitereInfos", key="WeitereInfos",
children=INITREC_ADDITIONAL_DATA, children=_INITREC_ADDITIONAL_DATA,
), ),
FormField( FormField(
"Schulbildung", "Schulbildung",
FormFieldType.DYNAMIC_LIST, FormFieldType.DYNAMIC_LIST,
children=INITREC_SCHOOL, children=_INITREC_SCHOOL,
key="Schulbildung", key="Schulbildung",
), ),
FormField( FormField(
"Studium/Ausbildung", "Studium/Ausbildung",
FormFieldType.DYNAMIC_LIST, FormFieldType.DYNAMIC_LIST,
children=INITREC_HIGHER_EDUCATION, children=_INITREC_HIGHER_EDUCATION,
key="HoehereBildung", key="HoehereBildung",
), ),
FormField( FormField(
"Arbeitserfahrung", "Arbeitserfahrung",
FormFieldType.DYNAMIC_LIST, FormFieldType.DYNAMIC_LIST,
children=INITREC_WORK_EXPERIENCE, children=_INITREC_WORK_EXPERIENCE,
key="Arbeitserfahrung", key="Arbeitserfahrung",
), ),
FormField( FormField(
"Sprachkenntnisse", "Sprachkenntnisse",
FormFieldType.DYNAMIC_LIST, FormFieldType.DYNAMIC_LIST,
children=INITREC_LANGUAGES, children=_INITREC_LANGUAGES,
key="Sprachkenntnisse", key="Sprachkenntnisse",
), ),
] ]
src/wce_crm/gui.py
+1142 -768
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 # ** GUI
logger_gui = logger_base.getChild("gui") logger_gui = logger_base.getChild("gui")
logger_gui.setLevel(logging.DEBUG) logger_gui.setLevel(logging.DEBUG)
logger_search_widget = logger_gui.getChild("search_widget") 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 = logger_gui.getChild("get_data")
logger_get_data.setLevel(logging.DEBUG) logger_get_data.setLevel(logging.WARNING)
logger_auto_form = logger_gui.getChild("get_data_auto_form")
logger_auto_form.setLevel(logging.DEBUG)
logger_custom_fields = logger_gui.getChild("custom_fields") 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 = logger_gui.getChild("page_consulting")
logger_page_consulting.setLevel(logging.DEBUG) 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 # ** Backend
logger_back = logger_base.getChild("backend") logger_back = logger_base.getChild("backend")
logger_back.setLevel(logging.DEBUG) logger_back.setLevel(logging.DEBUG)
src/wce_crm/types.py
+152 -44
View File
@@ -1,9 +1,11 @@
from __future__ import annotations from __future__ import annotations
import copy
import dataclasses as dc import dataclasses as dc
import datetime import datetime
import enum 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: if TYPE_CHECKING:
import polars as pl import polars as pl
@@ -12,13 +14,22 @@ if TYPE_CHECKING:
UserId: TypeAlias = int UserId: TypeAlias = int
RecId: TypeAlias = int RecId: TypeAlias = int
ConsId: TypeAlias = int ConsId: TypeAlias = int
IntLinkId: TypeAlias = int
ExtMaId: TypeAlias = int ExtMaId: TypeAlias = int
ExtAnId: TypeAlias = int ExtAnId: TypeAlias = int
PolarsSchema: TypeAlias = dict[str, type["pl.DataType"]] 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): class EntityType(enum.IntEnum):
METADATA = enum.auto()
COMPANY = enum.auto() COMPANY = enum.auto()
PERSON = enum.auto() PERSON = enum.auto()
@@ -30,54 +41,150 @@ class ConsultingType(enum.StrEnum):
@dc.dataclass(slots=True, kw_only=True) @dc.dataclass(slots=True, kw_only=True)
class EntityIds: class EntityIds:
un_id: RecId | None = None un_id: ExtMaId | None = None
an_id: ExtAnId | None = None
pers_id: RecId | None = None pers_id: RecId | None = None
type: dc.InitVar[EntityType | None] = None link_id: IntLinkId | None = None
_ent_type: EntityType | None = dc.field(init=False) _field_names: frozenset[str] = dc.field(init=False, repr=False, compare=False)
def __post_init__(self, type: EntityType | None) -> None: def __post_init__(self) -> None:
self._ent_type = type field_names = set(f.name for f in dc.fields(EntityIds))
field_names.remove("_field_names")
self._field_names = frozenset(field_names)
@property def set_by_name(
def ent_type(self) -> EntityType | None: self,
return self._ent_type attr_name: str,
value: int | None,
def any(self) -> bool: ) -> None:
return self.un_id is not None or self.pers_id is not None if attr_name not in self._field_names:
raise KeyError(
def all(self) -> bool: f"Attribute {attr_name} not member of this class. Valid "
return self.un_id is not None and self.pers_id is not None f"options: {self._field_names}"
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_id( setattr(self, attr_name, value)
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_
def get_id( def update(
self, self,
ent_type: EntityType, ids: EntityIds,
) -> RecId | None: ignore_none: bool = False,
if ent_type is EntityType.COMPANY: deep: bool = False,
return self.un_id ) -> 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: 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 # // database interaction data structures
@@ -137,10 +244,10 @@ class ContactPersonInfo(TypedDict):
# ** internals # ** internals
@dc.dataclass(slots=True) @dc.dataclass(slots=True)
class MainPageEntry: class MainPageEntry:
rec_id: RecId ids: EntityIds
display_name: str display_name: str
Metadaten_aktualisierung: datetime.datetime dt_updated: datetime.datetime
type: EntityType recording_type: RecordingType
@dc.dataclass(slots=True) @dc.dataclass(slots=True)
@@ -166,4 +273,5 @@ class LinkingConsultationsEntryCompany:
@dc.dataclass(slots=True) @dc.dataclass(slots=True)
class LinkingConsultationsEntryPerson: class LinkingConsultationsEntryPerson:
pers_id: RecId pers_id: RecId
un_ids: set[RecId]
display_name: str display_name: str