feat: Add SQL Models for Anwendungshandbuecher (#85)

Author: hf-kklein

pyproject.toml

@@ -23,6 +23,10 @@ dependencies = [
 dynamic = ["readme", "version"]
 
 [project.optional-dependencies]
+sqlmodels = [
+  "sqlmodel>=0.0.22",
+  "sqlalchemy[mypy]>=2.0.37"
+]
 coverage = [
     "coverage==7.6.10"
 ]

requirements.txt

@@ -2,7 +2,7 @@
 # This file is autogenerated by pip-compile with Python 3.12
 # by the following command:
 #
-#    pip-compile '.\pyproject.toml'
+#    pip-compile pyproject.toml
 #
 annotated-types==0.7.0
     # via pydantic

src/fundamend/sqlmodels/__init__.py

@@ -0,0 +1,16 @@
+"""
+models that work together with SQLModel+SQLAlchemy
+you need to install fundamend[sqlmodels] to use this sub-package
+"""
+
+# the models here do NOT inherit from the original models, because I didn't manage to fix the issue that arise:
+# <frozen abc>:106: in __new__
+#     ???
+# E   TypeError: Anwendungshandbuch.__init_subclass__() takes no keyword arguments
+#
+# or
+#
+# ..\.tox\dev\Lib\site-packages\sqlmodel\main.py:697: in get_sqlalchemy_type
+#     raise ValueError(f"{type_} has no matching SQLAlchemy type")
+# E   ValueError: <class 'fundamend.models.anwendungshandbuch.Anwendungshandbuch'> has no matching SQLAlchemy type
+# => you need to keep the models in sync manually by now

src/fundamend/sqlmodels/anwendungshandbuch.py

@@ -15,6 +15,7 @@ commands = python -m pip install --upgrade pip
 deps =
     -r requirements.txt
     .[tests]
+    .[sqlmodels]
 setenv = PYTHONPATH = {toxinidir}/src
 commands = python -m pytest --basetemp={envtmpdir} {posargs} -vv
 
@@ -31,6 +32,7 @@ deps =
     {[testenv:tests]deps}
     .[linting]
     .[cli]
+    .[sqlmodels]
     # add your fixtures like e.g. pytest_datafiles here
 setenv = PYTHONPATH = {toxinidir}/src
 commands =
@@ -45,6 +47,7 @@ deps =
     {[testenv:tests]deps}
     .[type_check]
     .[cli]
+    .[sqlmodels]
 commands =
     mypy --show-error-codes src/fundamend --strict
     mypy --show-error-codes unittests --strict
@@ -68,6 +71,7 @@ deps =
     {[testenv:tests]deps}
     .[coverage]
     .[cli]
+    .[sqlmodels]
 setenv = PYTHONPATH = {toxinidir}/src
 commands =
     coverage run -m pytest --basetemp={envtmpdir} {posargs}

tox.ini

@@ -0,0 +1,7 @@
+from pathlib import Path
+
+private_submodule_root = Path(__file__).parent.parent / "xml-migs-and-ahbs"
+
+
+def is_private_submodule_checked_out() -> bool:
+    return any(private_submodule_root.iterdir())

unittests/conftest.py

@@ -5,11 +5,9 @@
 
 from fundamend.reader import AhbReader, MigReader
 
-data_path: Path = Path(__file__).parent.parent / "xml-migs-and-ahbs"
-
+from .conftest import is_private_submodule_checked_out
 
-def private_submodule_is_checked_out() -> bool:
-    return any(data_path.iterdir())
+data_path: Path = Path(__file__).parent.parent / "xml-migs-and-ahbs"
 
 
 @pytest.mark.parametrize(
@@ -98,15 +96,15 @@ def private_submodule_is_checked_out() -> bool:
     ],
 )
 def test_read_ahb_xml(ahb_xml_file_path: Path, expected_date: date) -> None:
-    if not private_submodule_is_checked_out():
+    if not is_private_submodule_checked_out():
         pytest.skip("Skipping test because of missing private submodule")
     reader = AhbReader(ahb_xml_file_path)
     actual = reader.get_publishing_date()
     assert actual == expected_date
 
 
 def test_deserializing_all_ahbs() -> None:
-    if not private_submodule_is_checked_out():
+    if not is_private_submodule_checked_out():
         pytest.skip("Skipping test because of missing private submodule")
     for ahb_file_path in data_path.rglob("**/*AHB*.xml"):
         reader = AhbReader(ahb_file_path)
@@ -199,15 +197,15 @@ def test_deserializing_all_ahbs() -> None:
     ],
 )
 def test_read_mig_xml(mig_xml_file_path: Path, expected_date: date) -> None:
