fix(S2): the remaining findings (2 of 2)

CHANGELOG.rst

@@ -21,19 +21,20 @@ Added
 
 * ``server_version_at_least``, which answers whether the server is of a given version or newer, looking its version up where that is not known yet, for a caller that sends a field or calls an endpoint only where the server has it. Release generations are compared, so a development build such as ``0.33.0.dev26`` counts as ``0.33.0`` [`#248 <https://github.com/FlexMeasures/flexmeasures-client/pull/248>`__]
 * ``ingestion_polling_interval``, ``ingestion_polling_max_interval`` and ``ingestion_polling_timeout``, which govern how the ingestion of posted sensor data is waited for, on a tighter profile than the ``job_polling_*`` settings give other background jobs [`#242 <https://github.com/FlexMeasures/flexmeasures-client/pull/242>`__]
+* ``post_sensor_data`` takes ``confirm_ingestion``. With ``True``, it raises an error if it cannot confirm that FlexMeasures ingested the posted data, instead of logging this and carrying on. Use it if your code remembers what it has recorded, and would otherwise never record the data again [`#273 <https://github.com/FlexMeasures/flexmeasures-client/pull/273>`__]
 * Optional ``rate_limit_notifier`` and ``job_status_notifier`` callbacks, to report rate limits and the progress of jobs being waited for, instead of logging them [`#225 <https://github.com/FlexMeasures/flexmeasures-client/pull/225>`__]
 * ``get_automations``, ``get_automation``, ``add_automation``, ``update_automation``, ``delete_automation``, ``trigger_automation`` and ``trigger_and_await_automation``, to manage and run automations through the API on FlexMeasures v1.1.0 and up [`#230 <https://github.com/FlexMeasures/flexmeasures-client/pull/230>`__]
 
 Changed
 -------
 
-* ``add_asset`` no longer requires a ``latitude`` and ``longitude``, which FlexMeasures does not require either, for an asset with no location, such as one standing for a community. ``generic_asset_type_id`` stays required, and half a location is refused [`#261 <https://github.com/FlexMeasures/flexmeasures-client/pull/261>`__]
+* ``add_asset`` no longer requires a ``latitude`` and ``longitude``, just like FlexMeasures. This allows assets without a location, such as one that represents a community. ``generic_asset_type_id`` is still required, and a latitude without a longitude (or the other way around) is refused [`#261 <https://github.com/FlexMeasures/flexmeasures-client/pull/261>`__]
 * ``post_sensor_data``, and the deprecated ``post_measurements`` it was renamed from, wait for a server of FlexMeasures 0.33.0 or above that ingests the posted data asynchronously, answering ``202 Accepted`` with a job id, to finish doing so, so that a caller can read the data back or schedule over it; a failed job raises ``JobFailedError``, while a job still pending at the timeout, or one that cannot be looked up, is logged as an error and leaves the post successful. Pass ``await_ingestion=False`` to return as soon as the data is accepted [`#242 <https://github.com/FlexMeasures/flexmeasures-client/pull/242>`__]
 
 Removed
 -------
 
-* ``response_handling.check_content_type``, which refused an answer in a JSON variant that aiohttp parses, such as ``application/problem+json``, with an error that raised an ``AttributeError`` of its own when printed. Such an answer is now read like any other. A content type that is not JSON at all is still reported as a ``ConnectionError``, by aiohttp's own check [`#257 <https://github.com/FlexMeasures/flexmeasures-client/pull/257>`__]
+* ``response_handling.check_content_type``. It refused responses in JSON variants that aiohttp can parse, such as ``application/problem+json``, and the error it raised caused an ``AttributeError`` of its own when printed. Such responses are now read like any other. A response that is not JSON at all is still reported as a ``ConnectionError``, by aiohttp's own check [`#257 <https://github.com/FlexMeasures/flexmeasures-client/pull/257>`__]
 
 Fixed
 -----

src/flexmeasures_client/client.py

@@ -459,11 +459,11 @@ def report_rate_limit(message: str) -> None:
                 "Request retry timeout while connecting to the API."
             ) from exception
 
