[APP-352] feat: wip - convert NBS SR 02

apps/report-execution/src/libraries/nbs_sr_02.py

@@ -0,0 +1,33 @@
+from src.db_transaction import Transaction
+from src.models import ReportResult, TimeRange
+
+
+def execute(
+    trx: Transaction,
+    subset_query: str,
+    data_source_name: str,
+    time_range: TimeRange | None = None,
+    **kwargs,
+):
+    """Basic tabular report. Executes the query described by the data
+    source and filters (executing subset_sql) and returns the table.
+
+    Conversion notes:
+    * Lifted the size check to a global check and refined the env var names
+    * Date formatting still needs to be figured out
+    """
+    content = trx.execute(
+        f'WITH subset as ({subset_query})\n' +
+        'SELECT state, county, phc_code_short_desc, sum(group_case_cnt) as cases\n' +
+        'FROM subset\n' + 
+        'GROUP BY state, county, phc_code_short_desc'
+        )
+
+    description = 'SR2: Counts of Reportable Diseases by County for Selected Time frame'
+    if time_range is not None:
+        description += f'\n{time_range.start} - {time_range.end}'
+
+    # TODO: datetimes are formatted '%0m/%0d/%Y %0H:%0M:%0S' in sas. # noqa: FIX002
+    # Should this be a util/post processing/sql step in python land?
+
+    return ReportResult(content_type='table', content=content, description=description)

apps/report-execution/tests/integration/libraries/nbs_sr_02.py

@@ -0,0 +1,72 @@
+import pytest
+
+from src.execute_report import execute_report
+from src.models import ReportSpec
+
+
+@pytest.mark.usefixtures('setup_containers')
+@pytest.mark.integration
+class TestIntegrationNbsSr02Library:
+    """Integration tests for the nbs_custom library."""
+
+    def test_execute_report_with_time_range(self):
+        report_spec = ReportSpec.model_validate(
+            {
+                'version': 1,
+                'is_export': True,
+                'is_builtin': True,
+                'report_title': 'NBS Custom',
+                'library_name': 'nbs_sr_02',
+                # Filter operator is used here as it is a stable, small table
+                'data_source_name': '[NBS_ODSE].[dbo].[PHCDemographic]',
+                'subset_query': 'SELECT * FROM [NBS_ODSE].[dbo].[PHCDemographic]',
+                'time_range': {'start': '2024-01-01', 'end': '2024-12-31'},
+            }
+        )
+
+        result = execute_report(report_spec)
+        assert result.description == (
+            'SR2: Counts of Reportable Diseases by County for Selected Time frame\n'
+            + '2024-01-01 - 2024-12-31'
+        )
+        assert result.content_type == 'table'
+
+        assert len(result.content.data) >= 1
+        assert len(result.content.data[0]) == 4
+        assert len(result.content.data[0]) == len(result.content.columns)
+
+        record = None
+        for row in result.content.data:
+            if (
+                row[0] == 'Georgia'
+                and row[1] == 'Gwinnett County'
+                and row[2] == 'Pertussis'
+            ):
+                record = row
+                break
+
+        assert record is not None
+        assert record[3] >= 1
+
+    def test_execute_report_without_time_range(self):
+        report_spec = ReportSpec.model_validate(
+            {
+                'version': 1,
+                'is_export': True,
+                'is_builtin': True,
+                'report_title': 'NBS Custom',
+                'library_name': 'nbs_custom',
+                # Filter operator is used here as it is a stable, small table
+                'data_source_name': '[NBS_ODSE].[dbo].[Filter_operator]',
+                'subset_query': 'SELECT * FROM [NBS_ODSE].[dbo].[Filter_operator]',
+            }
+        )
+
+        result = execute_report(report_spec)
+        assert result.description == (
+            'Custom Report For Table: [NBS_ODSE].[dbo].[Filter_operator]'
+        )
+        assert result.content_type == 'table'
+
+        assert len(result.content.data) == 11
+        assert len(result.content.data[0]) == len(result.content.columns)