improved realtime update handling

tests/integration/server_rest_api/datex2/datex2_import_api_test.py

@@ -177,6 +177,55 @@ def test_push_realtime_v35_async_applies_status_updates(
         # The celery task deletes its source file once it has been processed.
         assert list(isolated_datex2_dir.iterdir()) == []
 
+    @staticmethod
+    def test_static_reimport_after_realtime_push_keeps_status(
+        db: SQLAlchemy,
+        test_client: OpenApiFlaskClient,
+        requests_mock: Mocker,
+        isolated_datex2_dir: Path,
+        eager_celery_helper,
+    ) -> None:
+        """
+        Regression test: a static import that happens *after* realtime updates must not reset
+        the EVSE status back to UNKNOWN.
+
+        Sequence:
+          1. Import static data (all EVSEs start as UNKNOWN).
+          2. Push realtime data via the push endpoint (statuses become AVAILABLE/CHARGING/...).
+          3. Import static data *again*.
+          4. The realtime statuses must still be there - not reset to UNKNOWN.
+        """
+        # 1. Static import - all EVSEs start as UNKNOWN.
+        _import_static_data(requests_mock)
+        assert db.session.query(Evse).filter(Evse.status == EvseStatus.UNKNOWN).count() == 57
+
+        # 2. Push realtime data; the eager celery fixture applies it inline.
+        realtime_data = _load_test_data('datex2_enbw_realtime_reduced.json')
+        response = test_client.post(
+            path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}',
+            json=realtime_data,
+        )
+        assert response.status_code == HTTPStatus.OK
+
+        db.session.expire_all()
+        assert db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*2').first().status == EvseStatus.AVAILABLE
+        assert db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first().status == EvseStatus.CHARGING
+
+        # 3. Static import again - this is what used to wipe the realtime status.
+        _import_static_data(requests_mock)
+
+        # 4. The realtime statuses must survive the static re-import.
+        db.session.expire_all()
+        evse_available = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*2').first()
+        assert evse_available.status == EvseStatus.AVAILABLE, 'static re-import reset an AVAILABLE EVSE to UNKNOWN'
+
+        evse_charging = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*1').first()
+        assert evse_charging.status == EvseStatus.CHARGING, 'static re-import reset a CHARGING EVSE to UNKNOWN'
+
+        # EVSEs that never received a realtime update legitimately stay UNKNOWN.
+        evse_unchanged = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E916701*2').first()
+        assert evse_unchanged.status == EvseStatus.UNKNOWN
+
     @staticmethod
     def test_push_realtime_v35_gzip_encoded(
         db: SQLAlchemy,
@@ -208,6 +257,65 @@ def test_push_realtime_v35_gzip_encoded(
         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_delta_push_applies_directly(
+        db: SQLAlchemy,
+        test_client: OpenApiFlaskClient,
+        requests_mock: Mocker,
+        isolated_datex2_dir: Path,
+        stubbed_celery_delay,
+    ) -> None:
+        """
+        deltaPush payloads are small incremental updates and must be applied synchronously on the
+        request thread, without persisting a file or queueing a celery task.
+        """
+        _import_static_data(requests_mock)
+
+        realtime_data = _load_test_data('datex2_enbw_realtime_reduced.json')
+        realtime_data['messageContainer']['exchangeInformation']['exchangeContext']['codedExchangeProtocol'] = {
+            'value': 'deltaPush',
+        }
+
+        response = test_client.post(
+            path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}',
+            json=realtime_data,
+        )
+        assert response.status_code == HTTPStatus.OK
+
+        # Stored directly: nothing persisted to disk and no celery task queued.
+        assert list(isolated_datex2_dir.iterdir()) == []
+        stubbed_celery_delay.assert_not_called()
+
+        # EVSE statuses are updated synchronously within the request.
+        db.session.expire_all()
+        evse_available = db.session.query(Evse).filter(Evse.uid == 'DE*EBW*E914082*2').first()
+        assert evse_available.status == EvseStatus.AVAILABLE
+        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_invalid_payload_returns_400(
+        db: SQLAlchemy,
+        test_client: OpenApiFlaskClient,
+        requests_mock: Mocker,
+        isolated_datex2_dir: Path,
+        stubbed_celery_delay,
+    ) -> None:
+        """
+        A structurally invalid payload must be rejected synchronously with HTTP 400 - before any
+        file is persisted or task queued - regardless of the (async) exchange protocol.
+        """
+        _import_static_data(requests_mock)
+
+        response = test_client.post(
+            path=f'/api/server/v1/datex/v3.5/{SOURCE_UID}/realtime?key={API_KEY}',
+            json={},
+        )
+
+        assert response.status_code == HTTPStatus.BAD_REQUEST
+        assert list(isolated_datex2_dir.iterdir()) == []
+        stubbed_celery_delay.assert_not_called()
+
     @staticmethod
     def test_push_realtime_v35_empty_body_returns_400(
         db: SQLAlchemy,

tests/integration/services/import_services/bnetza/bnetza_excel_service_test.py

@@ -109,7 +109,7 @@ def test_bnetza_excel_import(db: SQLAlchemy, requests_mock: Mocker) -> None:
         'presence': PresenceStatus.PRESENT,
         'physical_reference': None,
         'last_updated': ANY,
-        'status_last_updated': None,
+        'status_last_updated': ANY,
         'max_reservation': None,
         'parking_restrictions': [],
         'terms_and_conditions': None,

webapp/server_rest_api/datex2/datex2_handler.py

@@ -38,13 +38,16 @@ def __init__(self, *, import_services: ImportServices, **kwargs):
         super().__init__(**kwargs)
         self.import_services = import_services
 
-    def validate_realtime_request(self, source_uid: str, key: str | None) -> None:
+    def validate_realtime_request(self, source_uid: str, key: str | None) -> BaseDatex2V35ImportService:
         """
         Synchronous pre-flight validation for an incoming realtime push: makes sure the source
         exists, is a DATEX II v3.5 source, and the supplied ``key`` query parameter matches the
         source's configured ``api_key``. Raises ``NotFoundException`` / ``UnauthorizedException``.
+
+        Returns the resolved import service so the caller can validate and apply the payload.
         """
-        self._authenticate(source_uid, key)
+        _, service = self._authenticate(source_uid, key)
+        return service
 
     def get_last_modified(self, source_uid: str, key: str | None) -> datetime | None:
         source, _ = self._authenticate(source_uid, key)

webapp/server_rest_api/datex2/datex2_rest_api.py

@@ -16,6 +16,7 @@
 along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 
+import json
 from email.utils import format_datetime
 from http import HTTPStatus
 from pathlib import Path
@@ -41,6 +42,7 @@
 from webapp.common.server_auth import skip_basic_auth
 from webapp.dependencies import dependencies
 from webapp.server_rest_api.base_blueprint import ServerApiBaseBlueprint
+from webapp.services.import_services.exceptions import ImportException
 from webapp.services.import_services.import_celery import datex2_v3_5_realtime_import_by_file
 
 from .datex2_handler import Datex2Handler
@@ -101,10 +103,11 @@ def head(self, source_uid: str) -> tuple[Response, HTTPStatus]:
         summary='Push DATEX II v3.5 realtime EVSE statuses',
         description=(
             'Accepts a Mobilithek-style DATEX II v3.5 JSON payload (`messageContainer.payload[0]` '
-            '`aegiEnergyInfrastructureStatusPublication`). The request body is 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`."
+            '`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 '
+            '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`."
         ),
         path=[Parameter('source_uid', schema=StringField())],
         query=[
@@ -122,8 +125,8 @@ def head(self, source_uid: str) -> tuple[Response, HTTPStatus]:
     )
     def post(self, source_uid: str) -> tuple[Response, HTTPStatus]:
         # Validate the source + API key synchronously so unauthorized/unknown sources fail fast
-        # before we persist the payload to disk.
-        self.datex2_handler.validate_realtime_request(
+        # before we touch the payload.
+        service = self.datex2_handler.validate_realtime_request(
             source_uid=source_uid,
             key=self.request_helper.get_query_args().get('key'),
         )
@@ -132,6 +135,24 @@ def post(self, source_uid: str) -> tuple[Response, HTTPStatus]:
         if not data:
             raise InputValidationException(message='no realtime payload')
 
+        try:
+            parsed_data = json.loads(data)
+        except json.JSONDecodeError as e:
+            raise InputValidationException(message='invalid JSON realtime payload') from e
+
+        # Validate the envelope up-front via the import service so malformed pushes fail fast.
+        try:
+            message_container = service.validate_realtime_data(parsed_data)
+        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):
+            service.store_realtime_data(message_container)
+            # Mobilithek expects HTTP 200 instead of HTTP 204
+            return empty_json_response(), HTTPStatus.OK
+
         base_path = Path(self.config_helper.get('DATEX2_IMPORT_DIR'))
         if not base_path.is_dir():
             base_path.mkdir(parents=True, exist_ok=True)

webapp/services/import_services/base_import_service.py

@@ -239,7 +239,7 @@ def save_location_update(
         self.set_image_list(location, location_update.images, images_by_url)
 
         old_charging_stations_by_uid = {
-            charging_station.id: charging_station for charging_station in location.charging_pool
+            charging_station.uid: charging_station for charging_station in location.charging_pool
         }
         new_charging_stations: list[ChargingStation] = []
         for charging_station_update in location_update.charging_pool:
@@ -354,10 +354,19 @@ def get_evse(
         evse.last_updated = evse_update.last_updated or datetime.now(tz=timezone.utc)
 
         for key, value in evse_update.to_dict().items():
-            if key == 'last_updated':
+            if key in ['last_updated', 'status', 'status_last_updated', 'presence']:
                 continue
             setattr(evse, key, value)
 
+        # presence is not-nullable; only overwrite it when the update actually carries a value,
+        # otherwise a static import (which never sets presence) would clear an existing presence.
+        if evse_update.presence is not None:
+            evse.presence = evse_update.presence
+
+        if evse_update.status is not None and evse_update.status != evse.status:
+            evse.status_last_updated = evse_update.status_last_updated or datetime.now(tz=timezone.utc)
+            evse.status = evse_update.status
+
         old_connectors_by_uid = {connector.uid: connector for connector in evse.connectors}
         new_connectors: list[Connector] = []
         for connector_update in evse_update.connectors:

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

@@ -35,7 +35,7 @@
 from webapp.services.import_services.base_import_service import BaseImportService
 from webapp.services.import_services.exceptions import ImportException
 from webapp.services.import_services.models import EvseRealtimeUpdate, LocationUpdate
-from webapp.shared.datex2.models import MessageContainerWrapperInput
+from webapp.shared.datex2.models import MessageContainerWrapperInput, ProtocolTypeEnum
 from webapp.shared.datex2.v3_5_json_realtime.models.energy_infrastructure_station_status_input import (
     EnergyInfrastructureStationStatusInput,
 )
@@ -169,7 +169,8 @@ def fetch_realtime_data(self):
                 return
 
             try:
-                message_container = self.add_realtime_data(data, result)
+                message_container = self.validate_realtime_data(data)
+                self.add_realtime_data(message_container, result)
             except ImportException as e:
                 logger.error(
                     e.message,
@@ -218,16 +219,14 @@ def import_realtime_data_from_file(self, import_file_path: Path) -> None:
 
     def import_realtime_data(self, data: dict) -> None:
         """
-        Apply a parsed DATEX II v3.5 realtime payload to this source: parse the
-        ``messageContainer`` envelope, upsert EVSE statuses, and update the source's
+        Apply a parsed DATEX II v3.5 realtime payload to this source: validate the
+        ``messageContainer`` envelope, then upsert EVSE statuses and update the source's
         ``realtime_data_updated_at`` / ``realtime_status`` accordingly.
         """
         source = self.get_source()
-        last_modified = datetime.now(tz=timezone.utc)
-        result = RealtimeResult()
 
         try:
-            message_container = self.add_realtime_data(data, result)
+            message_container = self.validate_realtime_data(data)
         except ImportException as e:
             logger.error(
                 e.message,
@@ -236,6 +235,22 @@ def import_realtime_data(self, data: dict) -> None:
             self.update_source(source, realtime_status=SourceStatus.FAILED)
             return
 
+        self.store_realtime_data(message_container)
+
+    def store_realtime_data(self, message_container: MessageContainerWrapperInput) -> None:
+        """
+        Apply an already-validated DATEX II v3.5 realtime message container to this source: upsert
+        EVSE statuses and update the source's ``realtime_data_updated_at`` / ``realtime_status``.
+
+        Use this when the envelope has already been validated via :meth:`validate_realtime_data`
+        (e.g. a synchronous ``deltaPush`` from the REST API); otherwise use :meth:`import_realtime_data`.
+        """
+        source = self.get_source()
+        last_modified = datetime.now(tz=timezone.utc)
+        result = RealtimeResult()
+
+        self.add_realtime_data(message_container, result)
+
         self.save_evse_updates(list(result.evse_updates_by_evse.values()))
 
         if message_container.messageContainer.payload:
@@ -249,13 +264,22 @@ def import_realtime_data(self, data: dict) -> None:
             realtime_error_count=result.realtime_error_count,
             realtime_data_updated_at=last_modified,
         )
+        exchange_context = message_container.messageContainer.exchangeInformation.exchangeContext
+        protocol_type = exchange_context.codedExchangeProtocol.value
+        protocol_label = f'{protocol_type.value} ' if protocol_type else ''
         logger.info(
-            f'Imported DATEX2 realtime data for {self.source_info.uid} via async push with '
+            f'Imported DATEX2 realtime {protocol_label}data for {self.source_info.uid} via push with '
             f'{result.realtime_success_count} valid EVSEs and {result.realtime_error_count} failed EVSEs.',
             extra={'attributes': {'type': LogMessageType.IMPORT_LOCATION}},
         )
 
-    def add_realtime_data(self, data: dict, result: RealtimeResult) -> MessageContainerWrapperInput:
+    def validate_realtime_data(self, data: dict) -> MessageContainerWrapperInput:
+        """
+        Validate the raw DATEX II v3.5 realtime envelope and return the parsed message container.
+
+        Raises :class:`ImportException` if the payload cannot be parsed or is missing the
+        ``messageContainer`` payload / ``aegiEnergyInfrastructureStatusPublication``.
+        """
         try:
             datex_input: MessageContainerWrapperInput = self.message_container_validator.validate(data)
         except ValidationError as e:
@@ -281,6 +305,18 @@ def add_realtime_data(self, data: dict, result: RealtimeResult) -> MessageContai
                 message=f'missing aegiEnergyInfrastructureStatusPublication in {self.source_info.uid} realtime data',
             )
 
+        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 (
+            message_container.messageContainer.exchangeInformation.exchangeContext.codedExchangeProtocol.value
+            is ProtocolTypeEnum.DELTA_PUSH
+        )
+
+    def add_realtime_data(self, message_container: MessageContainerWrapperInput, result: RealtimeResult) -> None:
+        payload = message_container.messageContainer.payload[0]
         for site_status in payload.aegiEnergyInfrastructureStatusPublication.energyInfrastructureSiteStatus:
             if site_status.energyInfrastructureStationStatus is UnsetValue:
                 continue
@@ -297,8 +333,6 @@ def add_realtime_data(self, data: dict, result: RealtimeResult) -> MessageContai
                     result.evse_updates_by_evse[evse_update.evse_id] = evse_update
                     result.realtime_success_count += 1
 
-        return datex_input
-
     def request_data(
         self,
         subscription_id: int,