From ffe38a4ee726ac206aa73e0ad609a23509e70a96 Mon Sep 17 00:00:00 2001 From: foefl Date: Wed, 22 Jul 2026 11:19:07 +0200 Subject: [PATCH] basic cleansing and hints --- src/wce_crm/backend/backend.py | 216 +++++++++++++++++---------------- src/wce_crm/data_models.py | 62 ++++------ src/wce_crm/gui.py | 8 +- src/wce_crm/types.py | 6 + 4 files changed, 145 insertions(+), 147 deletions(-) diff --git a/src/wce_crm/backend/backend.py b/src/wce_crm/backend/backend.py index bed6f23..b4fb30e 100644 --- a/src/wce_crm/backend/backend.py +++ b/src/wce_crm/backend/backend.py @@ -15,7 +15,7 @@ from wce_crm.constants import TIMEZONE_CEST from wce_crm.data_models import ( Beratungsgespraech_Einzelgespraech, Beratungsgespraech_Vorgang, - Initrec, + InitRec, ) from wce_crm.logging import logger_back as logger from wce_crm.types import ( @@ -121,38 +121,39 @@ def initrec_comp_contact_person_search_get_info( # // internals -def initrec_company_insert_initial_recording( - data: dict[str, Any], -) -> RecId: - logger.debug("[Call backend] insert_initial_recording") - stmt = db.grunderfassung_unternehmen.insert() - with db.ENGINE.begin() as conn: - ret = conn.execute(stmt, data) +# TODO check removal +# def initrec_company_insert_initial_recording( +# data: dict[str, Any], +# ) -> RecId: +# logger.debug("[Call backend] insert_initial_recording") +# stmt = db.grunderfassung_unternehmen.insert() +# with db.ENGINE.begin() as conn: +# ret = conn.execute(stmt, data) - if ret.rowcount == 0: - raise IOError("Entry was not inserted correctly") +# if ret.rowcount == 0: +# raise IOError("Entry was not inserted correctly") - prim_keys = ret.inserted_primary_key - assert prim_keys +# prim_keys = ret.inserted_primary_key +# assert prim_keys - return prim_keys[0] +# return prim_keys[0] -def initrec_company_update_initial_recording( - id_: RecId, - data: dict[str, Any], -) -> None: - logger.debug("[Call backend] update_initial_recording") - stmt = db.grunderfassung_unternehmen.update().where( - db.grunderfassung_unternehmen.c.un_id == id_ - ) - with db.ENGINE.begin() as conn: - conn.execute(stmt, data) +# def initrec_company_update_initial_recording( +# id_: RecId, +# data: dict[str, Any], +# ) -> None: +# logger.debug("[Call backend] update_initial_recording") +# stmt = db.grunderfassung_unternehmen.update().where( +# db.grunderfassung_unternehmen.c.un_id == id_ +# ) +# with db.ENGINE.begin() as conn: +# conn.execute(stmt, data) def initrec_company_to_db( - auto_form_data: Initrec, -) -> Initrec: + auto_form_data: InitRec, +) -> InitRec: logger.debug("[AutoForm -- backend] Call database saving routine...") dump_data = copy.deepcopy(auto_form_data.db_data) @@ -212,7 +213,7 @@ def initrec_company_to_db( def initrec_company_from_db( id_: RecId, -) -> Initrec: +) -> InitRec: logger.debug("[AutoForm -- backend] Call database loading routine...") stmt = db.grunderfassung_unternehmen.select().where( db.grunderfassung_unternehmen.c.un_id == id_ @@ -227,7 +228,7 @@ def initrec_company_from_db( geloescht = data_from_db["geloescht"] del data_from_db["geloescht"] - return Initrec( + return InitRec( rec_id=id_, geloescht=geloescht, db_data=data_from_db, @@ -235,8 +236,8 @@ def initrec_company_from_db( def initrec_person_to_db( - auto_form_data: Initrec, -) -> Initrec: + auto_form_data: InitRec, +) -> InitRec: logger.debug("[AutoForm -- backend] Call database saving routine...") dump_data = copy.deepcopy(auto_form_data.db_data) @@ -296,7 +297,7 @@ def initrec_person_to_db( def initrec_person_from_db( id_: RecId, -) -> Initrec: +) -> InitRec: logger.debug("[AutoForm -- backend] Call database loading routine...") stmt = db.grunderfassung_personen.select().where( db.grunderfassung_personen.c.pers_id == id_ @@ -311,112 +312,113 @@ def initrec_person_from_db( geloescht = data_from_db["geloescht"] del data_from_db["geloescht"] - return Initrec( + return InitRec( rec_id=id_, geloescht=geloescht, db_data=data_from_db, ) -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) +# 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") +# 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" +# assert len(results) == 1, "more than one company initial recording obtained" - row = results[0] - assert row, "row was not obtained" +# row = results[0] +# assert row, "row was not obtained" - return dict(row) +# 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) +# 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") +# 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) +# 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") +# if ret.rowcount == 0: +# raise IOError("Entry was not inserted correctly") - prim_keys = ret.inserted_primary_key - assert prim_keys +# prim_keys = ret.inserted_primary_key +# assert prim_keys - return prim_keys[0] +# 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_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) +# 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") +# 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" +# assert len(results) == 1, "more than one person initial recording obtained" - row = results[0] - assert row, "row was not obtained" +# row = results[0] +# assert row, "row was not obtained" - return dict(row) +# 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) +# 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") +# if ret.rowcount == 0: +# raise KeyError(f"Database ID {id_} not found for deletion") @wrap_result(10) diff --git a/src/wce_crm/data_models.py b/src/wce_crm/data_models.py index 10e19ee..a008ccf 100644 --- a/src/wce_crm/data_models.py +++ b/src/wce_crm/data_models.py @@ -148,47 +148,48 @@ class Module(WrapperModule, Protocol): def save_data(self) -> None: ... -class AutoFormInsert(Protocol): - def __call__( - self, - data: dict[str, Any], - ) -> RecId: ... +# 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 AutoFormUpdate(Protocol): +# def __call__( +# self, +# id_: int, +# data: dict[str, Any], +# ) -> None: ... -class AutoFormGet(Protocol): - def __call__( - self, - id_: int, - ) -> dict[str, Any]: ... +# class AutoFormGet(Protocol): +# def __call__( +# self, +# id_: int, +# ) -> dict[str, Any]: ... -class AutoFormDelete(Protocol): - def __call__( - self, - id_: int, - ) -> None: ... +# class AutoFormDelete(Protocol): +# def __call__( +# self, +# id_: int, +# ) -> None: ... class AutoFormToDb(Protocol): def __call__( self, - auto_form_data: Initrec, - ) -> Initrec: ... + auto_form_data: InitRec, + ) -> InitRec: ... class AutoFormFromDb(Protocol): def __call__( self, id_: int, - ) -> Initrec: ... + ) -> InitRec: ... @dc.dataclass(slots=True) @@ -225,10 +226,6 @@ class AutoFormConfig: model: type[FlatBaseModel] to_db: AutoFormToDb from_db: AutoFormFromDb - # data_insert: AutoFormInsert - # data_update: AutoFormUpdate - # data_get: AutoFormGet - # data_delete: AutoFormDelete form_fields: Sequence[FormField] ignored_keys: Iterable[str] = tuple() add_buttons: bool = True @@ -421,14 +418,7 @@ class FlatBaseModel(BaseModel): # ** InitRec -# class Initrec_FromDb(BaseModel): -# rec_id: RecId -# geloescht: AwareDatetime | None -# Metadaten_aktualisierung: AwareDatetime -# form_data: dict[str, Any] - - -class Initrec(BaseModel): +class InitRec(BaseModel): rec_id: RecId | None geloescht: AwareDatetime | None = None db_data: dict[str, Any] diff --git a/src/wce_crm/gui.py b/src/wce_crm/gui.py index 873e506..2031744 100644 --- a/src/wce_crm/gui.py +++ b/src/wce_crm/gui.py @@ -90,7 +90,7 @@ from wce_crm.data_models import ( Beratungsgespraech_Vorgang, Grunderfassung_Personen, Grunderfassung_Unternehmen, - Initrec, + InitRec, Page_CompanyProfile_State, Page_Consulting_ConsultingSession_State, Page_Consulting_Linking_State, @@ -1678,7 +1678,7 @@ class AutoForm(QWidget): def _load_from_pydantic( self, - data: Initrec, + data: InitRec, ) -> None: self.STATE.rec_id = data.rec_id self.STATE.form_data = None @@ -1700,7 +1700,7 @@ class AutoForm(QWidget): ) deleted_datetime = datetime.datetime.now(datetime.UTC) assert self.STATE.form_data, "cannot delete entry with uninitialised form data" - to_db = Initrec( + to_db = InitRec( rec_id=self.STATE.rec_id, geloescht=deleted_datetime, db_data={}, @@ -1758,7 +1758,7 @@ class AutoForm(QWidget): deleted_datetime = datetime.datetime.now(datetime.UTC) try: # TODO change to result pattern - to_db = Initrec( + to_db = InitRec( rec_id=self.STATE.rec_id, geloescht=deleted_datetime, db_data=db_data, diff --git a/src/wce_crm/types.py b/src/wce_crm/types.py index 9f9f943..dcf3e05 100644 --- a/src/wce_crm/types.py +++ b/src/wce_crm/types.py @@ -28,6 +28,12 @@ class ConsultingType(enum.StrEnum): INDIVIDUAL = enum.auto() +@dc.dataclass(slots=True) +class EntityIds: + un_id: RecId | None = None + pers_id: RecId | None = None + + # // database interaction data structures # ** externals class CompanyInfo(TypedDict):