feat: csv for powerBI integration (#3327)

backend/benefit/applications/api/v1/power_bi_integration_views.py

@@ -0,0 +1,86 @@
+import logging
+
+from django.db.models import QuerySet
+from django.http import StreamingHttpResponse
+from django.utils import timezone
+from django_filters import DateFromToRangeFilter, rest_framework as filters
+from django_filters.rest_framework import DjangoFilterBackend
+from rest_framework.permissions import AllowAny
+from rest_framework.views import APIView
+
+from applications.models import Application
+from applications.services.applications_power_bi_csv_report import (
+    ApplicationsPowerBiCsvService,
+)
+from common.authentications import RobotBasicAuthentication
+
+LOGGER = logging.getLogger(__name__)
+
+
+class ApplicationPowerBiFilter(filters.FilterSet):
+    decision_date = DateFromToRangeFilter(
+        field_name="batch__decision_date", label="Batch decision date (range)"
+    )
+
+    class Meta:
+        model = Application
+        fields = ["decision_date"]
+
+    def filter_queryset(self, queryset):
+        """
+        Custom filtering logic that ensures only applications with a batch
+        and a non-null decision_date are returned, and then applies any
+        additional filters such as decision_date range if provided.
+        """
+        queryset = queryset.filter(
+            batch__isnull=False, batch__decision_date__isnull=False
+        )
+        return super().filter_queryset(queryset)
+
+
+class PowerBiIntegrationView(APIView):
+    authentication_classes = [RobotBasicAuthentication]
+    permission_classes = [AllowAny]
+    filter_backends = [DjangoFilterBackend]
+    filterset_class = ApplicationPowerBiFilter
+
+    def get(self, request, *args, **kwargs) -> StreamingHttpResponse:
+        # Apply the filter
+        filterset = ApplicationPowerBiFilter(
+            request.GET, queryset=Application.objects.all()
+        )
+
+        if filterset.is_valid():
+            # Get the filtered queryset from the filter class
+            applications_with_batch_and_decision_date = filterset.qs
+            # Generate the CSV response from the filtered queryset
+            response = self._csv_response(
+                queryset=applications_with_batch_and_decision_date
+            )
+            return response
+        else:
+            # Handle invalid filters (e.g., return a default queryset or handle the error)
+            return StreamingHttpResponse("Invalid filters", status=400)
+
+    def _csv_response(
+        self,
+        queryset: QuerySet[Application],
+        prune_data_for_talpa: bool = False,
+    ) -> StreamingHttpResponse:
+        csv_service = ApplicationsPowerBiCsvService(
+            queryset,
+            prune_data_for_talpa,
+        )
+        response = StreamingHttpResponse(
+            csv_service.get_csv_string_lines_generator(),
+            content_type="text/csv",
+        )
+
+        response["Content-Disposition"] = "attachment; filename={filename}.csv".format(
+            filename=self._filename()
+        )
+        return response
+
+    @staticmethod
+    def _filename():
+        return f"power_bi_data_{timezone.now().strftime('%Y%m%d_%H%M%S')}"

backend/benefit/applications/services/applications_power_bi_csv_report.py

@@ -0,0 +1,27 @@
+from applications.services.applications_csv_report import (  # csv_default_column,
+    ApplicationsCsvService,
+)
+
+# from applications.services.csv_export_base import CsvColumn
+
+
+class ApplicationsPowerBiCsvService(ApplicationsCsvService):
+    """
+    This subclass customizes the CSV_COLUMNS for a different export format.
+    """
+
+    @property
+    def CSV_COLUMNS(self):
+        """
+        Customize the CSV columns but also return the parent class's columns.
+        """
+        # Get the parent class CSV_COLUMNS
+        parent_columns = super().CSV_COLUMNS
+
+        # Define custom columns to add to or modify the parent columns
+        custom_columns = [
+            #    CsvColumn("Custom Column 1", "custom_field_1"),
+            #    CsvColumn("Custom Column 2", "custom_field_2"),
+        ]
+
+        return parent_columns + custom_columns

backend/benefit/applications/tests/conftest.py

@@ -825,6 +825,18 @@ def batch_for_decision_details(application_with_ahjo_decision):
     )
 
 
