Move sensor data ingestion to job queue

docker-compose.yml

@@ -122,7 +122,7 @@ services:
       if [ -f /usr/var/flexmeasures-instance/requirements.txt ]; then
         pip install --no-cache-dir -r /usr/var/flexmeasures-instance/requirements.txt
       fi
-      flexmeasures jobs run-worker --name flexmeasures-worker --queue forecasting\|scheduling
+      flexmeasures jobs run-worker --name flexmeasures-worker --queue forecasting\|scheduling\|ingestion
   test-db:
     image: postgres
     expose:

documentation/changelog.rst

@@ -7,6 +7,14 @@ FlexMeasures Changelog
 v0.33.0 | May XX, 2026
 ============================
 
+.. note:: It is recommended to assign a worker to the ``ingestion`` queue (or configure existing workers to handle it, as well),
+          so that sensor data posted via the API is processed asynchronously [see `PR #2101 <https://www.github.com/FlexMeasures/flexmeasures/pull/2101>`_].
+          For instance, using:
+
+          .. code-block:: bash
+
+             $ flexmeasures jobs run-worker --queue "forecasting|scheduling|ingestion"
+
 .. warning:: We are deprecating ``FLEXMEASURES_MONITORING_MAIL_RECIPIENTS`` in favor of ``FLEXMEASURES_DEFAULT_MONITORING_MAIL_RECIPIENTS``.
 
 New features
@@ -21,6 +29,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>`_]
 * 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>`_]
 * Run ``flexmeasures jobs run-worker`` with RQ's embedded scheduler on by default so jobs created with ``enqueue_in`` are promoted from the scheduled registry when due; pass ``--without-scheduler`` to disable [see `PR #2112 <https://www.github.com/FlexMeasures/flexmeasures/pull/2112>`_]
@@ -32,6 +41,7 @@ Bugfixes
 -----------
 * Fix forecasting regressor filtering to use only regressor beliefs known at the forecast ``belief_time`` [see `PR #2134 <https://www.github.com/FlexMeasures/flexmeasures/pull/2134>`_]
 * Check read permissions for sensors referenced in forecasting and scheduling config payloads, and return a clearer 403 error when a referenced sensor is not readable [see `PR #2096 <https://www.github.com/FlexMeasures/flexmeasures/pull/2096>`_ and `PR #2125 <https://www.github.com/FlexMeasures/flexmeasures/pull/2125>`_]
+* Clean up stale sensor references from ``flex-config`` and ``sensors_to_show`` when deleting a sensor, using JSONB queries to find affected assets before pruning those references [see `PR #2106 <https://www.github.com/FlexMeasures/flexmeasures/pull/2106>`_]
 * Standardize resolution formatting across API endpoints for consistent response payloads [see `PR #2152 <https://www.github.com/FlexMeasures/flexmeasures/pull/2152>`_]
 * Make the auth check for CLI commands work with ``flask``, too, instead of only with the ``flexmeasures`` alias [see `PR #2169 <https://www.github.com/FlexMeasures/flexmeasures/pull/2169>`_]
 
@@ -93,7 +103,6 @@ New features
 * Separate the ``StorageScheduler``'s tie-breaking preference for a full :abbr:`SoC (state of charge)` from its reported energy costs [see `PR #2023 <https://www.github.com/FlexMeasures/flexmeasures/pull/2023>`_ and `PR #2108 <https://www.github.com/FlexMeasures/flexmeasures/pull/2108>`_]
 * Improve asset graph hover interaction with a vertical ruler across subcharts, while keeping hover dots for easier visual tracking [see `PR #2079 <https://www.github.com/FlexMeasures/flexmeasures/pull/2079>`_]
 * Improve asset audit log messages for JSON field edits (especially ``sensors_to_show`` and nested flex-config values) [see `PR #2055 <https://www.github.com/FlexMeasures/flexmeasures/pull/2055>`_]
-* Clean up stale sensor references from ``flex-config`` and ``sensors_to_show`` when deleting a sensor, using JSONB queries to find affected assets before pruning those references [see `PR #2106 <https://www.github.com/FlexMeasures/flexmeasures/pull/2106>`_]
 
 Infrastructure / Support
 ----------------------

documentation/configuration.rst

@@ -347,6 +347,14 @@ Set a negative value to persist forever.
 
 Default: ``3600``
 
+FLEXMEASURES_MAX_SENSOR_DATA_INGESTION_BYTES
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+Maximum request body size for sensor data posted to the sensor data API, both for JSON data and file uploads.
+Set to ``None`` to disable this FlexMeasures-specific limit.
+
+Default: ``3 * 1024 * 1024``
+
 .. _datasource_config:
 
 FLEXMEASURES_DEFAULT_DATASOURCE

