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

Dev #7

Merged
foefl merged 74 commits from dev into main 2026-07-31 14:19:01 +00:00
Conversation 0 Commits 74 Files Changed 38
+1338 -983
8 changed files with 1338 additions and 983 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 868db09d94 - Show all commits
data_model/project_db_schema.aml
+10 -7
View File
@@ -7,23 +7,24 @@ Ext_Kontaktliste.Ansprechpartner
Unternehmen | Es wird nur die ID für die Kontaktliste referenziert. Alle relevanten Infos werden live aus der Kontaktliste geladen via 'ATTACH'. Unternehmen | Es wird nur die ID für die Kontaktliste referenziert. Alle relevanten Infos werden live aus der Kontaktliste geladen via 'ATTACH'.
un_id int pk -> Ext_Kontaktliste.Master(ma_id) un_id int pk -> Ext_Kontaktliste.Master(ma_id)
hinzugefuegt_am "timestamp with time zone" erstellt "timestamp with time zone"
geloescht_am "timestamp with time zone" nullable geloescht "timestamp with time zone" nullable
Grunderfassung_Personen_Unternehmen_Zuordnung Personen_Unternehmen_Zuordnung
id int pk {autoIncrement} id int pk {autoIncrement}
pers_id int pk -> Grunderfassung_Personen(pers_id) unique=uq_aktive_zuordnung 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' 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 an_id int -> Ext_Kontaktliste.Ansprechpartner(an_id) | alias 'Partnersuche__person_suche
Partnersuche__kanal_aufmerksamkeit text nullable Partnersuche__kanal_aufmerksamkeit text nullable
gueltig_ab "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)` erstellt "timestamp with time zone"=`datetime.datetime.now(datetime.UTC)`
gueltig_bis "timestamp with time zone" nullable | Soft-Delete Marker (partieller Unique-Index: WHERE deleted_at IS NULL) 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
@@ -70,6 +71,8 @@ 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.
prototypes/tests.py
+43 -16
View File
@@ -19,9 +19,48 @@ 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 NewEntityIds, RecordingType
# %% # %%
int("test") raw_ent_ids = NewEntityIds()
db_ent_ids = NewEntityIds(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 = NewEntityIds(un_id=None, an_id=2, link_id=1)
raw_ent_ids.update(db_ent_ids)
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty()=}")
print(f"{raw_ent_ids.valid()=}")
raw_ent_ids.update(NewEntityIds())
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty()=}")
print(f"{raw_ent_ids.valid()=}")
# %%
raw_ent_ids.update(NewEntityIds(un_id=None, pers_id=2))
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty()=}")
print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}")
print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}")
raw_ent_ids.update(NewEntityIds(un_id=None, an_id=2, link_id=3, pers_id=2))
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}")
print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}")
raw_ent_ids.update(NewEntityIds(un_id=2, pers_id=None, an_id=2, link_id=3))
print(f"{raw_ent_ids=}")
print(f"{raw_ent_ids.empty(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.empty(RecordingType.PERSON)=}")
print(f"{raw_ent_ids.valid(RecordingType.COMPANY)=}")
print(f"{raw_ent_ids.valid(RecordingType.PERSON)=}")
# %% # %%
db_path = constants.Config.DB_PATH_MAIN db_path = constants.Config.DB_PATH_MAIN
@@ -29,22 +68,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 +245,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
+225 -202
View File
@@ -24,11 +24,11 @@ from wce_crm.types import (
CompanyProfileConsultations, CompanyProfileConsultations,
ConsultingType, ConsultingType,
ContactPersonInfo, ContactPersonInfo,
EntityIds,
EntityType,
LinkingConsultationsEntryCompany, LinkingConsultationsEntryCompany,
LinkingConsultationsEntryPerson, LinkingConsultationsEntryPerson,
MainPageEntry, MainPageEntry,
NewEntityIds,
RecordingType,
) )
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -152,200 +152,210 @@ def initrec_comp_contact_person_search_get_info(
# conn.execute(stmt, data) # conn.execute(stmt, data)
def initrec_company_to_db( # def initrec_company_to_db(
auto_form_data: InitRec, # auto_form_data: InitRec,
) -> InitRec: # ) -> InitRec:
logger.debug("[AutoForm -- backend] Call database saving routine...") # logger.debug("[AutoForm -- backend] Call database saving routine...")
dump_data = copy.deepcopy(auto_form_data.db_data) # dump_data = copy.deepcopy(auto_form_data.db_data)
dump_data["geloescht"] = auto_form_data.geloescht # 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: # if auto_form_data.ids.un_id is None:
logger.debug("[AutoForm -- backend] Insert...") # logger.debug("[AutoForm -- backend] Insert...")
stmt = db.grunderfassung_unternehmen.insert().returning( # stmt = db.grunderfassung_unternehmen.insert().returning(
db.grunderfassung_unternehmen.c.un_id, # db.grunderfassung_unternehmen.c.un_id,
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung, # db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
db.grunderfassung_unternehmen.c.geloescht, # db.grunderfassung_unternehmen.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 excactly one returned row" # assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0] # data_from_db = from_db[0]
auto_form_data.ids.un_id = data_from_db["un_id"] # auto_form_data.ids.un_id = data_from_db["un_id"]
auto_form_data.geloescht = data_from_db["geloescht"] # auto_form_data.geloescht = data_from_db["geloescht"]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ # auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung" # "Metadaten_aktualisierung"
] # ]
logger.debug("[AutoForm -- backend] Inserted InitRec Company successfully") # logger.debug("[AutoForm -- backend] Inserted InitRec Company successfully")
else: # else:
logger.debug("[AutoForm -- backend] Update...") # logger.debug("[AutoForm -- backend] Update...")
stmt = ( # stmt = (
db.grunderfassung_unternehmen.update() # db.grunderfassung_unternehmen.update()
.where(db.grunderfassung_unternehmen.c.un_id == auto_form_data.ids.un_id) # .where(db.grunderfassung_unternehmen.c.un_id == auto_form_data.ids.un_id)
.returning( # .returning(
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung, # db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
db.grunderfassung_unternehmen.c.geloescht, # db.grunderfassung_unternehmen.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 = from_db[0]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ # auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung" # "Metadaten_aktualisierung"
] # ]
logger.debug( # logger.debug(
"[AutoForm -- backend] Updated InitRec Company with ID %d successfully", # "[AutoForm -- backend] Updated InitRec Company with ID %d successfully",
auto_form_data.ids.un_id, # auto_form_data.ids.un_id,
) # )
return auto_form_data # return auto_form_data
def initrec_company_from_db( # def initrec_company_from_db(
id_: RecId, # id_: RecId,
) -> InitRec: # ) -> InitRec:
logger.debug("[AutoForm -- backend] Call database loading routine...") # logger.debug("[AutoForm -- backend] Call database loading routine...")
stmt = db.grunderfassung_unternehmen.select().where( # stmt = db.grunderfassung_unternehmen.select().where(
db.grunderfassung_unternehmen.c.un_id == id_ # db.grunderfassung_unternehmen.c.un_id == 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_from_db = dict(from_db[0])
geloescht = data_from_db["geloescht"] # geloescht = data_from_db["geloescht"]
del data_from_db["geloescht"] # del data_from_db["geloescht"]
return InitRec( # return InitRec(
ids=EntityIds(un_id=id_), # ids=EntityIds(un_id=id_),
# rec_id=id_, # TODO remove # # rec_id=id_, # TODO remove
geloescht=geloescht, # geloescht=geloescht,
db_data=data_from_db, # db_data=data_from_db,
) # )
def initrec_person_to_db( # def initrec_person_to_db(
auto_form_data: InitRec, # auto_form_data: InitRec,
) -> InitRec: # ) -> InitRec:
logger.debug("[AutoForm -- backend] Call database saving routine...") # logger.debug("[AutoForm -- backend] Call database saving routine...")
dump_data = copy.deepcopy(auto_form_data.db_data) # dump_data = copy.deepcopy(auto_form_data.db_data)
dump_data["geloescht"] = auto_form_data.geloescht # dump_data["geloescht"] = auto_form_data.geloescht
with db.ENGINE.begin() as conn: # with db.ENGINE.begin() as conn:
if auto_form_data.ids.pers_id is None: # if auto_form_data.ids.pers_id is None:
logger.debug("[AutoForm -- backend] Insert...") # logger.debug("[AutoForm -- backend] Insert...")
stmt = sa.insert(db.grunderfassung_personen).returning( # stmt = sa.insert(db.t_grunderfassung_personen).returning(
db.grunderfassung_personen.c.pers_id, # db.t_grunderfassung_personen.c.pers_id,
db.grunderfassung_personen.c.Metadaten_aktualisierung, # db.t_grunderfassung_personen.c.Metadaten_aktualisierung,
db.grunderfassung_personen.c.geloescht, # db.t_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", # "[AutoForm -- backend] Data to insert:\nStatement: %s\n%s",
stmt_compiled, # stmt_compiled,
pformat(dump_data), # pformat(dump_data),
) # )
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 excactly one returned row" # assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0] # data_from_db = from_db[0]
auto_form_data.ids.pers_id = data_from_db["pers_id"] # auto_form_data.ids.pers_id = data_from_db["pers_id"]
auto_form_data.geloescht = data_from_db["geloescht"] # auto_form_data.geloescht = data_from_db["geloescht"]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ # auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung" # "Metadaten_aktualisierung"
] # ]
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 == auto_form_data.ids.pers_id)
.returning( # .returning(
db.grunderfassung_personen.c.Metadaten_aktualisierung, # db.t_grunderfassung_personen.c.Metadaten_aktualisierung,
db.grunderfassung_personen.c.geloescht, # db.t_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 = from_db[0]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[ # auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung" # "Metadaten_aktualisierung"
] # ]
logger.debug( # logger.debug(
"[AutoForm -- backend] Updated InitRec Person with ID %d successfully", # "[AutoForm -- backend] Updated InitRec Person with ID %d successfully",
auto_form_data.ids.pers_id, # auto_form_data.ids.pers_id,
) # )
return auto_form_data # return auto_form_data
def initrec_person_from_db( # def initrec_person_from_db(
id_: RecId, # id_: RecId,
) -> InitRec: # ) -> InitRec:
logger.debug("[AutoForm -- backend] Call database loading routine...") # logger.debug("[AutoForm -- backend] Call database loading routine...")
stmt = db.grunderfassung_personen.select().where( # stmt = db.t_grunderfassung_personen.select().where(
db.grunderfassung_personen.c.pers_id == id_ # db.t_grunderfassung_personen.c.pers_id == 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_from_db = dict(from_db[0])
geloescht = data_from_db["geloescht"] # geloescht = data_from_db["geloescht"]
del data_from_db["geloescht"] # del data_from_db["geloescht"]
return InitRec( # return InitRec(
ids=EntityIds(pers_id=id_), # ids=EntityIds(pers_id=id_),
# rec_id=id_, # TODO remove # # rec_id=id_, # TODO remove
geloescht=geloescht, # geloescht=geloescht,
db_data=data_from_db, # db_data=data_from_db,
) # )
# TODO check removal # TODO check removal
def initrec_company_get_initial_recording( # def initrec_company_get_initial_recording(
id_: RecId, # id_: RecId,
) -> dict[str, Any]: # ) -> dict[str, Any]:
logger.debug("[Call backend] get_initial_recording") # logger.debug("[Call backend] get_initial_recording")
stmt = db.grunderfassung_unternehmen.select().where( # stmt = db.grunderfassung_unternehmen.select().where(
db.grunderfassung_unternehmen.c.un_id == id_ # db.grunderfassung_unternehmen.c.un_id == id_
) # )
with db.ENGINE.connect() as conn: # with db.ENGINE.connect() as conn:
ret = conn.execute(stmt) # ret = conn.execute(stmt)
results = ret.mappings().all() # results = ret.mappings().all()
if not results: # if not results:
raise KeyError(f"Database ID {id_} not found") # raise KeyError(f"Database ID {id_} not found")
assert len(results) == 1, "more than one company initial recording obtained" # assert len(results) == 1, "more than one company initial recording obtained"
row = results[0] # row = results[0]
assert row, "row was not obtained" # assert row, "row was not obtained"
return dict(row) # return dict(row)
def page_initrec_to_db(
data: InitRec,
) -> InitRec: ...
def page_initrec_from_db(
ids: NewEntityIds,
) -> InitRec: ...
# def initrec_company_delete_initial_recording( # def initrec_company_delete_initial_recording(
@@ -444,7 +454,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 +470,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 +497,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 +507,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 +528,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 +549,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 +584,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()
@@ -604,9 +618,12 @@ def page_consulting_from_db(
# // consulting page interaction # // consulting page interaction
# TODO change to correct logic
def page_consulting_linking_companies( def page_consulting_linking_companies(
un_id: RecId | None, un_id: RecId | None,
) -> list[LinkingConsultationsEntryCompany]: ) -> list[LinkingConsultationsEntryCompany]:
raise RuntimeError("TODOs to implement")
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.grunderfassung_unternehmen.c.un_id,
@@ -624,12 +641,12 @@ def page_consulting_linking_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.Partnersuche__un_suche
== db.ext_crm_master_attach.c.ma_id, == db.t_ext_crm_master_attach.c.ma_id,
) )
) )
@@ -660,20 +677,20 @@ def page_consulting_linking_persons(
# else: list just the persons associated with the company # else: list just the persons associated with the company
person_select = sa.select( person_select = sa.select(
db.grunderfassung_personen.c.pers_id, db.t_grunderfassung_personen.c.pers_id,
db.grunderfassung_personen.c.Stammdaten__name, db.t_grunderfassung_personen.c.Stammdaten__name,
db.grunderfassung_personen.c.Stammdaten__vorname, db.t_grunderfassung_personen.c.Stammdaten__vorname,
db.grunderfassung_personen.c.Stammdaten__ort, db.t_grunderfassung_personen.c.Stammdaten__ort,
) )
if un_id is not None: if un_id is not None:
person_select = person_select.select_from( person_select = person_select.select_from(
db.grunderfassung_personen.join( db.t_grunderfassung_personen.join(
db.zuordnung_personen_unternehmen, db.t_zuordnung_personen_unternehmen,
db.grunderfassung_personen.c.pers_id db.t_grunderfassung_personen.c.pers_id
== db.zuordnung_personen_unternehmen.c.pers_id, == db.t_zuordnung_personen_unternehmen.c.pers_id,
) )
).where(db.zuordnung_personen_unternehmen.c.un_id == un_id) ).where(db.t_zuordnung_personen_unternehmen.c.un_id == un_id)
ret = conn.execute(person_select) ret = conn.execute(person_select)
person_names = ret.mappings().all() person_names = ret.mappings().all()
@@ -711,8 +728,8 @@ def page_consulting_eastablish_link(
) -> 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)
) )
@@ -725,11 +742,13 @@ def companyprofile_page_get_consultations(
) -> 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)
@@ -782,17 +801,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,8 +824,12 @@ def initrec_link_person_company(
# // main page interaction # // main page interaction
# TODO change to correct logic
def _main_page_get_company_list() -> list[MainPageEntry]: def _main_page_get_company_list() -> list[MainPageEntry]:
logger.debug("[Call backend] get_company_list") logger.debug("[Call backend] get_company_list")
raise RuntimeError("TODOs to implement")
stmt = sa.select( stmt = sa.select(
db.grunderfassung_unternehmen.c.un_id, db.grunderfassung_unternehmen.c.un_id,
db.grunderfassung_unternehmen.c.Partnersuche__un_suche, db.grunderfassung_unternehmen.c.Partnersuche__un_suche,
@@ -834,7 +857,7 @@ def _main_page_get_company_list() -> list[MainPageEntry]:
rec_id=rec_id, rec_id=rec_id,
display_name=display_name, display_name=display_name,
Metadaten_aktualisierung=datetime_akt, Metadaten_aktualisierung=datetime_akt,
type=EntityType.COMPANY, type=RecordingType.COMPANY,
) )
) )
@@ -844,11 +867,11 @@ def _main_page_get_company_list() -> list[MainPageEntry]:
def _main_page_get_person_list() -> list[MainPageEntry]: def _main_page_get_person_list() -> list[MainPageEntry]:
logger.debug("[Call backend] get_company_list") logger.debug("[Call backend] get_company_list")
stmt = sa.select( stmt = sa.select(
db.grunderfassung_personen.c.pers_id, db.t_grunderfassung_personen.c.pers_id,
db.grunderfassung_personen.c.Stammdaten__vorname, db.t_grunderfassung_personen.c.Stammdaten__vorname,
db.grunderfassung_personen.c.Stammdaten__name, db.t_grunderfassung_personen.c.Stammdaten__name,
db.grunderfassung_personen.c.Metadaten_aktualisierung, db.t_grunderfassung_personen.c.Metadaten_aktualisierung,
).where(db.grunderfassung_personen.c.geloescht.is_(None)) ).where(db.t_grunderfassung_personen.c.geloescht.is_(None))
with db.ENGINE.connect() as conn: with db.ENGINE.connect() as conn:
res = conn.execute(stmt) res = conn.execute(stmt)
@@ -871,7 +894,7 @@ def _main_page_get_person_list() -> list[MainPageEntry]:
rec_id=rec_id, rec_id=rec_id,
display_name=display_name, display_name=display_name,
Metadaten_aktualisierung=datetime_akt, Metadaten_aktualisierung=datetime_akt,
type=EntityType.PERSON, type=RecordingType.PERSON,
) )
) )
src/wce_crm/data_models.py
+46 -32
View File
@@ -19,11 +19,11 @@ 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 FormField
from wce_crm.types import EntityIds from wce_crm.types import NewEntityIds
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, ConsultingType, RecId, RecordingType, 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] = "__"
@@ -163,7 +163,7 @@ class InitRecFormToDb(Protocol):
class InitRecFormFromDb(Protocol): class InitRecFormFromDb(Protocol):
def __call__( def __call__(
self, self,
id_: int, ids: NewEntityIds,
) -> InitRec: ... ) -> InitRec: ...
@@ -177,7 +177,7 @@ def set_page_state(
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")
@@ -212,47 +212,61 @@ 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
# !! there must be exactly one InitRecForm for each recording type
@dc.dataclass(slots=True, kw_only=True) @dc.dataclass(slots=True, kw_only=True)
class InitRecFormConfig: class Page_InitRec_State(PageState["InitRecForm"]):
model: type[FlatBaseModel] session: Session
to_db: InitRecFormToDb ids: NewEntityIds = dc.field(default_factory=NewEntityIds)
from_db: InitRecFormFromDb record_type: RecordingType
# un_id: RecId | None = None
locked: bool
geloescht: bool = False
form_data: dict[RecordingType, dict[str, Any]] = dc.field(default_factory=dict)
pydantic_models: dict[RecordingType, type[FlatBaseModel]] = dc.field(default_factory=dict)
@dc.dataclass(slots=True, kw_only=True)
class Page_InitRec_Form_Config:
pydantic_model: type[FlatBaseModel]
# to_db: InitRecFormToDb # TODO remove
# from_db: InitRecFormFromDb # TODO remove
form_fields: Sequence[FormField] form_fields: Sequence[FormField]
ignored_keys: Iterable[str] = tuple() ignored_keys: Iterable[str] = tuple()
add_buttons: bool = True add_buttons: bool = False
@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: NewEntityIds
ent_type: EntityType recording_type: RecordingType
# 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) # TODO remove
class Page_InitRecCompany_State(PageState[Module]): # @dc.dataclass(slots=True, kw_only=True)
session: Session # class Page_InitRecCompany_State(PageState[Module]):
ids: EntityIds # session: Session
# un_id: RecId | None = None # ids: EntityIds
locked: bool # # un_id: RecId | None = None
# locked: bool
@dc.dataclass(slots=True, kw_only=True) # @dc.dataclass(slots=True, kw_only=True)
class Page_InitRecPerson_State(PageState[Module]): # class Page_InitRecPerson_State(PageState[Module]):
session: Session # session: Session
ids: EntityIds # ids: EntityIds
# pers_id: RecId | None = None # # pers_id: RecId | None = None
# un_id: RecId | None = None # # un_id: RecId | None = None
locked: bool # locked: bool
# TODO change to entity IDs # TODO change to entity IDs
@@ -308,7 +322,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 ids: NewEntityIds
# rec_id: RecId | None = None # rec_id: RecId | None = None
@@ -427,10 +441,10 @@ class FlatBaseModel(BaseModel):
# ** InitRec # ** InitRec
class InitRec(BaseModel): class InitRec(BaseModel):
ids: EntityIds ids: NewEntityIds
# rec_id: RecId | None # rec_id: RecId | None
geloescht: AwareDatetime | None = None geloescht: AwareDatetime | None = None
db_data: dict[str, Any] db_data: dict[RecordingType, dict[str, Any]]
class Grunderfassung_Unternehmen(FlatBaseModel): class Grunderfassung_Unternehmen(FlatBaseModel):
src/wce_crm/db.py
+129 -48
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,37 @@ 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( # Column(
"Metadaten_erstellung", # "Metadaten_erstellung",
UTCDateTime, # UTCDateTime,
nullable=True, # nullable=True,
default=lambda: datetime.datetime.now(datetime.UTC), # default=lambda: datetime.datetime.now(datetime.UTC),
), # ),
Column( # Column(
"Metadaten_aktualisierung", # "Metadaten_aktualisierung",
UTCDateTime, # UTCDateTime,
nullable=True, # nullable=True,
default=lambda: datetime.datetime.now(datetime.UTC), # default=lambda: datetime.datetime.now(datetime.UTC),
onupdate=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),
@@ -398,8 +460,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 +469,50 @@ 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(
# "gueltig_ab",
# UTCDateTime,
# nullable=False,
# default=lambda: datetime.datetime.now(datetime.UTC),
# ),
# Column(
# "gueltig_bis",
# UTCDateTime,
# 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",
@@ -464,8 +545,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",
src/wce_crm/gui.py
+750 -626
View File
File diff suppressed because it is too large Load Diff
src/wce_crm/logging.py
+15 -6
View File
@@ -30,18 +30,27 @@ 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.WARNING)
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")
src/wce_crm/types.py
+119 -45
View File
@@ -1,8 +1,10 @@
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 collections.abc import Generator
from typing import TYPE_CHECKING, TypeAlias, TypedDict from typing import TYPE_CHECKING, TypeAlias, TypedDict
if TYPE_CHECKING: if TYPE_CHECKING:
@@ -12,13 +14,16 @@ 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"]]
class EntityType(enum.IntEnum): # TODO rename: these are not entity types or entities but different forms of
# initial recordings
class RecordingType(enum.IntEnum):
COMPANY = enum.auto() COMPANY = enum.auto()
PERSON = enum.auto() PERSON = enum.auto()
@@ -29,55 +34,124 @@ class ConsultingType(enum.StrEnum):
@dc.dataclass(slots=True, kw_only=True) @dc.dataclass(slots=True, kw_only=True)
class EntityIds: class NewEntityIds:
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)
def __post_init__(self, type: EntityType | None) -> None: def update(
self._ent_type = type
@property
def ent_type(self) -> EntityType | None:
return self._ent_type
def any(self) -> bool:
return self.un_id is not None or self.pers_id is not None
def all(self) -> bool:
return self.un_id is not None and self.pers_id is not None
def valid(self) -> bool:
if self.ent_type is None:
raise TypeError("Initrec type not defined. Can not determine valid state.")
elif self.ent_type is EntityType.COMPANY:
return self.all()
elif self.ent_type is EntityType.PERSON:
return self.any()
else:
raise RuntimeError(
f"Unknown initrec type {self.ent_type}. This is not supported."
)
def set_id(
self, self,
id_: RecId | None, ids: NewEntityIds,
ent_type: EntityType, ignore_none: bool = False,
deep: bool = False,
) -> None: ) -> None:
if ent_type is EntityType.COMPANY: if not isinstance(ids, NewEntityIds):
self.un_id = id_ raise TypeError("New IDs must be have the same type as this object")
elif ent_type is EntityType.PERSON:
self.pers_id = id_
def get_id( 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, self,
ent_type: EntityType, recording_type: RecordingType | None = None,
) -> RecId | None: ) -> Generator[int | None, None, None]:
if ent_type is EntityType.COMPANY: if recording_type is None:
return self.un_id relevant_fields = (getattr(self, field.name) for field in dc.fields(self))
elif ent_type is EntityType.PERSON: elif recording_type is RecordingType.COMPANY:
return self.pers_id RELEVANT_ATTRIBUTES = {"un_id", "an_id", "link_id"}
relevant_fields = (
getattr(self, field.name)
for field in dc.fields(self)
if field.name in RELEVANT_ATTRIBUTES
)
elif recording_type is RecordingType.PERSON:
RELEVANT_ATTRIBUTES = {"pers_id"}
relevant_fields = (
getattr(self, field.name)
for field in dc.fields(self)
if field.name in RELEVANT_ATTRIBUTES
)
else:
raise RuntimeError(f"Unspecified recording type {recording_type}")
return relevant_fields
def empty(
self,
recording_type: RecordingType | None = None,
) -> bool:
relevant_fields = self._get_relevant_fields(recording_type=recording_type)
return not any(relevant_fields)
def valid(
self,
recording_type: RecordingType | None = None,
) -> bool:
relevant_fields = self._get_relevant_fields(recording_type=recording_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
@@ -140,7 +214,7 @@ class MainPageEntry:
rec_id: RecId rec_id: RecId
display_name: str display_name: str
Metadaten_aktualisierung: datetime.datetime Metadaten_aktualisierung: datetime.datetime
type: EntityType type: RecordingType
@dc.dataclass(slots=True) @dc.dataclass(slots=True)