feat: Add view v_ahbtabellen on top of ahb_hierarchy_materialized (#118)

---------

Co-authored-by: Konstantin <[email protected]>
Co-authored-by: kevin <[email protected]>

Author: 3 people

src/fundamend/sqlmodels/ahbtabellen_view.py

@@ -0,0 +1,59 @@
+"""
+This module contains the SQLModel class for the AHBesser view and a function to create it.
+If you never heard about ahbesser, you can safely ignore this module.
+https://github.com/Hochfrequenz/ahbesser
+"""
+
+import logging
+from pathlib import Path
+from uuid import UUID
+
+from efoli import EdifactFormatVersion
+from sqlalchemy.sql.functions import func
+from sqlmodel import Field, Session, SQLModel, select
+
+from fundamend.sqlmodels.internals import _execute_bare_sql
+
+_logger = logging.getLogger(__name__)
+
+
+def create_ahbtabellen_view(session: Session) -> None:
+    """
+    Create a view for the AHB-Tabellen application: https://github.com/Hochfrequenz/ahbesser
+    This assumes that create_db_and_populate_with_ahb_view has already been called.
+    If you don't know what ahbesser is, you can safely ignore this function.
+    """
+    _execute_bare_sql(session=session, path_to_sql_commands=Path(__file__).parent / "create_ahbtabellen_view.sql")
+    number_of_rows = session.scalar(
+        select(func.count(AhbTabellenLine.id))  # type:ignore[arg-type] # pylint:disable=not-callable
+    )
+    _logger.info(
+        "There are %d rows in the AHBTabellen view %s",
+        number_of_rows,
+        AhbTabellenLine.__tablename__,
+    )
+
+
+class AhbTabellenLine(SQLModel, table=True):
+    """
+    Model that represents thew view used by ahbesser. It's created by executing 'create_ahbtabellen_view(session)'
+    """
+
+    __tablename__ = "v_ahbtabellen"
+    id: UUID = Field(primary_key=True)
+    format_version: EdifactFormatVersion = Field()
+    pruefidentifikator: str = Field()
+    path: str = Field()
+    id_path: str = Field()
+    direction: str = Field()
+    description: str = Field()
+    segmentgroup_key: str | None = Field()
+    segment_code: str | None = Field()
+    data_element: str | None = Field()
+    qualifier: str | None = Field()
+    line_ahb_status: str | None = Field()
+    line_name: str | None = Field()
+    sort_path: str = Field()
+
+
+__all__ = ["create_ahbtabellen_view", "AhbTabellenLine"]

src/fundamend/sqlmodels/ahbview.py

@@ -30,6 +30,7 @@
     SegmentGroup,
     SegmentGroupLink,
 )
+from fundamend.sqlmodels.internals import _execute_bare_sql
 
 _logger = logging.getLogger(__name__)
 
