uploading/posting beliefs works without Redis (again)

documentation/changelog.rst

@@ -32,7 +32,7 @@ New features
 
 Infrastructure / Support
 ----------------------
-* Move sensor data ingestion to a job queue for improved performance when POSTing large amounts of data to the sensor data API, returning a ``202 Accepted`` response with a job status URL when queued [see `PR #2101 <https://www.github.com/FlexMeasures/flexmeasures/pull/2101>`_]
+* Move sensor data ingestion to a job queue for improved performance when POSTing large amounts of data to the sensor data API, returning a ``202 Accepted`` response with a job status URL when queued [see `PR #2101 <https://www.github.com/FlexMeasures/flexmeasures/pull/2101>`_ and `PR #2207 <https://www.github.com/FlexMeasures/flexmeasures/pull/2207>`_]
 * Remove legacy rolling viewpoint forecasting code and utilities after migrating to fixed-point forecasting [see `PR #2082 <https://www.github.com/FlexMeasures/flexmeasures/pull/2082>`_]
 * Upgraded dependencies [see `PR #2114 <https://www.github.com/FlexMeasures/flexmeasures/pull/2114>`_, `PR #2148 <https://www.github.com/FlexMeasures/flexmeasures/pull/2148>`_, `PR #2161 <https://www.github.com/FlexMeasures/flexmeasures/pull/2161>`_ and `PR #2177 <https://www.github.com/FlexMeasures/flexmeasures/pull/2177>`_]
 * Let RQ Dashboard job page load even when job data includes non-JSON-serializable values [see `PR #2200 <https://www.github.com/FlexMeasures/flexmeasures/pull/2200>`_]

flexmeasures/api/common/utils/api_utils.py

@@ -9,10 +9,11 @@
 from datetime import timedelta
 
 from flask import current_app
+from redis.exceptions import ConnectionError as RedisConnectionError
 from werkzeug.exceptions import Forbidden, Unauthorized
 from numpy import array
 from psycopg2.errors import UniqueViolation
-from rq import Worker
+from rq import Queue, Worker
 from rq.job import Job
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
@@ -118,6 +119,14 @@ def save_and_enqueue(
     return invalid_replacement()
 
 
+def queue_has_connected_workers(queue: Queue) -> bool:
+    """Return True when an RQ queue is reachable and has connected workers."""
+    try:
+        return bool(Worker.all(queue=queue))
+    except RedisConnectionError:
+        return False
+
+
 def process_sensor_data_ingestion(
     sensor_id: int,
     user_id: int,
@@ -138,14 +147,13 @@ def process_sensor_data_ingestion(
         current_app.logger.warning(
             "No ingestion queue configured. Processing sensor data directly."
         )
-    else:
-        workers = Worker.all(queue=ingestion_queue)
-        if workers:
-            forecasting_job_ids = (
-                [job.id for job in forecasting_jobs]
-                if forecasting_jobs is not None
-                else None
-            )
+    elif queue_has_connected_workers(ingestion_queue):
+        forecasting_job_ids = (
+            [job.id for job in forecasting_jobs]
+            if forecasting_jobs is not None
+            else None
+        )
+        try:
             job = ingestion_queue.enqueue(
                 add_beliefs_to_db_and_enqueue_forecasting_jobs,
                 sensor_id=sensor_id,
@@ -166,14 +174,21 @@ def process_sensor_data_ingestion(
                     ).total_seconds()
                 ),
             )
+        except RedisConnectionError as error:
+            current_app.logger.warning(
+                "Unable to enqueue ingestion job in Redis queue %s: %s. Processing synchronously instead.",
+                getattr(ingestion_queue, "name", "<unknown>"),
+                error,
+            )
+        else:
             return request_accepted_for_processing(
                 job.id,
                 "Sensor data has been accepted for processing.",
             )
-        else:
-            current_app.logger.warning(
-                "No workers connected to the ingestion queue. Processing sensor data directly."
-            )
+    else:
+        current_app.logger.warning(
+            "No workers connected to the ingestion queue. Processing sensor data directly."
+        )
 
     status = add_beliefs_to_db_and_enqueue_forecasting_jobs(
         sensor_id=sensor_id,

flexmeasures/api/v3_0/tests/test_sensor_data.py

@@ -3,6 +3,7 @@
 from datetime import timedelta
 from flask import current_app, url_for
 import pytest
+from redis.exceptions import ConnectionError as RedisConnectionError
 from sqlalchemy import event
 from sqlalchemy.engine import Engine
 
@@ -326,6 +327,36 @@ def test_post_sensor_data_returns_accepted_job(
     assert "data" not in job.kwargs
 
 
+@pytest.mark.parametrize(
+    "requesting_user", ["test_supplier_user_4@seita.nl"], indirect=True
+)
+def test_post_sensor_data_falls_back_when_redis_unavailable(
+    client,
+    setup_api_test_data,
+    requesting_user,
+    monkeypatch,
+):
+    def raise_connection_error(queue):
+        raise RedisConnectionError("Connection refused")
+
+    monkeypatch.setattr(
+        "flexmeasures.api.common.utils.api_utils.Worker.all",
+        raise_connection_error,
+    )
+    current_app.queues["ingestion"].empty()
+    post_data = make_sensor_data_request_for_gas_sensor()
+    sensor = setup_api_test_data["some gas sensor"]
+
+    response = client.post(
+        url_for("SensorAPI:post_data", id=sensor.id),
+        json=post_data,
+    )
+
+    assert response.status_code == 200
+    assert response.json["status"] == "PROCESSED"
+    assert current_app.queues["ingestion"].count == 0
+
+
 @pytest.mark.parametrize(
     "requesting_user", ["test_supplier_user_4@seita.nl"], indirect=True
 )