-        # request_once has already parsed this answer with aiohttp's json(),
-        # which refuses a content type that is not JSON, and the loop above reports that as a ConnectionError.
-        # A check that used to follow here refused the JSON variants aiohttp accepts too, such as application/problem+json,
-        # with an error that could not even be printed, so it is gone.
-        # A body that is not JSON under a JSON content type is refused by json() itself, as a JSONDecodeError.
+        # request_once has already parsed this response with aiohttp's json().
+        # json() refuses a content type that is not JSON, and the loop above reports that as a ConnectionError.
+        # A check used to follow here, but it also refused JSON variants that aiohttp accepts, such as application/problem+json.
+        # Its error could not even be printed, so the check was removed.
+        # A body that is not JSON under a JSON content type makes json() itself raise a JSONDecodeError.
         return await response.json(), response.status
 
     async def request_once(
@@ -664,6 +664,7 @@ async def post_sensor_data(
         file_path: str | None = None,
         belief_time_measured_instantly: bool = False,
         await_ingestion: bool = True,
+        confirm_ingestion: bool = False,
     ):
         """
         Post sensor data for the given time range.
@@ -676,19 +677,24 @@ async def post_sensor_data(
 
         The method automatically chooses the appropriate API endpoint based on the provided parameters.
 
-        Since FlexMeasures 0.33.0, a server with an ingestion worker available answers ``202 Accepted`` with the id of a background job,
-        rather than ``200 OK``, and ingests the data after the request has returned.
-        By default this method then waits for that job to finish,
-        so that a caller which reads the data back, or asks for a schedule over it, sees what it just posted.
+        Since FlexMeasures 0.33.0, a server with an available ingestion worker responds ``202 Accepted`` instead of ``200 OK``,
+        with the id of a background job.
+        The server then ingests the data after the request has returned.
+        By default, this method waits for that job to finish.
+        This way, a caller that reads the data back, or asks for a schedule over it, sees the data it just posted.
 
         :param await_ingestion: whether to wait for asynchronous ingestion to finish,
-            when the server answered ``202 Accepted`` with a job id.
-            Waiting is polled on the ``ingestion_polling_*`` profile, which is tighter than the one other jobs are polled on.
+            when the server responded ``202 Accepted`` with a job id.
+            The job is polled on the ``ingestion_polling_*`` profile, which is tighter than the profile for other jobs.
             A job that fails raises :class:`JobFailedError`.
-            A job still pending when ``self.ingestion_polling_timeout`` is reached is logged as an error and otherwise ignored,
-            since the data may well land moments later, and a caller that has waited that long is better served by carrying on than by failing.
+            A job that is still pending when ``self.ingestion_polling_timeout`` is reached is logged as an error and otherwise ignored.
+            The data may well land moments later, and a caller that has waited that long is better off carrying on than failing.
             Pass ``False`` to return as soon as the server has accepted the data.
-            A synchronous ``200 OK`` response is unaffected either way.
+            Either way, a synchronous ``200 OK`` response is unaffected.
+        :param confirm_ingestion: whether to raise when ingestion could not be confirmed while waiting for it, instead of logging and ignoring that.
+            If the job was still pending at the timeout, this raises :class:`JobTimeoutError`.
+            If looking the job up failed, this raises the error from the lookup.
+            Use this if the caller remembers having recorded the data, and would otherwise never record it again.
 
         This function raises a ValueError when an unhandled status code is returned.
         """
@@ -740,6 +746,7 @@ async def post_sensor_data(
                 unit=unit,
                 prior=prior,
                 await_ingestion=await_ingestion,
+                confirm_ingestion=confirm_ingestion,
             )
         else:
             # Type assertion to help the type checker understand that file_path is not None
@@ -752,6 +759,7 @@ async def post_sensor_data(
                 belief_time_measured_instantly=belief_time_measured_instantly,
                 unit=unit,
                 await_ingestion=await_ingestion,
+                confirm_ingestion=confirm_ingestion,
             )
 
     @staticmethod
@@ -767,20 +775,26 @@ def _ingestion_job_id(response) -> str | None:
         job_id = response.get("job") or response.get("job_id")
         return job_id if isinstance(job_id, str) else None
 
-    async def _await_sensor_data_ingestion(self, job_id: str) -> None:
+    async def _await_sensor_data_ingestion(
+        self, job_id: str, confirm_ingestion: bool = False
+    ) -> None:
         """Wait for an ingestion job to finish, so that the data it carries can be read back.
 
-        Waiting is left to :func:`wait_for_job`, so ingestion is backed off like every other job this client waits for,
-        and reported through the same ``job_status_notifier``, on the ``ingestion_polling_*`` intervals rather than the ``job_polling_*`` ones.
-
-        A failed job propagates as :class:`JobFailedError`, for the caller to handle like any other failed post.
-        Nothing else does.
-        A timeout is logged as an error and otherwise ignored, for the reason given in :func:`post_sensor_data`,
-        and so is a failure to look the job up at all, which says nothing about the data:
-        the server accepted that, and a job record can be gone, or answered for by something in front of the server, by the time it is asked about.
-        Raising there would report a post that succeeded as a post that failed.
-        A lookup that 404s is reported as a ``ConnectionError`` by :func:`request`, which wraps every client error that way,
-        so that is among the ones caught here, alongside the ``ValueError`` an unexpected status code raises.
+        Waiting is delegated to :func:`wait_for_job`.
+        So ingestion is backed off like every other job this client waits for,
+        and reported through the same ``job_status_notifier``.
+        The only difference is that it uses the ``ingestion_polling_*`` intervals, instead of the ``job_polling_*`` ones.
+
+        A failed job raises :class:`JobFailedError`, which the caller handles like any other failed post.
+        By default, no other error is raised.
+        A timeout is logged as an error and otherwise ignored, for the reason given in :func:`post_sensor_data`.
+        The same goes for a failure to look the job up at all, which says nothing about the data.
+        After all, the server did accept the data.
+        By the time the job is looked up, its record may be gone, or something in front of the server may respond instead.
+        Raising in these cases would report a successful post as a failed post.
+        If the caller asked for ingestion to be confirmed, both cases are logged and then raised.
+        A lookup that returns 404 is reported as a ``ConnectionError`` by :func:`request`, which wraps every client error that way.
+        So ``ConnectionError`` is among the errors caught here, alongside the ``ValueError`` that an unexpected status code raises.
         """
         try:
             await self.wait_for_job(
@@ -793,6 +807,8 @@ async def _await_sensor_data_ingestion(self, job_id: str) -> None:
             self.logger.error(
                 f"Ingestion of the data posted as job {job_id} is unconfirmed, so a read of it may come back without it. {e}"
             )
+            if confirm_ingestion:
+                raise
         except (
             ConnectionError,
             TimeoutError,
@@ -803,6 +819,8 @@ async def _await_sensor_data_ingestion(self, job_id: str) -> None:
             self.logger.error(
                 f"Ingestion of the data posted as job {job_id} is unconfirmed, because looking that job up failed, so a read of it may come back without it. {e}"
             )
+            if confirm_ingestion:
+                raise
 
     async def _post_sensor_data_json(
         self,
@@ -813,6 +831,7 @@ async def _post_sensor_data_json(
         unit: str,
         prior: str | datetime | None = None,
         await_ingestion: bool = True,
+        confirm_ingestion: bool = False,
     ):
         """
         Post sensor data using JSON payload.
@@ -849,7 +868,7 @@ async def _post_sensor_data_json(
             self.logger.info("Sensor data sent successfully via JSON.")
 
         if await_ingestion and status == 202 and job_id:
-            await self._await_sensor_data_ingestion(job_id)
+            await self._await_sensor_data_ingestion(job_id, confirm_ingestion)
 
         return response, status
 
@@ -860,6 +879,7 @@ async def _post_sensor_data_file(
         belief_time_measured_instantly: bool = False,
         unit: str | None = None,
         await_ingestion: bool = True,
+        confirm_ingestion: bool = False,
     ):
         """
         Post sensor data using file upload.
@@ -981,7 +1001,7 @@ async def _post_sensor_data_file(
         # Wait for ingestion outside the request timeout above, and outside the error handler that names this an upload failure:
         # a job takes as long as self.ingestion_polling_timeout allows, which is no concern of the request that queued it.
         if await_ingestion and status == 202 and upload_job_id:
-            await self._await_sensor_data_ingestion(upload_job_id)
+            await self._await_sensor_data_ingestion(upload_job_id, confirm_ingestion)
 
         return response_data, status
 
@@ -995,6 +1015,7 @@ async def post_measurements(
         unit: str,
         prior: str | datetime | None = None,
         await_ingestion: bool = True,
+        confirm_ingestion: bool = False,
     ):
         """
         Post sensor data for the given time range.
@@ -1019,6 +1040,7 @@ async def post_measurements(
             unit=unit,
             prior=prior,
             await_ingestion=await_ingestion,
+            confirm_ingestion=confirm_ingestion,
         )
 
     async def get_schedule(
@@ -1538,10 +1560,10 @@ async def add_asset(
                 }
 
         This function raises a ValueError when an unhandled status code is returned,
-        when no generic_asset_type_id is given, and when only one of latitude and longitude is.
+        when no generic_asset_type_id is given, and when only one of latitude and longitude is given.
         """
         if generic_asset_type_id is None:
-            # Optional in the signature only so that latitude and longitude, before it, can be left out,
+            # The signature makes this optional only so that latitude and longitude, which come before it, can be left out,
             # while every argument keeps its position.
             raise ValueError("add_asset needs a generic_asset_type_id.")
         if (latitude is None) != (longitude is None):
@@ -1553,7 +1575,7 @@ async def add_asset(
             account_id=account_id,
             generic_asset_type_id=generic_asset_type_id,
         )
-        # A location is optional to FlexMeasures, which many assets have no use for, such as one standing for a community.
+        # FlexMeasures does not require a location, and many assets have no use for one, such as an asset that represents a community.
         if latitude is not None:
             asset["latitude"] = latitude
         if longitude is not None:

src/flexmeasures_client/s2/__init__.py

@@ -127,7 +127,7 @@ def __init__(self, max_size: int = 100, timezone: str = "UTC") -> None:
         self.outgoing_messages_status = SizeLimitOrderedDict(max_size=max_size)
 
         self._timezone = ZoneInfo(timezone)
-        #: The time this handler goes by; a CEM gives each handler it registers its own clock.
+        #: The clock this handler reads the time from. A CEM sets this to its own clock for each handler it registers.
         self._clock: Clock = WallClock()
 
         self.discover()
@@ -253,5 +253,5 @@ def handle_revoke_object(self, message: RevokeObject) -> ReceptionStatus:
         )
 
     def now(self) -> datetime:
-        """The current moment by this handler's clock, in its time zone."""
+        """Return the current moment by this handler's clock, in the handler's timezone."""
         return self._clock.now().astimezone(self._timezone)