documentation/host/installation.rst

@@ -385,7 +385,7 @@ Then, start workers in a console (or some other method to keep a long-running pr
 
 .. code-block:: bash
 
-   $ flexmeasures jobs run-worker --queue "scheduling|forecasting"
+   $ flexmeasures jobs run-worker --queue "scheduling|forecasting|ingestion"
 
 
 You can go to `http://localhost:5000/tasks/` and see the state of job queues and find individual jobs (and investigate why they failed, for instance).

documentation/host/queues.rst

@@ -24,25 +24,30 @@ Here is how to run one worker for each kind of job (in separate terminals):
 
 .. code-block:: bash
 
-   $ flexmeasures jobs run-worker --name our-only-worker --queue forecasting|scheduling
+   $ flexmeasures jobs run-worker --name our-only-worker --queue forecasting|scheduling|ingestion
 
 Running multiple workers in parallel might be a great idea.
 
 .. code-block:: bash
 
    $ flexmeasures jobs run-worker --name forecaster --queue forecasting
    $ flexmeasures jobs run-worker --name scheduler --queue scheduling
+   $ flexmeasures jobs run-worker --name ingester --queue ingestion
 
 You can also clear the job queues:
 
 .. code-block:: bash
 
    $ flexmeasures jobs clear-queue --queue forecasting
    $ flexmeasures jobs clear-queue --queue scheduling
+   $ flexmeasures jobs clear-queue --queue ingestion
 
 
 When the main FlexMeasures process runs (e.g. by ``flexmeasures run``\ ), the queues of forecasting and scheduling jobs can be visited at ``http://localhost:5000/tasks/forecasting`` and ``http://localhost:5000/tasks/schedules``\ , respectively (by admins).
 
+.. note::
+   The ``ingestion`` queue is used for sensor data posted via the API. If the queue is not configured, or if no worker is connected to it, data is processed synchronously (in the web process) with a warning logged. Running a dedicated ingestion worker is recommended in production to keep API responses fast when large amounts of data are posted. When ingestion is queued, the API returns ``202 Accepted`` with a job status URL.
+
 
 
 Inspect the queue and jobs

documentation/tut/forecasting_scheduling.rst

@@ -28,6 +28,7 @@ Start to run one worker for each kind of job (in a separate terminal):
 
    $ flexmeasures jobs run-worker --queue forecasting
    $ flexmeasures jobs run-worker --queue scheduling
+   $ flexmeasures jobs run-worker --queue ingestion
 
 
 You can also clear the job queues:
@@ -36,6 +37,7 @@ You can also clear the job queues:
 
    $ flexmeasures jobs clear-queue --queue forecasting
    $ flexmeasures jobs clear-queue --queue scheduling
+   $ flexmeasures jobs clear-queue --queue ingestion
 
 
 When the main FlexMeasures process runs (e.g. by ``flexmeasures run``), the queues of forecasting and scheduling jobs can be visited at ``http://localhost:5000/tasks/forecasting`` and ``http://localhost:5000/tasks/schedules``, respectively (by admins).

flexmeasures/api/__init__.py

@@ -17,9 +17,11 @@
 from flexmeasures.data.models.user import User
 from flexmeasures.api.common.utils.args_parsing import (
     validation_error_handler,
+    request_entity_too_large_handler,
 )
 from flexmeasures.api.common.responses import invalid_sender
 from flexmeasures.data.schemas.utils import FMValidationError
+from werkzeug.exceptions import RequestEntityTooLarge
 from flexmeasures.api.v3_0.users import AuthRequestSchema
 
 # The api blueprint. It is registered with the Flask app (see app.py)
@@ -150,6 +152,7 @@ def register_at(app: Flask):
 
     # handle API specific errors
     app.register_error_handler(FMValidationError, validation_error_handler)
+    app.register_error_handler(RequestEntityTooLarge, request_entity_too_large_handler)
     app.register_error_handler(IntegrityError, catch_timed_belief_replacements)
     app.unauthorized_handler_api = invalid_sender
 

flexmeasures/api/common/responses.py

@@ -3,6 +3,8 @@
 import inflect
 from functools import wraps
 
+from flask import url_for
+
 from flexmeasures.auth.error_handling import FORBIDDEN_MSG, FORBIDDEN_STATUS_CODE
 
 p = inflect.engine()
@@ -62,7 +64,7 @@ def already_received_and_successfully_processed(message: str) -> ResponseTuple:
 
 
 @BaseMessage(
-    "Some of the data represents a replacement, which is reserved for customized servers. If you are hosting FlexMeasures, you can enable replacements by setting FLEXMEASURES_ALLOW_DATA_OVERWRITE=True in the configuration settings. Alternatively, update the prior in your request."
+    "Some of the data represents a replacement: existing values would be changed for the same sensor, source, event timestamp and belief time. This is reserved for customized servers. If you are hosting FlexMeasures, you can enable replacements by setting FLEXMEASURES_ALLOW_DATA_OVERWRITE=True in the configuration settings. Alternatively, submit the data with a different prior, or check whether the same file was already uploaded with different values."
 )
 def invalid_replacement(message: str) -> ResponseTuple:
     return (
@@ -378,6 +380,25 @@ def request_processed(message: str) -> ResponseTuple:
     return dict(status="PROCESSED", message=message), 200
 
 
+def request_accepted_for_processing(
+    job_id: str,
+    message: str = "Request has been accepted for processing.",
+) -> ResponseTuple:
+    return (
+        dict(
+            status="ACCEPTED",
+            message=message,
+            job_monitor_url=url_for("JobAPI:get_job_status", uuid=job_id),
+            job_id=job_id,
+        ),
+        202,
+    )
+
+
+def request_too_large(message: str) -> ResponseTuple:
+    return dict(result="Rejected", status="PAYLOAD_TOO_LARGE", message=message), 413
+
+
 def pluralize(usef_role_name: str) -> str:
     """Adding a trailing 's' works well for USEF roles."""
     return "%ss" % usef_role_name

flexmeasures/api/common/schemas/sensor_data.py

@@ -12,6 +12,7 @@
 
 from flexmeasures.data import ma
 from flexmeasures.data.models.time_series import Sensor
+from flexmeasures.data.models.user import User
 from flexmeasures.api.common.schemas.sensors import (
     SensorEntityAddressField,
     SensorIdField,
@@ -328,6 +329,10 @@ class PostSensorDataSchema(SensorDataDescriptionSchema):
     This schema includes data (values) and still describes it.
     """
 
+    def __init__(self, *args, source_user: User | None = None, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.source_user = source_user
+
     values = PolyField(
         deserialization_schema_selector=select_schema_to_ensure_list_of_floats,
         serialization_schema_selector=select_schema_to_ensure_list_of_floats,
@@ -469,12 +474,11 @@ def possibly_upsample_values(data):
             )
         return data
 
-    @staticmethod
-    def load_bdf(sensor_data: dict) -> BeliefsDataFrame:
+    def load_bdf(self, sensor_data: dict) -> BeliefsDataFrame:
         """
         Turn the de-serialized and validated data into a BeliefsDataFrame.
         """
-        source = get_or_create_source(current_user)
+        source = get_or_create_source(self.source_user or current_user)
         num_values = len(sensor_data["values"])
         event_resolution = sensor_data["duration"] / num_values
         start = sensor_data["start"]
@@ -512,6 +516,29 @@ def load_bdf(sensor_data: dict) -> BeliefsDataFrame:
         )
 
 
+class PostSensorDataRequestSchema(PostSensorDataSchema):
+    """Validate posted sensor data without building a BeliefsDataFrame."""
+
+    @post_load()
+    def post_load_sequence(self, data: dict, **kwargs) -> dict:
+        sensor_data = {
+            "values": data["values"],
+            "start": datetime_isoformat(data["start"]),
+            "duration": duration_isoformat(data["duration"]),
+            "unit": data["unit"],
+        }
+        if "prior" in data:
+            sensor_data["prior"] = datetime_isoformat(data["prior"])
+        elif "horizon" in data:
+            sensor_data["horizon"] = duration_isoformat(data["horizon"])
+        else:
+            # Preserve request-time semantics when processing happens later in a worker.
+            sensor_data["prior"] = datetime_isoformat(server_now())
+        if "type" in data:
+            sensor_data["type"] = data["type"]
+        return dict(sensor=data["sensor"], sensor_data=sensor_data)
+
+
 class GetSensorDataSchemaEntityAddress(GetSensorDataSchema):
     """DEPRECATED, only here to support deprecated endpoints"""
 

flexmeasures/api/common/utils/api_utils.py

@@ -12,21 +12,30 @@
 from werkzeug.exceptions import Forbidden, Unauthorized
 from numpy import array
 from psycopg2.errors import UniqueViolation
+from rq import Worker
 from rq.job import Job, JobStatus, NoSuchJobError
 from sqlalchemy import select
 from sqlalchemy.exc import IntegrityError
 
 from flexmeasures.data import db
 from flexmeasures.data.models.audit_log import AssetAuditLog
 from flexmeasures.data.models.user import Account
+from flexmeasures.data.services.data_ingestion import (
+    add_beliefs_to_db_and_enqueue_forecasting_jobs,
+)
 from flexmeasures.data.models.generic_assets import GenericAsset
 from flexmeasures.data.models.time_series import Sensor
-from flexmeasures.data.utils import save_to_db
+from flexmeasures.data.utils import (
+    SAVE_TO_DB_SUCCESS,
+    SAVE_TO_DB_SUCCESS_BUT_NOTHING_NEW,
+    SAVE_TO_DB_SUCCESS_WITH_UNCHANGED_BELIEFS_SKIPPED,
+)
 from flexmeasures.auth.policy import check_access
 from flexmeasures.api.common.responses import (
     invalid_replacement,
     ResponseTuple,
     request_processed,
+    request_accepted_for_processing,
     already_received_and_successfully_processed,
 )
 from flexmeasures.data.schemas.generic_assets import GenericAssetSchema as AssetSchema
@@ -146,20 +155,95 @@ def save_and_enqueue(
     forecasting_jobs: list[Job] | None = None,
     save_changed_beliefs_only: bool = True,
 ) -> ResponseTuple:
-    # Attempt to save
-    status = save_to_db(data, save_changed_beliefs_only=save_changed_beliefs_only)
-    db.session.commit()
-
-    # Only enqueue forecasting jobs upon successfully saving new data
-    if status[:7] == "success" and status != "success_but_nothing_new":
-        enqueue_forecasting_jobs(forecasting_jobs)
+    status = add_beliefs_to_db_and_enqueue_forecasting_jobs(
+        data,
+        forecasting_jobs=forecasting_jobs,
+        save_changed_beliefs_only=save_changed_beliefs_only,
+    )
 
     # Pick a response
-    if status == "success":
+    if status == SAVE_TO_DB_SUCCESS:
+        return request_processed()
+    elif status in (
+        SAVE_TO_DB_SUCCESS_WITH_UNCHANGED_BELIEFS_SKIPPED,
+        SAVE_TO_DB_SUCCESS_BUT_NOTHING_NEW,
+    ):
+        return already_received_and_successfully_processed()
+    return invalid_replacement()
+
+
+def process_sensor_data_ingestion(
+    sensor_id: int,
+    user_id: int,
+    sensor_data: dict | None = None,
+    uploaded_files: list[dict] | None = None,
+    upload_data: dict | None = None,
+    forecasting_jobs: list[Job] | None = None,
+    save_changed_beliefs_only: bool = True,
+) -> ResponseTuple:
+    """Process sensor data ingestion asynchronously when possible.
+
+    If an ingestion queue with connected workers is available, enqueue a background
+    job and return ``202 Accepted``. Otherwise, process the data synchronously and
+    return the resulting ingestion response.
+    """
+    ingestion_queue = current_app.queues.get("ingestion")
+    if ingestion_queue is None:
+        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
+            )
+            job = ingestion_queue.enqueue(
+                add_beliefs_to_db_and_enqueue_forecasting_jobs,
+                sensor_id=sensor_id,
+                user_id=user_id,
+                sensor_data=sensor_data,
+                uploaded_files=uploaded_files,
+                upload_data=upload_data,
+                forecasting_job_ids=forecasting_job_ids,
+                save_changed_beliefs_only=save_changed_beliefs_only,
+                meta={"sensor_id": sensor_id},
+                ttl=current_app.config.get(
+                    "FLEXMEASURES_JOB_TTL", timedelta(-1)
+                ).total_seconds(),
+                # No need to keep ingestion results for the FLEXMEASURES_PLANNING_TTL
+                result_ttl=int(
+                    current_app.config.get(
+                        "FLEXMEASURES_JOB_TTL", timedelta(-1)
+                    ).total_seconds()
+                ),
+            )
+            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."
+            )
+
+    status = add_beliefs_to_db_and_enqueue_forecasting_jobs(
+        sensor_id=sensor_id,
+        user_id=user_id,
+        sensor_data=sensor_data,
+        uploaded_files=uploaded_files,
+        upload_data=upload_data,
+        forecasting_jobs=forecasting_jobs,
+        save_changed_beliefs_only=save_changed_beliefs_only,
+    )
+
+    if status == SAVE_TO_DB_SUCCESS:
         return request_processed()
     elif status in (
-        "success_with_unchanged_beliefs_skipped",
-        "success_but_nothing_new",
+        SAVE_TO_DB_SUCCESS_WITH_UNCHANGED_BELIEFS_SKIPPED,
+        SAVE_TO_DB_SUCCESS_BUT_NOTHING_NEW,
     ):
         return already_received_and_successfully_processed()
     return invalid_replacement()