@@ -39,24 +40,8 @@ def create_ahb_view(session: Session) -> None:
     Create a materialized view for the Anwendungshandbücher using a SQLAlchemy session.
     Warning: This is only tested for SQLite!
     """
-    path_to_sql_command = Path(__file__).parent / "materialize_ahb_view.sql"
-
-    with open(path_to_sql_command, "r", encoding="utf-8") as sql_file:
-        bare_sql = sql_file.read()
-
-    bare_statements = bare_sql.split(";")
-
-    for bare_statement in bare_statements:
-        statement = bare_statement.strip()
-        if statement:
-            try:
-                session.execute(sqlalchemy.text(statement))
-            except sqlalchemy.exc.IntegrityError:
-                if " UNIQUE " in bare_statement:
-                    session.execute(sqlalchemy.text(bare_statement.replace(" UNIQUE ", " ")))
-                else:
-                    raise
-    session.commit()
+    _execute_bare_sql(session=session, path_to_sql_commands=Path(__file__).parent / "materialize_ahb_view.sql")
+
     number_of_inserted_rows = session.scalar(
         select(func.count(AhbHierarchyMaterialized.id))  # type:ignore[arg-type] # pylint:disable=not-callable #
     )

src/fundamend/sqlmodels/anwendungshandbuch.py

@@ -20,7 +20,6 @@
     # sqlmodel is only an optional dependency when fundamend is used to fill a database
     raise
 
-
 from fundamend.models.anwendungshandbuch import Anwendungsfall as PydanticAnwendungsfall
 from fundamend.models.anwendungshandbuch import Anwendungshandbuch as PydanticAnwendungshandbuch
 from fundamend.models.anwendungshandbuch import Bedingung as PydanticBedingung

src/fundamend/sqlmodels/create_ahbtabellen_view.sql

@@ -0,0 +1,30 @@
+-- Assume that materialize_ahb_view.sql has been executed already.
+-- Then we can create another view on top of ahb_hierarchy_materialized.
+-- This is especially useful for our ahbesser frontend (aka "AHB-Tabellen")
+
+DROP TABLE IF EXISTS v_ahbtabellen; -- this is because sqlmodel tries to create a table first... it doesn't know that this is just a view. bit dirty but ok.
+DROP VIEW IF EXISTS v_ahbtabellen;
+CREATE VIEW v_ahbtabellen as
+SELECT id                                                                                                   as id,
+       edifact_format_version                                                                               as format_version,
+       pruefidentifikator                                                                                   as pruefidentifikator,
+       path,
+       id_path,
+       kommunikation_von                                                                                    as direction,
+       beschreibung                                                                                         as description,
+       'SG' || segmentgroup_id                                                                              as segmentgroup_key, -- eg 'SG6'
+       segment_id                                                                                           as segment_code,     -- e.g 'NAD'
+       dataelement_id                                                                                       as data_element,     -- e.g 'D_3035'
+       --CASE
+       --    WHEN dataelement_id IS NOT NULL THEN SUBSTR(dataelement_id, 3)
+       --    END                                                                                              AS dataelement_without_leading_d_, -- e.g '3035'
+       code_value                                                                                           as qualifier,
+
+       coalesce(code_ahb_status, coalesce(dataelement_ahb_status,
+                                          coalesce(segment_ahb_status, segmentgroup_ahb_status)))           as line_ahb_status,  -- e.g. 'Muss [28] ∧ [64]'
+       coalesce(code_name, coalesce(dataelement_name, coalesce(dataelementgroup_name,
+                                                               coalesce(segment_name, segmentgroup_name)))) as line_name,        -- e.g. 'Datums- oder Uhrzeit- oder Zeitspannen-Format, Code' or 'Produkt-Daten für Lieferant relevant'
+       sort_path                                                                                            as sort_path
+-- the bedingung column is still missing, but we'll solve this one separately
+FROM ahb_hierarchy_materialized
+WHERE TYPE != 'dataelementgroup' AND (TYPE != 'dataelement' OR dataelement_ahb_status IS NOT Null) ;

src/fundamend/sqlmodels/internals.py

@@ -0,0 +1,34 @@
+"""internal helper functions"""
+
+from pathlib import Path
+
+try:
+    import sqlalchemy
+    from sqlmodel import Session
+except ImportError as import_error:
+    import_error.msg += "; Did you install fundamend[sqlmodels] or did you try to import from fundamend.models instead?"
+    # sqlmodel is only an optional dependency when fundamend is used to fill a database
+    raise
+
+
+def _execute_bare_sql(session: Session, path_to_sql_commands: Path) -> None:
+    """
+    Execute bare SQL from the path_to_sqlcommands in the given SQLAlchemy session.
+    """
+
+    with open(path_to_sql_commands, "r", encoding="utf-8") as sql_file:
+        bare_sql = sql_file.read()
+
+    bare_statements = bare_sql.split(";")
+
+    for bare_statement in bare_statements:
+        statement = bare_statement.strip()
+        if statement:
+            try:
+                session.execute(sqlalchemy.text(statement))
+            except sqlalchemy.exc.IntegrityError:
+                if " UNIQUE " in bare_statement:
+                    session.execute(sqlalchemy.text(bare_statement.replace(" UNIQUE ", " ")))
+                else:
+                    raise
+    session.commit()