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

added entity ID container

Browse Source
This commit is contained in:
Florian Förster
2026-07-22 15:26:15 +02:00
parent ffe38a4ee7
commit 6e04a7485a
4 changed files with 293 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/wce_crm/backend/backend.py
+44 -38
View File
@@ -24,7 +24,8 @@ from wce_crm.types import (
CompanyProfileConsultations,
ConsultingType,
ContactPersonInfo,
InitRecType,
EntityIds,
EntityType,
LinkingConsultationsEntryCompany,
LinkingConsultationsEntryPerson,
MainPageEntry,
@@ -160,7 +161,7 @@ def initrec_company_to_db(
dump_data["geloescht"] = auto_form_data.geloescht
with db.ENGINE.begin() as conn:
if auto_form_data.rec_id is None:
if auto_form_data.ids.un_id is None:
logger.debug("[AutoForm -- backend] Insert...")
stmt = db.grunderfassung_unternehmen.insert().returning(
@@ -169,25 +170,23 @@ def initrec_company_to_db(
db.grunderfassung_unternehmen.c.geloescht,
)
ret = conn.execute(stmt, dump_data)
if ret.rowcount == 0:
raise IOError("Entry was not inserted correctly")
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0]
auto_form_data.rec_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.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
logger.debug("[AutoForm -- backend] Inserted InitRec successfully")
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.rec_id)
.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,
@@ -204,8 +203,8 @@ def initrec_company_to_db(
]
logger.debug(
"[AutoForm -- backend] Updated InitRec with ID %d successfully",
auto_form_data.rec_id,
"[AutoForm -- backend] Updated InitRec Company with ID %d successfully",
auto_form_data.ids.un_id,
)
return auto_form_data
@@ -229,7 +228,8 @@ def initrec_company_from_db(
del data_from_db["geloescht"]
return InitRec(
rec_id=id_,
ids=EntityIds(un_id=id_),
# rec_id=id_, # TODO remove
geloescht=geloescht,
db_data=data_from_db,
)
@@ -244,34 +244,39 @@ def initrec_person_to_db(
dump_data["geloescht"] = auto_form_data.geloescht
with db.ENGINE.begin() as conn:
if auto_form_data.rec_id is None:
if auto_form_data.ids.pers_id is None:
logger.debug("[AutoForm -- backend] Insert...")
stmt = db.grunderfassung_personen.insert().returning(
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))
logger.debug(
"[AutoForm -- backend] Data to insert:\nStatement: %s\n%s",
stmt_compiled,
pformat(dump_data),
)
ret = conn.execute(stmt, dump_data)
if ret.rowcount == 0:
raise IOError("Entry was not inserted correctly")
from_db = ret.mappings().fetchall()
assert len(from_db) == 1, "expected excactly one returned row"
data_from_db = from_db[0]
auto_form_data.rec_id = data_from_db["un_id"]
auto_form_data.ids.pers_id = data_from_db["pers_id"]
auto_form_data.geloescht = data_from_db["geloescht"]
auto_form_data.db_data["Metadaten_aktualisierung"] = data_from_db[
"Metadaten_aktualisierung"
]
logger.debug("[AutoForm -- backend] Inserted InitRec successfully")
logger.debug("[AutoForm -- backend] Inserted InitRec Person successfully")
else:
logger.debug("[AutoForm -- backend] Update...")
stmt = (
db.grunderfassung_personen.update()
.where(db.grunderfassung_personen.c.pers_id == auto_form_data.rec_id)
.where(db.grunderfassung_personen.c.pers_id == auto_form_data.ids.pers_id)
.returning(
db.grunderfassung_personen.c.Metadaten_aktualisierung,
db.grunderfassung_personen.c.geloescht,
@@ -288,8 +293,8 @@ def initrec_person_to_db(
]
logger.debug(
"[AutoForm -- backend] Updated InitRec with ID %d successfully",
auto_form_data.rec_id,
"[AutoForm -- backend] Updated InitRec Person with ID %d successfully",
auto_form_data.ids.pers_id,
)
return auto_form_data
@@ -313,33 +318,34 @@ def initrec_person_from_db(
del data_from_db["geloescht"]
return InitRec(
rec_id=id_,
ids=EntityIds(pers_id=id_),
# rec_id=id_, # TODO remove
geloescht=geloescht,
db_data=data_from_db,
)
# 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)
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(
@@ -828,7 +834,7 @@ def _main_page_get_company_list() -> list[MainPageEntry]:
rec_id=rec_id,
display_name=display_name,
Metadaten_aktualisierung=datetime_akt,
type=InitRecType.COMPANY,
type=EntityType.COMPANY,
)
)
@@ -865,7 +871,7 @@ def _main_page_get_person_list() -> list[MainPageEntry]:
rec_id=rec_id,
display_name=display_name,
Metadaten_aktualisierung=datetime_akt,
type=InitRecType.PERSON,
type=EntityType.PERSON,
)
)