generated from dopt-python/py311
prepare new loading scheme for auto form data (not working yet)
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import datetime
|
||||
from pprint import pformat
|
||||
from typing import TYPE_CHECKING, Any, cast
|
||||
@@ -13,6 +14,7 @@ from wce_crm.constants import TIMEZONE_CEST
|
||||
from wce_crm.data_models import (
|
||||
Beratungsgespraech_Einzelgespraech,
|
||||
Beratungsgespraech_Vorgang,
|
||||
Initrec_FromDb,
|
||||
)
|
||||
from wce_crm.logging import logger_back as logger
|
||||
from wce_crm.types import (
|
||||
@@ -26,6 +28,7 @@ from wce_crm.types import (
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from wce_crm.data_models import AutoForm_State, Initrec
|
||||
from wce_crm.types import ConsId, ExtAnId, ExtMaId, RecId
|
||||
|
||||
|
||||
@@ -145,6 +148,241 @@ def initrec_company_update_initial_recording(
|
||||
conn.execute(stmt, data)
|
||||
|
||||
|
||||
def initrec_company_to_db(
|
||||
auto_form_data: Initrec,
|
||||
) -> Initrec:
|
||||
logger.debug("[AutoForm -- backend] Call database saving routine...")
|
||||
|
||||
dump_data = copy.deepcopy(auto_form_data.form_data)
|
||||
dump_data["geloescht"] = auto_form_data.geloescht
|
||||
|
||||
with db.ENGINE.begin() as conn:
|
||||
if auto_form_data.rec_id is None:
|
||||
logger.debug("[AutoForm -- backend] Insert...")
|
||||
|
||||
stmt = db.grunderfassung_unternehmen.insert().returning(
|
||||
db.grunderfassung_unternehmen.c.un_id,
|
||||
db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
|
||||
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.geloescht = data_from_db["geloescht"]
|
||||
auto_form_data.form_data["Metadaten_aktualisierung"] = data_from_db[
|
||||
"Metadaten_aktualisierung"
|
||||
]
|
||||
|
||||
logger.debug("[AutoForm -- backend] Inserted InitRec successfully")
|
||||
|
||||
else:
|
||||
logger.debug("[AutoForm -- backend] Update...")
|
||||
stmt = (
|
||||
db.beratung_vorgang.update()
|
||||
.where(db.grunderfassung_unternehmen.c.un_id == auto_form_data.rec_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.geloescht = data_from_db["geloescht"]
|
||||
auto_form_data.form_data["Metadaten_aktualisierung"] = data_from_db[
|
||||
"Metadaten_aktualisierung"
|
||||
]
|
||||
|
||||
logger.debug(
|
||||
"[AutoForm -- backend] Updated InitRec with ID %d successfully",
|
||||
auto_form_data.rec_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(
|
||||
rec_id=id_,
|
||||
geloescht=geloescht,
|
||||
form_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.form_data)
|
||||
dump_data["geloescht"] = auto_form_data.geloescht
|
||||
|
||||
with db.ENGINE.begin() as conn:
|
||||
if auto_form_data.rec_id is None:
|
||||
logger.debug("[AutoForm -- backend] Insert...")
|
||||
|
||||
stmt = db.grunderfassung_personen.insert().returning(
|
||||
db.grunderfassung_personen.c.pers_id,
|
||||
db.grunderfassung_personen.c.Metadaten_aktualisierung,
|
||||
db.grunderfassung_personen.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.geloescht = data_from_db["geloescht"]
|
||||
auto_form_data.form_data["Metadaten_aktualisierung"] = data_from_db[
|
||||
"Metadaten_aktualisierung"
|
||||
]
|
||||
|
||||
logger.debug("[AutoForm -- backend] Inserted InitRec successfully")
|
||||
|
||||
else:
|
||||
logger.debug("[AutoForm -- backend] Update...")
|
||||
stmt = (
|
||||
db.beratung_vorgang.update()
|
||||
.where(db.grunderfassung_personen.c.pers_id == auto_form_data.rec_id)
|
||||
.returning(
|
||||
db.grunderfassung_personen.c.Metadaten_aktualisierung,
|
||||
db.grunderfassung_personen.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.geloescht = data_from_db["geloescht"]
|
||||
auto_form_data.form_data["Metadaten_aktualisierung"] = data_from_db[
|
||||
"Metadaten_aktualisierung"
|
||||
]
|
||||
|
||||
logger.debug(
|
||||
"[AutoForm -- backend] Updated InitRec with ID %d successfully",
|
||||
auto_form_data.rec_id,
|
||||
)
|
||||
|
||||
return auto_form_data
|
||||
|
||||
|
||||
def initrec_person_from_db(
|
||||
id_: RecId,
|
||||
) -> Initrec:
|
||||
logger.debug("[AutoForm -- backend] Call database loading routine...")
|
||||
stmt = db.grunderfassung_personen.select().where(
|
||||
db.grunderfassung_personen.c.pers_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(
|
||||
rec_id=id_,
|
||||
geloescht=geloescht,
|
||||
form_data=data_from_db,
|
||||
)
|
||||
|
||||
|
||||
# def initrec_person_to_db(
|
||||
# auto_form_data: Initrec,
|
||||
# ) -> Initrec_FromDb:
|
||||
# logger.debug("[AutoForm -- backend] Call database saving routine...")
|
||||
|
||||
# dump_data = copy.deepcopy(auto_form_data.form_data)
|
||||
# dump_data["geloescht"] = auto_form_data.geloescht
|
||||
|
||||
# return_from_db: Initrec_FromDb
|
||||
|
||||
# with db.ENGINE.begin() as conn:
|
||||
# if auto_form_data.rec_id is None:
|
||||
# # insert new "Vorgang"
|
||||
# logger.debug("[AutoForm -- backend] Insert...")
|
||||
|
||||
# stmt = db.grunderfassung_personen.insert().returning(
|
||||
# db.grunderfassung_personen.c.pers_id,
|
||||
# db.grunderfassung_personen.c.Metadaten_aktualisierung,
|
||||
# db.grunderfassung_personen.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]
|
||||
# return_from_db = Initrec_FromDb(
|
||||
# rec_id=data_from_db["un_id"],
|
||||
# geloescht=data_from_db["geloescht"],
|
||||
# Metadaten_aktualisierung=data_from_db["Metadaten_aktualisierung"],
|
||||
# )
|
||||
|
||||
# logger.debug("[AutoForm -- backend] Inserted InitRec successfully")
|
||||
|
||||
# else:
|
||||
# logger.debug("[AutoForm -- backend] Update...")
|
||||
# stmt = (
|
||||
# db.beratung_vorgang.update()
|
||||
# .where(db.grunderfassung_personen.c.pers_id == auto_form_data.rec_id)
|
||||
# .returning(
|
||||
# db.grunderfassung_personen.c.Metadaten_aktualisierung,
|
||||
# db.grunderfassung_personen.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]
|
||||
|
||||
# return_from_db = Initrec_FromDb(
|
||||
# rec_id=auto_form_data.rec_id,
|
||||
# geloescht=data_from_db["geloescht"],
|
||||
# Metadaten_aktualisierung=data_from_db["Metadaten_aktualisierung"],
|
||||
# )
|
||||
|
||||
# logger.debug(
|
||||
# "[AutoForm -- backend] Updated InitRec with ID %d successfully",
|
||||
# auto_form_data.rec_id,
|
||||
# )
|
||||
|
||||
# return return_from_db
|
||||
|
||||
|
||||
def initrec_company_get_initial_recording(
|
||||
id_: RecId,
|
||||
) -> dict[str, Any]:
|
||||
|
||||
Reference in New Issue
Block a user