improve realtime push

tests/integration/server_rest_api/datex2/datex2_import_api_test.py

@@ -293,6 +293,50 @@ def test_push_realtime_v35_delta_push_applies_directly(
         evse_charging = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first()
         assert evse_charging.status == EvseStatus.CHARGING
 
+    @staticmethod
+    def test_push_realtime_v35_large_delta_push_is_async(
+        db: SQLAlchemy,
+        test_client: OpenApiFlaskClient,
+        requests_mock: Mocker,
+        isolated_datex2_dir: Path,
+        stubbed_celery_delay,
+    ) -> None:
+        """
+        A deltaPush payload carrying more than DATEX2_ASYNC_STATION_STATUS_THRESHOLD station statuses
+        must be handed off to celery (persisted + queued) instead of applied inline, even though it
+        is a deltaPush.
+        """
+        _import_static_data(requests_mock)
+
+        realtime_data = _load_test_data('datex2_enbw_realtime_reduced.json')
+        realtime_data['messageContainer']['exchangeInformation']['exchangeContext']['codedExchangeProtocol'] = {
+            'value': 'deltaPush',
+        }
+
+        # The reduced payload has 5 station statuses - lower the threshold so it counts as "large".
+        config = dependencies.get_config_helper().get_config()
+        original = config.get('DATEX2_ASYNC_STATION_STATUS_THRESHOLD')
+        config['DATEX2_ASYNC_STATION_STATUS_THRESHOLD'] = 2
+        try:
+            response = test_client.post(
+                path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}',
+                json=realtime_data,
+            )
+        finally:
+            config['DATEX2_ASYNC_STATION_STATUS_THRESHOLD'] = original
+
+        assert response.status_code == HTTPStatus.OK
+
+        # Handed off to celery: payload persisted to disk and task queued.
+        written = list(isolated_datex2_dir.iterdir())
+        assert len(written) == 1
+        stubbed_celery_delay.assert_called_once()
+
+        # Because the celery task did not run, EVSE statuses stay untouched.
+        db.session.expire_all()
+        evse_charging = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first()
+        assert evse_charging.status == EvseStatus.UNKNOWN
+
     @staticmethod
     def test_push_realtime_v35_invalid_payload_returns_400(
         db: SQLAlchemy,

webapp/common/constants.py

@@ -38,6 +38,10 @@ class BaseConfig:
     BNETZA_IMPORT_DIR = os.path.abspath(os.path.join(PROJECT_ROOT, os.pardir, 'data', 'bnetza-import'))
     DATEX2_IMPORT_DIR = os.path.abspath(os.path.join(PROJECT_ROOT, os.pardir, 'data', 'datex2-import'))
 
+    # DATEX II realtime pushes with more than this many energyInfrastructureStationStatus entries are
+    # processed asynchronously via celery instead of inline on the request thread (snapshots are always async).
+    DATEX2_ASYNC_STATION_STATUS_THRESHOLD = 25
+
     PUBLIC_IMAGE_PATH = '/data/images'
 
     SQLALCHEMY_TRACK_MODIFICATIONS = False

webapp/server_rest_api/datex2/datex2_rest_api.py

@@ -104,8 +104,9 @@ def head(self, source_uid: str) -> tuple[Response, HTTPStatus]:
         description=(
             'Accepts a Mobilithek-style DATEX II v3.5 JSON payload (`messageContainer.payload[0]` '
             '`aegiEnergyInfrastructureStatusPublication`). The payload is validated synchronously; '
-            '`deltaPush` updates are small incremental changes and are applied directly, while any '
-            'other exchange protocol (e.g. snapshots) is persisted to `DATEX2_IMPORT_DIR` and '
+            'small `deltaPush` updates are applied directly, while snapshots (any other exchange '
+            'protocol) and `deltaPush` payloads with more than `DATEX2_ASYNC_STATION_STATUS_THRESHOLD` '
+            '`energyInfrastructureStationStatus` entries are persisted to `DATEX2_IMPORT_DIR` and '
             'processed asynchronously by a Celery worker so very large payloads do not block the '
             "request thread. The `key` query parameter must match the source's configured `api_key`."
         ),
@@ -146,9 +147,11 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]:
         except ImportException as e:
             raise InputValidationException(message=e.message) from e
 
-        # deltaPush updates are small incremental changes - apply them directly. Everything else
-        # (snapshots etc.) can be large, so hand it off to a celery worker via the import file.
-        if service.is_delta_push(message_container):
+        # Small deltaPush updates are small incremental changes - apply them directly. Everything
+        # else can be large, so hand it off to a celery worker via the import file: snapshots as well
+        # as deltaPush payloads with more than DATEX2_ASYNC_STATION_STATUS_THRESHOLD station statuses.
+        async_station_status_threshold = self.config_helper.get('DATEX2_ASYNC_STATION_STATUS_THRESHOLD', 25)
+        if not service.should_process_asynchronously(message_container, async_station_status_threshold):
             service.store_realtime_data(message_container)
             # Mobilithek expects HTTP 200 instead of HTTP 204
             return empty_json_response(), HTTPStatus.OK

webapp/services/import_services/datex2/v3_5/base_datex2_v3_5_import_service.py

@@ -308,12 +308,43 @@ def validate_realtime_data(self, data: dict) -> MessageContainerWrapperInput:
         return datex_input
 
     @staticmethod
-    def is_delta_push(message_container: MessageContainerWrapperInput) -> bool:
-        """Return whether the realtime payload uses the DATEX II ``deltaPush`` exchange protocol."""
-        return (
+    def count_station_statuses(message_container: MessageContainerWrapperInput) -> int:
+        """Return the number of ``energyInfrastructureStationStatus`` entries carried by the payload."""
+        payload = message_container.messageContainer.payload[0]
+        publication = payload.aegiEnergyInfrastructureStatusPublication
+        if publication is UnsetValue or publication.energyInfrastructureSiteStatus is UnsetValue:
+            return 0
+
+        count = 0
+        for site_status in publication.energyInfrastructureSiteStatus:
+            if site_status.energyInfrastructureStationStatus is UnsetValue:
+                continue
+            count += len(site_status.energyInfrastructureStationStatus)
+        return count
+
+    @classmethod
+    def should_process_asynchronously(
+        cls,
+        message_container: MessageContainerWrapperInput,
+        station_status_threshold: int,
+    ) -> bool:
+        """
+        Decide whether the realtime payload should be handed off to a celery worker instead of being
+        applied inline on the request thread.
+
+        Only small ``deltaPush`` payloads are applied synchronously. Everything else is processed
+        asynchronously: snapshots (any non-``deltaPush`` exchange protocol) as well as ``deltaPush``
+        payloads carrying more than ``station_status_threshold`` ``energyInfrastructureStationStatus``
+        entries, so large payloads never block the request thread.
+        """
+        is_delta_push = (
             message_container.messageContainer.exchangeInformation.exchangeContext.codedExchangeProtocol.value
             is ProtocolTypeEnum.DELTA_PUSH
         )
+        if not is_delta_push:
+            return True
+
+        return cls.count_station_statuses(message_container) > station_status_threshold
 
     def add_realtime_data(self, message_container: MessageContainerWrapperInput, result: RealtimeResult) -> None:
         payload = message_container.messageContainer.payload[0]