+@pytest.fixture
+def decided_application_with_decision_date(application_with_ahjo_decision):
+    batch = ApplicationBatch.objects.create(
+        handler=application_with_ahjo_decision.calculation.handler,
+        auto_generated_by_ahjo=True,
+        decision_date=date.today(),
+    )
+    application_with_ahjo_decision.batch = batch
+    application_with_ahjo_decision.save()
+    return application_with_ahjo_decision
+
+
 @pytest.fixture
 def application_alteration_csv_service():
     application_1 = DecidedApplicationFactory(application_number=100003)

backend/benefit/applications/tests/test_power_bi_integration.py

@@ -0,0 +1,32 @@
+import csv
+from io import StringIO
+
+from django.urls import reverse
+
+
+def test_get_power_bi_data(power_bi_client, decided_application_with_decision_date):
+    batch = decided_application_with_decision_date.batch
+    url = (
+        reverse("powerbi_integration_url")
+        + f"?decision_date_after={batch.decision_date}"
+        + f"&decision_date_before={batch.decision_date}"
+    )
+
+    response = power_bi_client.get(url)
+    assert response["Content-Type"] == "text/csv"
+
+    content = "".join([chunk.decode("utf-8") for chunk in response.streaming_content])
+
+    # Parse CSV content
+    csv_reader = csv.reader(StringIO(content), delimiter=";")
+    rows = list(csv_reader)
+
+    # Assert CSV has a header and at least one data row
+    assert len(rows) > 1
+    header = rows[0]
+    assert "Hakemusnumero" in header
+    assert "Työnantajan nimi" in header
+
+    assert rows[1][header.index("Hakemusnumero")] == str(
+        decided_application_with_decision_date.application_number
+    )

backend/benefit/common/tests/conftest.py

@@ -95,6 +95,16 @@ def talpa_client(anonymous_client, settings):
     return anonymous_client
 
 
+@pytest.fixture
+def power_bi_client(anonymous_client, settings):
+    credentials = base64.b64encode(settings.POWER_BI_AUTH_CREDENTIAL.encode("utf-8"))
+
+    anonymous_client.credentials(
+        HTTP_AUTHORIZATION="Basic {}".format(credentials.decode("utf-8"))
+    )
+    return anonymous_client
+
+
 def reseed(number):
     factory.random.reseed_random(str(number))
     random.seed(number)

backend/benefit/helsinkibenefit/settings.py

@@ -139,6 +139,7 @@
     TERMS_OF_SERVICE_SESSION_KEY=(str, "_tos_session"),
     ENABLE_DEBUG_ENV=(bool, False),
     TALPA_ROBOT_AUTH_CREDENTIAL=(str, "username:password"),
+    POWER_BI_AUTH_CREDENTIAL=(str, "username:password"),
     DISABLE_TOS_APPROVAL_CHECK=(bool, False),
     YRTTI_BASE_URL=(
         str,
@@ -506,6 +507,7 @@
 WKHTMLTOPDF_BIN = env("WKHTMLTOPDF_BIN")
 
 TALPA_ROBOT_AUTH_CREDENTIAL = env("TALPA_ROBOT_AUTH_CREDENTIAL")
+POWER_BI_AUTH_CREDENTIAL = env("POWER_BI_AUTH_CREDENTIAL")
 
 YRTTI_TIMEOUT = env("YRTTI_TIMEOUT")
 YRTTI_BASE_URL = env("YRTTI_BASE_URL")

backend/benefit/helsinkibenefit/urls.py

@@ -25,6 +25,7 @@
     AhjoDecisionCallbackView,
 )
 from applications.api.v1.ahjo_setting_views import AhjoSettingDetailView
+from applications.api.v1.power_bi_integration_views import PowerBiIntegrationView
 from applications.api.v1.review_state_views import ReviewStateView
 from applications.api.v1.search_views import SearchView
 from applications.api.v1.talpa_integration_views import TalpaCallbackView
@@ -112,6 +113,11 @@
         TalpaCallbackView.as_view(),
         name="talpa_callback_url",
     ),
+    path(
+        "v1/powerbi-integration/",
+        PowerBiIntegrationView.as_view(),
+        name="powerbi_integration_url",
+    ),
     path(
         "v1/decision-proposal-sections/",
         DecisionProposalTemplateSectionList.as_view(),