prototypes/tests.py

@@ -44,7 +44,7 @@ for r in res:
 
 
 # %%
-backend.front_get_company_list()
+backend._main_page_get_company_list()
 
 
 # %%

src/wce_crm/backend/backend.py

@@ -1,70 +1,18 @@
 from __future__ import annotations
 
-import dataclasses as dc
 import datetime
-from typing import TYPE_CHECKING, Any, TypedDict, cast
+from typing import TYPE_CHECKING, Any, cast
 
 import polars as pl
 import sqlalchemy as sql
 
 from wce_crm import db
+from wce_crm.constants import TIMEZONE_CEST
 from wce_crm.logging import logger_back as logger
+from wce_crm.types import CompanyInfo, ContactPersonInfo, InitRecType, MainPageEntry
 
 if TYPE_CHECKING:
-    from wce_crm.types import ExtAnId, ExtMaId, InitRecId
-
-
-# // externals
-class CompanyInfo(TypedDict):
-    ma_id: str
-    wce_id: str
-    ma_unternehmensname: str
-    ma_branche: str
-    ma_strasse: str
-    ma_hausnummer: str
-    ma_plz: str
-    ma_ort: str
-    ma_plz_postfach: str
-    ma_postfach: str
-    ma_website: str
-    ma_mail: str
-    ma_telefonnummer: str
-    ma_faxnummer: str
-    ma_ersteintrag_datum: str
-    ma_aktualisierung_datum: str
-    ma_aktualisierung_nutzer: str
-    ma_sollprozess: str
-    ma_auslaendische_mitarbeiter: str
-    ma_quelle_information: str
-    ma_bemerkung: str
-    ma_kontakt: str
-    ma_schlagworte: str
-    ma_archiviert: str
-
-
-class ContactPersonInfo(TypedDict):
-    an_id: str
-    ma_id: str
-    wce_id: str
-    st_id: str
-    an_sachgebiet: str
-    an_anrede: str
-    an_titel: str
-    an_nachname: str
-    an_vorname: str
-    an_position: str
-    an_mail: str
-    an_festnetz: str
-    an_mobil: str
-    an_faxnummer: str
-    an_hauptansprechpartner: str
-    an_anrede_anschrift: str
-    an_bemerkung: str
-    an_aktualisierung_datum: str
-    an_aktualisierung_nutzer: str
-    an_letztes_kontaktdatum: str
-    an_ersteintrag_datum: str
-    an_archiviert: str
+    from wce_crm.types import ExtAnId, ExtMaId, RecId
 
 
 def _transform_for_gui_output(
@@ -156,7 +104,7 @@ def initrec_comp_contact_person_search_get_info(
 # // internals
 def initrec_company_insert_initial_recording(
     data: dict[str, Any],
-) -> InitRecId:
+) -> RecId:
     logger.debug("[Call backend] insert_initial_recording")
     stmt = db.grunderfassung_unternehmen.insert()
     with db.ENGINE.begin() as conn:
@@ -172,7 +120,7 @@ def initrec_company_insert_initial_recording(
 
 
 def initrec_company_update_initial_recording(
-    id_: InitRecId,
+    id_: RecId,
     data: dict[str, Any],
 ) -> None:
     logger.debug("[Call backend] update_initial_recording")
@@ -184,7 +132,7 @@ def initrec_company_update_initial_recording(
 
 
 def initrec_company_get_initial_recording(
-    id_: InitRecId,
+    id_: RecId,
 ) -> dict[str, Any]:
     logger.debug("[Call backend] get_initial_recording")
     stmt = db.grunderfassung_unternehmen.select().where(
@@ -203,7 +151,7 @@ def initrec_company_get_initial_recording(
 
 
 def initrec_company_delete_initial_recording(
-    id_: InitRecId,
+    id_: RecId,
 ) -> None:
     logger.debug("[Call backend] delete_initial_recording")
     stmt = db.grunderfassung_unternehmen.delete().where(
@@ -218,7 +166,7 @@ def initrec_company_delete_initial_recording(
 
 def initrec_person_insert_initial_recording(
     data: dict[str, Any],
-) -> InitRecId:
+) -> RecId:
     logger.debug("[Call backend] insert_initial_recording")
     stmt = db.grunderfassung_personen.insert()
     with db.ENGINE.begin() as conn:
@@ -234,7 +182,7 @@ def initrec_person_insert_initial_recording(
 
 
 def initrec_person_update_initial_recording(
-    id_: InitRecId,
+    id_: RecId,
     data: dict[str, Any],
 ) -> None:
     logger.debug("[Call backend] update_initial_recording")
@@ -246,7 +194,7 @@ def initrec_person_update_initial_recording(
 
 
 def initrec_person_get_initial_recording(
-    id_: InitRecId,
+    id_: RecId,
 ) -> dict[str, Any]:
     logger.debug("[Call backend] get_initial_recording")
     stmt = db.grunderfassung_personen.select().where(
@@ -265,7 +213,7 @@ def initrec_person_get_initial_recording(
 
 
 def initrec_person_delete_initial_recording(
-    id_: InitRecId,
+    id_: RecId,
 ) -> None:
     logger.debug("[Call backend] delete_initial_recording")
     stmt = db.grunderfassung_personen.delete().where(
@@ -278,56 +226,86 @@ def initrec_person_delete_initial_recording(
         raise KeyError(f"Database ID {id_} not found for deletion")
 
 
-@dc.dataclass(slots=True)
-class FrontpageCompany:
-    erfassung_id: InitRecId
-    name: str
-    Metadaten_aktualisierung: datetime.datetime
-    is_company: bool
-
-
-# TODO separate functions for companies and person: focus on companies first
-def front_get_company_list() -> list[FrontpageCompany]:
+# // main page interaction
+def _main_page_get_company_list() -> list[MainPageEntry]:
     logger.debug("[Call backend] get_company_list")
     stmt = sql.select(
         db.grunderfassung_unternehmen.c.un_id,
         db.grunderfassung_unternehmen.c.Partnersuche__un_suche,
-        # db.grunderfassung_unternehmen.c.Stammdaten__vorname,
-        # db.grunderfassung_unternehmen.c.Stammdaten__name,
         db.grunderfassung_unternehmen.c.Metadaten_aktualisierung,
-    ).order_by(db.grunderfassung_unternehmen.c.Metadaten_aktualisierung.desc())
+    )
 
     with db.ENGINE.connect() as conn:
         res = conn.execute(stmt)
 
-    front_page_companies: list[FrontpageCompany] = []
+    main_page_companies: list[MainPageEntry] = []
 
     for entry in res.mappings():
-        erfassung_id = entry["un_id"]
-        ma_id = entry["Partnersuche__un_suche"]
+        rec_id = entry["un_id"]
+        assert rec_id, "no RecID defined"
+        ma_id_external = entry["Partnersuche__un_suche"]
+        assert ma_id_external is not None, "external MA ID is NULL"
+        datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"])
+        datetime_akt = datetime_akt.astimezone(TIMEZONE_CEST)
+
+        comp_info = initrec_comp_search_get_info(ma_id_external)
+        display_name = comp_info["ma_unternehmensname"]
+
+        main_page_companies.append(
+            MainPageEntry(
+                rec_id=rec_id,
+                display_name=display_name,
+                Metadaten_aktualisierung=datetime_akt,
+                type=InitRecType.COMPANY,
+            )
+        )
+
+    return main_page_companies
+
+
+def _main_page_get_person_list() -> list[MainPageEntry]:
+    logger.debug("[Call backend] get_company_list")
+    stmt = sql.select(
+        db.grunderfassung_personen.c.pers_id,
+        db.grunderfassung_personen.c.Stammdaten__vorname,
+        db.grunderfassung_personen.c.Stammdaten__name,
+        db.grunderfassung_personen.c.Metadaten_aktualisierung,
+    )
+
+    with db.ENGINE.connect() as conn:
+        res = conn.execute(stmt)
+
+    main_page_persons: list[MainPageEntry] = []
+
+    for entry in res.mappings():
+        rec_id = entry["pers_id"]
+        assert rec_id, "no RecID defined"
         datetime_akt = cast(datetime.datetime, entry["Metadaten_aktualisierung"])
         datetime_akt = datetime_akt.astimezone()
 
-        name: str
-        is_company: bool
-        if ma_id is not None:
-            comp_info = initrec_comp_search_get_info(ma_id)
-            name = comp_info["ma_unternehmensname"]
-            is_company = True
-        else:
         surname: str = entry["Stammdaten__name"]
         first_name: str = entry["Stammdaten__vorname"]
         names_to_join = [n for n in (first_name, surname) if n]
-            name = " ".join(names_to_join)
-            is_company = False
+        display_name = " ".join(names_to_join)
 
-        front_page_companies.append(
-            FrontpageCompany(
-                erfassung_id=erfassung_id,
-                name=name,
+        main_page_persons.append(
+            MainPageEntry(
+                rec_id=rec_id,
+                display_name=display_name,
                 Metadaten_aktualisierung=datetime_akt,
-                is_company=is_company,
+                type=InitRecType.PERSON,
             )
         )
 
-    return front_page_companies
+    return main_page_persons
+
+
+def main_page_get_entries() -> list[MainPageEntry]:
+    companies = _main_page_get_company_list()
+    persons = _main_page_get_person_list()
+
+    all_entries = companies + persons
+
+    all_entries.sort(key=lambda x: x.Metadaten_aktualisierung, reverse=True)
+
+    return all_entries

src/wce_crm/types.py

@@ -1,13 +1,84 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, TypeAlias
+import dataclasses as dc
+import datetime
+import enum
+from typing import TYPE_CHECKING, TypeAlias, TypedDict
 
 if TYPE_CHECKING:
     import polars as pl
 
 UserId: TypeAlias = int
-InitRecId: TypeAlias = int
+RecId: TypeAlias = int
 ExtMaId: TypeAlias = int
 ExtAnId: TypeAlias = int
 
 PolarsSchema: TypeAlias = dict[str, type["pl.DataType"]]
+
+
+class InitRecType(enum.IntEnum):
+    COMPANY = enum.auto()
+    PERSON = enum.auto()
+
+
+# // database interaction data structures
+# ** externals
+class CompanyInfo(TypedDict):
+    ma_id: str
+    wce_id: str
+    ma_unternehmensname: str
+    ma_branche: str
+    ma_strasse: str
+    ma_hausnummer: str
+    ma_plz: str
+    ma_ort: str
+    ma_plz_postfach: str
+    ma_postfach: str
+    ma_website: str
+    ma_mail: str
+    ma_telefonnummer: str
+    ma_faxnummer: str
+    ma_ersteintrag_datum: str
+    ma_aktualisierung_datum: str
+    ma_aktualisierung_nutzer: str
+    ma_sollprozess: str
+    ma_auslaendische_mitarbeiter: str
+    ma_quelle_information: str
+    ma_bemerkung: str
+    ma_kontakt: str
+    ma_schlagworte: str
+    ma_archiviert: str
+
+
+class ContactPersonInfo(TypedDict):
+    an_id: str
+    ma_id: str
+    wce_id: str
+    st_id: str
+    an_sachgebiet: str
+    an_anrede: str
+    an_titel: str
+    an_nachname: str
+    an_vorname: str
+    an_position: str
+    an_mail: str
+    an_festnetz: str
+    an_mobil: str
+    an_faxnummer: str
+    an_hauptansprechpartner: str
+    an_anrede_anschrift: str
+    an_bemerkung: str
+    an_aktualisierung_datum: str
+    an_aktualisierung_nutzer: str
+    an_letztes_kontaktdatum: str
+    an_ersteintrag_datum: str
+    an_archiviert: str
+
+
+# ** internals
+@dc.dataclass(slots=True)
+class MainPageEntry:
+    rec_id: RecId
+    display_name: str
+    Metadaten_aktualisierung: datetime.datetime
+    type: InitRecType