small refactoring with more precise declarations

src/wce_crm/backend/backend.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import dataclasses as dc
 import datetime
-from typing import Any, TypedDict, cast
+from typing import Any, TypeAlias, TypedDict, cast
 
 import polars as pl
 import sqlalchemy as sql
@@ -10,6 +10,8 @@ import sqlalchemy as sql
 from wce_crm import db
 from wce_crm.logging import logger_back as logger
 
+InitRecId: TypeAlias = int
+
 
 class CompanyInfo(TypedDict):
     ma_id: str
@@ -151,43 +153,63 @@ def initrec_comp_contact_person_search_get_info(
 
 def initrec_insert_initial_recording(
     data: dict[str, Any],
-) -> None:
+) -> InitRecId:
     logger.debug("[Call backend] insert_initial_recording")
-    stmt = db.grunderfassung.insert().values(data)
+    stmt = db.grunderfassung.insert()
     with db.ENGINE.begin() as conn:
-        conn.execute(stmt)
+        ret = conn.execute(stmt, data)
+
+    if ret.rowcount == 0:
+        raise IOError("Entry was not inserted correctly")
+
+    prim_keys = ret.inserted_primary_key
+    assert prim_keys
+
+    return prim_keys[0]
 
 
 def initrec_update_initial_recording(
-    id_: int,
+    id_: InitRecId,
     data: dict[str, Any],
 ) -> None:
     logger.debug("[Call backend] update_initial_recording")
-    stmt = (
-        db.grunderfassung.update().where(db.grunderfassung.c.erfassung_id == id_).values(data)
-    )
+    stmt = db.grunderfassung.update().where(db.grunderfassung.c.erfassung_id == id_)
     with db.ENGINE.begin() as conn:
-        conn.execute(stmt)
+        conn.execute(stmt, data)
 
 
 def initrec_get_initial_recording(
-    id_: int,
+    id_: InitRecId,
 ) -> dict[str, Any]:
     logger.debug("[Call backend] get_initial_recording")
     stmt = db.grunderfassung.select().where(db.grunderfassung.c.erfassung_id == id_)
     with db.ENGINE.begin() as conn:
         ret = conn.execute(stmt)
 
-    row = ret.fetchone()
-    if row is None:
+    if ret.rowcount == 0:
         raise KeyError(f"Database ID {id_} not found")
 
+    row = ret.fetchone()
+    assert row, "row was not obtained"
+
     return row._asdict()
 
 
+def initrec_delete_initial_recording(
+    id_: InitRecId,
+) -> None:
+    logger.debug("[Call backend] delete_initial_recording")
+    stmt = db.grunderfassung.delete().where(db.grunderfassung.c.erfassung_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")
+
+
 @dc.dataclass(slots=True)
 class FrontpageCompany:
-    erfassung_id: int
+    erfassung_id: InitRecId
     # ma_id: int
     name: str
     Metadaten_aktualisierung: datetime.datetime