feat: add JSON column w/ Kommunikationsrichtungen to Anwendungsfall table (fix cyclic import problem with Kommunikationseinrichtung model) (#181)

Author: hf-kklein

src/fundamend/models/anwendungshandbuch.py

@@ -8,6 +8,8 @@
 from efoli import EdifactFormat
 
 from fundamend.models.base import FundamendBaseModel
+from fundamend.models.kommunikationsrichtung import Kommunikationsrichtung
+from fundamend.utils import parse_kommunikation_von
 
 
 class Code(FundamendBaseModel):
@@ -121,15 +123,6 @@ class SegmentGroup(FundamendBaseModel):
     elements: tuple["Segment | SegmentGroup", ...]
 
 
-class Kommunikationsrichtung(FundamendBaseModel):
-    """
-    a strongly typed representation of the 'Kommunikation_von' attribute of anwendungsfall
-    """
-
-    sender: str  #: e.g. "NB"
-    empfaenger: str  #: e.g. "MSB"
-
-
 class Anwendungsfall(FundamendBaseModel):
     """
     One 'Anwendungsfall', indicated by `<AWF>` tag, corresponds to one Prüfidentifikator or type of Message
@@ -158,6 +151,15 @@ def is_outdated(self) -> bool:
         """
         return "##alt##" in self.pruefidentifikator.lower()  # table flip moment
 
+    @property
+    def kommunikationsrichtungen(self) -> list[Kommunikationsrichtung] | None:
+        """
+        the parsed 'kommunikation_von' attribute or None if it's unparsable (l)or outdated
+        """
+        if self.is_outdated:
+            return None
+        return parse_kommunikation_von(self.kommunikation_von)
+
 
 class Bedingung(FundamendBaseModel):
     """Ein ConditionKeyConditionText Mapping"""

src/fundamend/models/kommunikationsrichtung.py

@@ -0,0 +1,16 @@
+"""contains the Kommunikationsrichtung model"""
+
+from fundamend.models.base import FundamendBaseModel
+
+
+# needs to be separate file/module to avoid circular imports
+class Kommunikationsrichtung(FundamendBaseModel):
+    """
+    a strongly typed representation of the 'Kommunikation_von' attribute of anwendungsfall
+    """
+
+    sender: str  #: e.g. "NB"
+    empfaenger: str  #: e.g. "MSB"
+
+
+__all__ = ["Kommunikationsrichtung"]

src/fundamend/sqlmodels/anwendungshandbuch.py

@@ -13,7 +13,7 @@
 
 
 try:
-    from sqlalchemy import CheckConstraint, UniqueConstraint
+    from sqlalchemy import JSON, CheckConstraint, Column, UniqueConstraint
     from sqlmodel import Field, Relationship, SQLModel
 except ImportError as import_error:
     import_error.msg += "; Did you install fundamend[sqlmodels] or did you try to import from fundamend.models instead?"
@@ -341,6 +341,15 @@ class Anwendungsfall(SQLModel, table=True):
     pruefidentifikator: str = Field(index=True)  #: e.g. '25001'
     beschreibung: str = Field(index=True)  #: e.g. 'Berechnungsformel'
     kommunikation_von: str  #: e.g. 'NB an MSB / LF'
+    kommunikationsrichtungen: list[dict[str, str]] | None = Field(default=None, sa_column=Column(JSON))
+    """
+    JSON column containing a list of dicts with keys 'sender' and 'empfaenger'.
+    The columns value can be deserialized as list[Kommunikationsrichtung].
+    The column contains no more information than the stringly typed 'kommunikation_von' column alone,
+    but it's parsed already and hence easier to use and query.
+    """
+    # we use a json column because setting up a separate table and FK relationships for this seems overkill
+    # https://stackoverflow.com/a/70659555/10009545
     format: EdifactFormat = Field(index=True)  #: e.g. 'UTILTS'
     segments: list[Segment] = Relationship(back_populates="anwendungsfall")
     segment_groups: list[SegmentGroup] = Relationship(back_populates="anwendungsfall")
@@ -356,6 +365,11 @@ def from_model(cls, model: PydanticAnwendungsfall, position: Optional[int] = Non
             pruefidentifikator=model.pruefidentifikator,
             beschreibung=model.beschreibung,
             kommunikation_von=model.kommunikation_von,
+            kommunikationsrichtungen=(
+                [kr.model_dump(mode="json") for kr in model.kommunikationsrichtungen]
+                if model.kommunikationsrichtungen is not None
+                else None
+            ),
             format=model.format,
             position=position,
         )
@@ -373,6 +387,7 @@ def to_model(self) -> PydanticAnwendungsfall:
             pruefidentifikator=self.pruefidentifikator,
             beschreibung=self.beschreibung,
             kommunikation_von=self.kommunikation_von,
+            # the kommunikationsrichtungen are a computed property, so we reconstruct it from the kommunikation_von str
             format=self.format,
             elements=tuple(
                 x.to_model()

src/fundamend/utils.py

@@ -5,7 +5,7 @@
 import re
 from typing import Optional
 
-from fundamend.models.anwendungshandbuch import Kommunikationsrichtung
+from fundamend.models.kommunikationsrichtung import Kommunikationsrichtung
 
 
 def lstrip(prefix: str, text: str) -> str:

unittests/test_sqlmodels_anwendungshandbuch.py

@@ -8,11 +8,13 @@
 
 import pytest
 from efoli import EdifactFormatVersion
+from pydantic import RootModel
 from sqlmodel import Session, SQLModel, create_engine, select
 from syrupy.assertion import SnapshotAssertion
 
 from fundamend import AhbReader
 from fundamend.models.anwendungshandbuch import Anwendungshandbuch as PydanticAnwendunghandbuch
+from fundamend.models.kommunikationsrichtung import Kommunikationsrichtung
 from fundamend.sqlmodels import AhbHierarchyMaterialized
 from fundamend.sqlmodels import Anwendungshandbuch as SqlAnwendungshandbuch
 from fundamend.sqlmodels import create_ahb_view, create_db_and_populate_with_ahb_view
@@ -89,6 +91,9 @@ def test_sqlmodels_all_anwendungshandbuch(sqlite_session: Session) -> None:
         assert roundtrip_abb == ahb
 
 
+_Kommunikationsrichtungen = RootModel[list[Kommunikationsrichtung]]
+
+
 def test_sqlmodels_all_anwendungshandbuch_with_ahb_view(sqlite_session: Session) -> None:
     if not is_private_submodule_checked_out():
         pytest.skip("Skipping test because of missing private submodule")
@@ -97,6 +102,20 @@ def test_sqlmodels_all_anwendungshandbuch_with_ahb_view(sqlite_session: Session)
     for ahb_file_path in private_submodule_root.rglob("**/*AHB*.xml"):
         ahb = AhbReader(ahb_file_path).read()
         sql_ahb = SqlAnwendungshandbuch.from_model(ahb)
+        for awf, sql_awf in zip(
+            (x for x in ahb.anwendungsfaelle if not x.is_outdated),
+            # this is because outdated AWF are not included in the SQL model;
+            # see the SqlAnwendungshandbuch.from_model implementation.
+            sorted(sql_ahb.anwendungsfaelle, key=lambda _awf: _awf.position or 0),
+        ):
+            # this is for https://github.com/Hochfrequenz/xml-fundamend-python/issues/173
+            if awf.kommunikationsrichtungen is not None and any(awf.kommunikationsrichtungen):
+                sql_kommunikationsrichtungen = _Kommunikationsrichtungen.model_validate(
+                    sql_awf.kommunikationsrichtungen
+                ).root
+                assert sql_kommunikationsrichtungen == awf.kommunikationsrichtungen
+            else:
+                assert sql_awf.kommunikationsrichtungen is None or not any(sql_awf.kommunikationsrichtungen)
         sqlite_session.add(sql_ahb)
     sqlite_session.commit()
     create_ahb_view(session=sqlite_session)

unittests/test_utils.py

@@ -4,7 +4,8 @@
 import pytest
 
 from fundamend import AhbReader
-from fundamend.models.anwendungshandbuch import Anwendungsfall, Kommunikationsrichtung
+from fundamend.models.anwendungshandbuch import Anwendungsfall
+from fundamend.models.kommunikationsrichtung import Kommunikationsrichtung
 from fundamend.utils import parse_kommunikation_von
 
 from .conftest import is_private_submodule_checked_out