-    if not private_submodule_is_checked_out():
+    if not is_private_submodule_checked_out():
         pytest.skip("Skipping test because of missing private submodule")
     reader = MigReader(mig_xml_file_path)
     actual = reader.get_publishing_date()
     assert actual == expected_date
 
 
 def test_deserializing_all_migs() -> None:
-    if not private_submodule_is_checked_out():
+    if not is_private_submodule_checked_out():
         pytest.skip("Skipping test because of missing private submodule")
     for mig_file_path in data_path.rglob("**/*MIG*.xml"):
         reader = MigReader(mig_file_path)

unittests/test_ahb_serialisierung.py

@@ -0,0 +1,61 @@
+"""
+we try to fill a database using kohlrahbi[sqlmodels] and the data from the machine-readable AHB submodule
+"""
+
+from pathlib import Path
+from typing import Generator
+
+import pytest
+from sqlmodel import Session, SQLModel, create_engine
+
+from fundamend import AhbReader
+from fundamend.models.anwendungshandbuch import Anwendungshandbuch as PydanticAnwendunghandbuch
+from fundamend.sqlmodels.anwendungshandbuch import Anwendungshandbuch as SqlAnwendungshandbuch
+
+from .conftest import is_private_submodule_checked_out
+
+
+@pytest.fixture()
+def sqlite_session(tmp_path: Path) -> Generator[Session, None, None]:
+    database_path = tmp_path / "test.db"
+    engine = create_engine(f"sqlite:///{database_path}")
+    SQLModel.metadata.drop_all(engine)
+    SQLModel.metadata.create_all(engine)
+    with Session(bind=engine) as session:
+        yield session
+        session.commit()
+        session.flush()
+    print(f"Wrote all data to {database_path.absolute()}")
+
+
+def _load_anwendungshandbuch_ahb_to_and_from_db(
+    session: Session, pydantic_ahb: PydanticAnwendunghandbuch
+) -> PydanticAnwendunghandbuch:
+    sql_ahb = SqlAnwendungshandbuch.from_model(pydantic_ahb)
+    session.add(sql_ahb)
+    session.commit()
+    session.refresh(sql_ahb)
+    pydantic_ahb_after_roundtrip = sql_ahb.to_model()
+    return pydantic_ahb_after_roundtrip
+
+
+def test_sqlmodels_single_anwendungshandbuch(sqlite_session: Session) -> None:
+    ahb = AhbReader(
+        Path(__file__).parent / "example_files" / "UTILTS_AHB_1.1d_Konsultationsfassung_2024_04_02.xml"
+    ).read()
+    roundtrip_abb = _load_anwendungshandbuch_ahb_to_and_from_db(sqlite_session, ahb)
+    ahb_json = ahb.model_dump(mode="json")
+    roundtrip_json = roundtrip_abb.model_dump(mode="json")
+    assert ahb_json == roundtrip_json  # in pycharm the error message is much better when comparing plain python dicts
+    assert roundtrip_abb == ahb
+
+
+def test_sqlmodels_all_anwendungshandbuch(sqlite_session: Session) -> None:
+    if not is_private_submodule_checked_out():
+        pytest.skip("Skipping test because of missing private submodule")
+    private_submodule_root = Path(__file__).parent.parent / "xml-migs-and-ahbs"
+    assert private_submodule_root.exists() and private_submodule_root.is_dir()
+    for ahb_file_path in private_submodule_root.rglob("**/*AHB*.xml"):
+        ahb = AhbReader(ahb_file_path).read()
+        roundtrip_abb = _load_anwendungshandbuch_ahb_to_and_from_db(sqlite_session, ahb)
+        assert roundtrip_abb == ahb