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

basic cleansing and hints

Browse Source
This commit is contained in:
Florian Förster
2026-07-22 11:19:07 +02:00
parent c1843702f8
commit ffe38a4ee7
4 changed files with 145 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wce_crm/backend/backend.py
+109 -107
View File
@@ -15,7 +15,7 @@ from wce_crm.constants import TIMEZONE_CEST
from wce_crm.data_models import ( from wce_crm.data_models import (
Beratungsgespraech_Einzelgespraech, Beratungsgespraech_Einzelgespraech,
Beratungsgespraech_Vorgang, Beratungsgespraech_Vorgang,
Initrec, InitRec,
) )
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 (
@@ -121,38 +121,39 @@ def initrec_comp_contact_person_search_get_info(
# // internals # // internals
def initrec_company_insert_initial_recording( # TODO check removal
data: dict[str, Any], # def initrec_company_insert_initial_recording(
) -> RecId: # data: dict[str, Any],
logger.debug("[Call backend] insert_initial_recording") # ) -> RecId:
stmt = db.grunderfassung_unternehmen.insert() # logger.debug("[Call backend] insert_initial_recording")
with db.ENGINE.begin() as conn: # stmt = db.grunderfassung_unternehmen.insert()
ret = conn.execute(stmt, data) # with db.ENGINE.begin() as conn:
# ret = conn.execute(stmt, data)
if ret.rowcount == 0: # if ret.rowcount == 0:
raise IOError("Entry was not inserted correctly") # raise IOError("Entry was not inserted correctly")
prim_keys = ret.inserted_primary_key # prim_keys = ret.inserted_primary_key
assert prim_keys # assert prim_keys
return prim_keys[0] # return prim_keys[0]
def initrec_company_update_initial_recording( # def initrec_company_update_initial_recording(
id_: RecId, # id_: RecId,
data: dict[str, Any], # data: dict[str, Any],
) -> None: # ) -> None:
logger.debug("[Call backend] update_initial_recording") # logger.debug("[Call backend] update_initial_recording")
stmt = db.grunderfassung_unternehmen.update().where( # stmt = db.grunderfassung_unternehmen.update().where(
db.grunderfassung_unternehmen.c.un_id == id_ # db.grunderfassung_unternehmen.c.un_id == id_
) # )
with db.ENGINE.begin() as conn: # with db.ENGINE.begin() as conn:
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)
@@ -212,7 +213,7 @@ def initrec_company_to_db(
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_
@@ -227,7 +228,7 @@ def initrec_company_from_db(
geloescht = data_from_db["geloescht"] geloescht = data_from_db["geloescht"]
del data_from_db["geloescht"] del data_from_db["geloescht"]
return Initrec( return InitRec(
rec_id=id_, rec_id=id_,
geloescht=geloescht, geloescht=geloescht,
db_data=data_from_db, db_data=data_from_db,
@@ -235,8 +236,8 @@ def initrec_company_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)
@@ -296,7 +297,7 @@ def initrec_person_to_db(
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.grunderfassung_personen.select().where(
db.grunderfassung_personen.c.pers_id == id_ db.grunderfassung_personen.c.pers_id == id_
@@ -311,112 +312,113 @@ def initrec_person_from_db(
geloescht = data_from_db["geloescht"] geloescht = data_from_db["geloescht"]
del data_from_db["geloescht"] del data_from_db["geloescht"]
return Initrec( return InitRec(
rec_id=id_, rec_id=id_,
geloescht=geloescht, geloescht=geloescht,
db_data=data_from_db, db_data=data_from_db,
) )
def initrec_company_get_initial_recording( # TODO check removal
id_: RecId, # def initrec_company_get_initial_recording(
) -> dict[str, Any]: # id_: RecId,
logger.debug("[Call backend] get_initial_recording") # ) -> dict[str, Any]:
stmt = db.grunderfassung_unternehmen.select().where( # logger.debug("[Call backend] get_initial_recording")
db.grunderfassung_unternehmen.c.un_id == id_ # stmt = db.grunderfassung_unternehmen.select().where(
) # db.grunderfassung_unternehmen.c.un_id == id_
with db.ENGINE.connect() as conn: # )
ret = conn.execute(stmt) # with db.ENGINE.connect() as conn:
# 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 initrec_company_delete_initial_recording( # def initrec_company_delete_initial_recording(
id_: RecId, # id_: RecId,
) -> None: # ) -> None:
logger.debug("[Call backend] delete_initial_recording") # logger.debug("[Call backend] delete_initial_recording")
stmt = db.grunderfassung_unternehmen.delete().where( # stmt = db.grunderfassung_unternehmen.delete().where(
db.grunderfassung_unternehmen.c.un_id == id_ # db.grunderfassung_unternehmen.c.un_id == id_
) # )
with db.ENGINE.begin() as conn: # with db.ENGINE.begin() as conn:
ret = conn.execute(stmt) # ret = conn.execute(stmt)
if ret.rowcount == 0: # if ret.rowcount == 0:
raise KeyError(f"Database ID {id_} not found for deletion") # raise KeyError(f"Database ID {id_} not found for deletion")
def initrec_person_insert_initial_recording( # def initrec_person_insert_initial_recording(
data: dict[str, Any], # data: dict[str, Any],
) -> RecId: # ) -> RecId:
logger.debug("[Call backend] insert_initial_recording") # logger.debug("[Call backend] insert_initial_recording")
stmt = db.grunderfassung_personen.insert() # stmt = db.grunderfassung_personen.insert()
with db.ENGINE.begin() as conn: # with db.ENGINE.begin() as conn:
ret = conn.execute(stmt, data) # ret = conn.execute(stmt, data)
if ret.rowcount == 0: # if ret.rowcount == 0:
raise IOError("Entry was not inserted correctly") # raise IOError("Entry was not inserted correctly")
prim_keys = ret.inserted_primary_key # prim_keys = ret.inserted_primary_key
assert prim_keys # assert prim_keys
return prim_keys[0] # return prim_keys[0]
def initrec_person_update_initial_recording( # def initrec_person_update_initial_recording(
id_: RecId, # id_: RecId,
data: dict[str, Any], # data: dict[str, Any],
) -> None: # ) -> None:
logger.debug("[Call backend] update_initial_recording") # logger.debug("[Call backend] update_initial_recording")
stmt = db.grunderfassung_personen.update().where( # stmt = db.grunderfassung_personen.update().where(
db.grunderfassung_personen.c.pers_id == id_ # db.grunderfassung_personen.c.pers_id == id_
) # )
with db.ENGINE.begin() as conn: # with db.ENGINE.begin() as conn:
conn.execute(stmt, data) # conn.execute(stmt, data)
def initrec_person_get_initial_recording( # def initrec_person_get_initial_recording(
id_: RecId, # id_: RecId,
) -> dict[str, Any]: # ) -> dict[str, Any]:
logger.debug("[Call backend] get_initial_recording person") # logger.debug("[Call backend] get_initial_recording person")
stmt = db.grunderfassung_personen.select().where( # stmt = db.grunderfassung_personen.select().where(
db.grunderfassung_personen.c.pers_id == id_ # db.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)
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 person initial recording obtained" # assert len(results) == 1, "more than one person 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 initrec_person_delete_initial_recording( # def initrec_person_delete_initial_recording(
id_: RecId, # id_: RecId,
) -> None: # ) -> None:
logger.debug("[Call backend] delete_initial_recording") # logger.debug("[Call backend] delete_initial_recording")
stmt = db.grunderfassung_personen.delete().where( # stmt = db.grunderfassung_personen.delete().where(
db.grunderfassung_personen.c.pers_id == id_ # db.grunderfassung_personen.c.pers_id == id_
) # )
with db.ENGINE.begin() as conn: # with db.ENGINE.begin() as conn:
ret = conn.execute(stmt) # ret = conn.execute(stmt)
if ret.rowcount == 0: # if ret.rowcount == 0:
raise KeyError(f"Database ID {id_} not found for deletion") # raise KeyError(f"Database ID {id_} not found for deletion")
@wrap_result(10) @wrap_result(10)
src/wce_crm/data_models.py
+26 -36
View File
@@ -148,47 +148,48 @@ class Module(WrapperModule, Protocol):
def save_data(self) -> None: ... def save_data(self) -> None: ...
class AutoFormInsert(Protocol): # TODO check removal
def __call__( # class AutoFormInsert(Protocol):
self, # def __call__(
data: dict[str, Any], # self,
) -> RecId: ... # data: dict[str, Any],
# ) -> RecId: ...
class AutoFormUpdate(Protocol): # class AutoFormUpdate(Protocol):
def __call__( # def __call__(
self, # self,
id_: int, # id_: int,
data: dict[str, Any], # data: dict[str, Any],
) -> None: ... # ) -> None: ...
class AutoFormGet(Protocol): # class AutoFormGet(Protocol):
def __call__( # def __call__(
self, # self,
id_: int, # id_: int,
) -> dict[str, Any]: ... # ) -> dict[str, Any]: ...
class AutoFormDelete(Protocol): # class AutoFormDelete(Protocol):
def __call__( # def __call__(
self, # self,
id_: int, # id_: int,
) -> None: ... # ) -> None: ...
class AutoFormToDb(Protocol): class AutoFormToDb(Protocol):
def __call__( def __call__(
self, self,
auto_form_data: Initrec, auto_form_data: InitRec,
) -> Initrec: ... ) -> InitRec: ...
class AutoFormFromDb(Protocol): class AutoFormFromDb(Protocol):
def __call__( def __call__(
self, self,
id_: int, id_: int,
) -> Initrec: ... ) -> InitRec: ...
@dc.dataclass(slots=True) @dc.dataclass(slots=True)
@@ -225,10 +226,6 @@ class AutoFormConfig:
model: type[FlatBaseModel] model: type[FlatBaseModel]
to_db: AutoFormToDb to_db: AutoFormToDb
from_db: AutoFormFromDb from_db: AutoFormFromDb
# data_insert: AutoFormInsert
# data_update: AutoFormUpdate
# data_get: AutoFormGet
# data_delete: AutoFormDelete
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 = True
@@ -421,14 +418,7 @@ class FlatBaseModel(BaseModel):
# ** InitRec # ** InitRec
# class Initrec_FromDb(BaseModel): class InitRec(BaseModel):
# rec_id: RecId
# geloescht: AwareDatetime | None
# Metadaten_aktualisierung: AwareDatetime
# form_data: dict[str, Any]
class Initrec(BaseModel):
rec_id: RecId | None rec_id: RecId | None
geloescht: AwareDatetime | None = None geloescht: AwareDatetime | None = None
db_data: dict[str, Any] db_data: dict[str, Any]
src/wce_crm/gui.py
+4 -4
View File
@@ -90,7 +90,7 @@ from wce_crm.data_models import (
Beratungsgespraech_Vorgang, Beratungsgespraech_Vorgang,
Grunderfassung_Personen, Grunderfassung_Personen,
Grunderfassung_Unternehmen, Grunderfassung_Unternehmen,
Initrec, InitRec,
Page_CompanyProfile_State, Page_CompanyProfile_State,
Page_Consulting_ConsultingSession_State, Page_Consulting_ConsultingSession_State,
Page_Consulting_Linking_State, Page_Consulting_Linking_State,
@@ -1678,7 +1678,7 @@ class AutoForm(QWidget):
def _load_from_pydantic( def _load_from_pydantic(
self, self,
data: Initrec, data: InitRec,
) -> None: ) -> None:
self.STATE.rec_id = data.rec_id self.STATE.rec_id = data.rec_id
self.STATE.form_data = None self.STATE.form_data = None
@@ -1700,7 +1700,7 @@ class AutoForm(QWidget):
) )
deleted_datetime = datetime.datetime.now(datetime.UTC) deleted_datetime = datetime.datetime.now(datetime.UTC)
assert self.STATE.form_data, "cannot delete entry with uninitialised form data" assert self.STATE.form_data, "cannot delete entry with uninitialised form data"
to_db = Initrec( to_db = InitRec(
rec_id=self.STATE.rec_id, rec_id=self.STATE.rec_id,
geloescht=deleted_datetime, geloescht=deleted_datetime,
db_data={}, db_data={},
@@ -1758,7 +1758,7 @@ class AutoForm(QWidget):
deleted_datetime = datetime.datetime.now(datetime.UTC) deleted_datetime = datetime.datetime.now(datetime.UTC)
try: try:
# TODO change to result pattern # TODO change to result pattern
to_db = Initrec( to_db = InitRec(
rec_id=self.STATE.rec_id, rec_id=self.STATE.rec_id,
geloescht=deleted_datetime, geloescht=deleted_datetime,
db_data=db_data, db_data=db_data,
src/wce_crm/types.py
+6
View File
@@ -28,6 +28,12 @@ class ConsultingType(enum.StrEnum):
INDIVIDUAL = enum.auto() INDIVIDUAL = enum.auto()
@dc.dataclass(slots=True)
class EntityIds:
un_id: RecId | None = None
pers_id: RecId | None = None
# // database interaction data structures # // database interaction data structures
# ** externals # ** externals
class CompanyInfo(TypedDict): class CompanyInfo(TypedDict):