From 45bd3c4d04d31f3a261f0a93a2fb8eac218e7c91 Mon Sep 17 00:00:00 2001 From: Ernesto Ruge Date: Tue, 7 Jul 2026 12:38:02 +0200 Subject: [PATCH] improve realtime push --- .../datex2/datex2_import_api_test.py | 44 +++++++++++++++++++ webapp/common/constants.py | 4 ++ .../server_rest_api/datex2/datex2_rest_api.py | 13 +++--- .../v3_5/base_datex2_v3_5_import_service.py | 37 ++++++++++++++-- 4 files changed, 90 insertions(+), 8 deletions(-) diff --git a/tests/integration/server_rest_api/datex2/datex2_import_api_test.py b/tests/integration/server_rest_api/datex2/datex2_import_api_test.py index 3b181b8d..78db676d 100644 --- a/tests/integration/server_rest_api/datex2/datex2_import_api_test.py +++ b/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, diff --git a/webapp/common/constants.py b/webapp/common/constants.py index 4c704742..84ec54f3 100644 --- a/webapp/common/constants.py +++ b/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 diff --git a/webapp/server_rest_api/datex2/datex2_rest_api.py b/webapp/server_rest_api/datex2/datex2_rest_api.py index 9becdafe..4abbe27d 100644 --- a/webapp/server_rest_api/datex2/datex2_rest_api.py +++ b/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 diff --git a/webapp/services/import_services/datex2/v3_5/base_datex2_v3_5_import_service.py b/webapp/services/import_services/datex2/v3_5/base_datex2_v3_5_import_service.py index 771bed9f..85bb136f 100644 --- a/webapp/services/import_services/datex2/v3_5/base_datex2_v3_5_import_service.py +++ b/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]