Move sensor data ingestion to job queue

[Copilot]

Description

• Move sensor data ingestion (incl. resampling) to a dedicated ingestion job queue so large POSTs can be processed asynchronously when a worker is available
• Keep a synchronous fallback when no ingestion queue or worker is available
• Return 202 Accepted responses with a job_monitor_url and job_id when ingestion is queued
• Keep unchanged-belief detection consistent between synchronous and queued ingestion paths
• Add a configurable sensor data ingestion payload limit via FLEXMEASURES_MAX_SENSOR_DATA_INGESTION_BYTES
• Update API documentation, host documentation, configuration docs, and the OpenAPI spec for queued ingestion and payload limits
• Improve sensor-page upload feedback with polling, longer-lived Toast notifications, and chart/stat refresh after processing finishes
• Use constants for the three possible success status strings when saving beliefs
• Added changelog item in documentation/changelog.rst

Look & Feel

• API requests that post large amounts of sensor data can return quickly with 202 Accepted while ingestion is processed in the background when an ingestion worker is available
• Accepted upload responses include a job status URL that API clients and the UI can use to monitor processing
• Sensor-page uploads show immediate feedback, keep polling the background job, and refresh the chart once processing succeeds
• If no ingestion worker is available, the existing synchronous path still works as a fallback

How to test

1. Start FlexMeasures with Redis configured.
2. Run a worker that handles the ingestion queue, for example:
   flexmeasures jobs run-worker --queue "forecasting|scheduling|ingestion"
3. Upload a large CSV file to a sensor through the UI or POST data to the sensor data API and confirm the request returns 202 Accepted quickly with a job_monitor_url.
4. Confirm the UI shows an accepted/processing Toast, keeps polling the job status endpoint, and refreshes the chart after the worker finishes.
5. Repeat the same flow without an ingestion worker and confirm the synchronous fallback still works.
6. Try a payload larger than FLEXMEASURES_MAX_SENSOR_DATA_INGESTION_BYTES and confirm it is rejected with 413 Payload Too Large.
7. Run the relevant tests:
   • uv run pytest flexmeasures/api/v3_0/tests/test_sensor_data.py
   • uv run pytest flexmeasures/api/v3_0/tests/test_sensors_api.py
   • uv run pytest flexmeasures/data/tests/test_data_ingestion.py

Also, here is a small script to test the sensor data endpoint:

import requests

token_localhost = "xxxxxxxxxxxxxxxxxxxxxxx"

res = requests.post("http://localhost:5000/api/v3_0/sensors/123/data",
    headers = {
        "Authorization": token_localhost,
    },
    json={
        "values": [
            -11.28,
            -11.28,
            -20,
            -11.28
        ],
        "start": "2026-04-20T13:00:00+02:00",
        "duration": "PT1H",
        "unit": "kWh"
    }
)
print(res.text)

Further Improvements

• Support larger ingestion payloads through temporary object/file storage instead of sending the full payload through Redis

Related Items

• Follow-up story: Use HTTP status 202 for job-triggering endpoints #2170

---

Sign-off

• I agree to contribute to the project under Apache 2 License.
• To the best of my knowledge, the proposed patch is not based on code under GPL or other license that is incompatible with FlexMeasures

[read-the-docs-community]

Documentation build overview

📚 flexmeasures | 🛠️ Build #32464225 | 📁 Comparing 6380541 against latest (e97a0c4)

  🔍 Preview build  

25 files changed · + 1 added · ± 24 modified

+ Added

• _autosummary/flexmeasures.data.services.data_ingestion.html

± Modified

• changelog.html
• genindex.html
• http-routingtable.html
• py-modindex.html
• _autosummary/flexmeasures.api.common.schemas.assets.html
• _autosummary/flexmeasures.api.common.schemas.search.html
• _autosummary/flexmeasures.api.common.schemas.sensor_data.html
• _autosummary/flexmeasures.api.common.utils.api_utils.html
• _autosummary/flexmeasures.api.v3_0.accounts.html
• _autosummary/flexmeasures.api.v3_0.assets.html
• and 14 more...

[joshuaunity]

@copilot, add a proper step-by-step guide on how to test this

[Copilot]

@copilot, add a proper step-by-step guide on how to test this

Updated the PR description with a step-by-step testing guide covering both the queued path (with an ingestion worker) and the synchronous fallback path (no workers). See the "How to test" section below.

[joshuaunity]

This works from my tests

From 2.5+ sec to about 300ms

The fallback (run synchronously if no available worker) also works nicely without any issues

[joshuaunity]

This works pretty well, and the code is clean and straightforward.

@nhoening, are there any additions you would like to make to this?

[nhoening]

This works pretty well, and the code is clean and straightforward.

@nhoening, are there any additions you would like to make to this?

So basically you are asking for an early review :)

I also like the code and that there is some documentation already, as well. And the fallback is good.

• Documentation might not catch all places that hosts should see when they get started. For instance, here we don't mention the new queue.
• Create a changelog entry, plus a notice for upgrading - it is recommended to add a worker to the ingestion queue or let existing workers handle it, as well. @Flix6x how would you do this in our infra, for example?
• The function api_utils.save_and_enqueue() is where the real logic happens - which other places use that util function, and are they affected?
• There is a warning about "No workers connected", but we can also add a likewise warning about "No ingestion queue configured".

[joshuaunity]

The function api_utils.save_and_enqueue() is where the real logic happens - which other places use that util function, and are they affected?

Yes, they are all affected and will use the queue if a worker is available.

One more place I found is in the file below, the function post_data_deprecated
flexmeasures/api/v3_0/deprecated.py

[Flix6x]

Create a changelog entry, plus a notice for upgrading - it is recommended to add a worker to the ingestion queue or let existing workers handle it, as well. @Flix6x how would you do this in our infra, for example?

I guess by running flexmeasures jobs run-worker --queue scheduling|ingestion|forecasting, and until I make that change I'd expect the ingestion to still happen within the request.

[nhoening]

I could take on the crowded toasts. I worked on that a bit with Codex in the end.

You don't want to get rid of that Success Toast with the link right? (Link could remain internal, indeed).

I do like that there is new in-progress Toasts every 2 seconds, which you only see as the counter increases. It tells you that we are still checking ...

[Flix6x]

You don't want to get rid of that Success Toast with the link right? (Link could remain internal, indeed).

My first thought was that I like the two ingestion toasts (of which only one is shown in my screenshot) that indicate success for:

1. Job queued
2. Job finished

In particular, I like that they show "Success", where the forecasting toast just says "Info".

But then, I think, isn't it a bit premature to claim success when the job is queued? From a user perspective, the action can only be considered a success when the job is actually finished. The forecasting toasts only show "Success" when the job is finished, so actually I prefer that. After all, you hit the button to get the job done (here: new data in the database), and nothing less should be claimed a success.

I do like that there is new in-progress Toasts every 2 seconds, which you only see as the counter increases. It tells you that we are still checking ...

For the UI, the "Checking status..." followed by actual status updates (i.e. QUEUED, STARTED, etc.) is probably more suitable than a link to the JSON representation of the job itself.

The "Upload received. Processing data in the background" in particular seems redundant.

[Flix6x]

@nhoening if you have an opinion about how long ingestion jobs should be kept around, let's hear it. :) I think we're just using an RQ default right now (500 seconds), unlike for the forecasting and scheduling queues, which we default to keep around for a full week.

[nhoening]

@nhoening if you have an opinion about how long ingestion jobs should be kept around, let's hear it. :) I think we're just using an RQ default right now (500 seconds), unlike for the forecasting and scheduling queues, which we default to keep around for a full week.

Shorter, for sure. Maybe a day.

[nhoening]

You don't want to get rid of that Success Toast with the link right?

I meant actually, that that should probably go, sorry for being unclear.

@joshuaunity wrote a JS function pollJobStatus() (in ui-utils.js) which we could use in a unified way to poll the other jobs, too. But forecasting and scheduling jobs have these dedicated endpoints ...

[nhoening]

Scheduling/forecasting Jobs live a day, their results 7 days:

FLEXMEASURES_JOB_TTL: timedelta = timedelta(days=1)
FLEXMEASURES_PLANNING_TTL: timedelta = timedelta(days=7)

ttl=int(
            current_app.config.get(
                "FLEXMEASURES_JOB_TTL", timedelta(-1)
            ).total_seconds()
        ),
 result_ttl=int(
            current_app.config.get(
                "FLEXMEASURES_PLANNING_TTL", timedelta(-1)
            ).total_seconds()
        ),

Should we do just the same for ingestion? There might be more (smaller and less interesting) jobs for this.

We could simply reuse FLEXMEASURES_JOB_TTL for both right?

[nhoening]

@Flix6x here is one thing about those most-recent-only filters: If in the uploaded data we tick the measured-instantly checkbox, we effectively overwrite belief times in the data, so if there were several per event, we now have multiple beliefs for each event with identical belief time. Then the filter would make sense, right?

[Flix6x]

@Flix6x here is one thing about those most-recent-only filters: If in the uploaded data we tick the measured-instantly checkbox, we effectively overwrite belief times in the data, so if there were several per event, we now have multiple beliefs for each event with identical belief time. Then the filter would make sense, right?

Hmm, best to try it. My first thought is that the filter would still not be needed. A good test might be to post data with a 2-hour resolution to a 6-hour sensor, then hit the (1-hour frequency) replay. My hunch is that with the filter we'd see a realization once every 6 hours, and without the filter we'd see an extra two forecasts, 2 and 4 hours into the 6-hour event.

[Flix6x]

Actually, I shouldn't call them forecasts, but partial realizations. That is, I expect each 6-hour realization to be updated twice.

[nhoening]

@Flix6x here is one thing about those most-recent-only filters: If in the uploaded data we tick the measured-instantly checkbox, we effectively overwrite belief times in the data, so if there were several per event, we now have multiple beliefs for each event with identical belief time. Then the filter would make sense, right?

Hmm, best to try it. My first thought is that the filter would still not be needed. A good test might be to post data with a 2-hour resolution to a 6-hour sensor, then hit the (1-hour frequency) replay. My hunch is that with the filter we'd see a realization once every 6 hours, and without the filter we'd see an extra two forecasts, 2 and 4 hours into the 6-hour event.

I'm not sure I get why this answers our question.

• so the original data in this example does not have multiple beliefs per event - it gets resampled to one event, right?
• why would the replay be 1-hour frequency when I look at a 6-h sensor?

How about adding the filter, but only if the measured-instantly checkbox is checked?

[Flix6x]

so the original data in this example does not have multiple beliefs per event - it gets resampled to one event, right?

It has one belief per 2-hour event, and each of those are recorded right at the end of the 2-hour event. If you resample to 6 hours, then 2 hours into that 6-hour event, you have some partial information about the outcome of the event. And 4 hours in, you have some more. That's how you could get 3 beliefs per 6-hour event.

why would the replay be 1-hour frequency when I look at a 6-h sensor?

Our replay just has a 1-hour frequency by definition. We just don't have UI support right now to pick a different step.

How about adding the filter, but only if the measured-instantly checkbox is checked?

Let's see the test results first.

[joshuaunity]

Scheduling/forecasting Jobs live a day, their results 7 days:

FLEXMEASURES_JOB_TTL: timedelta = timedelta(days=1)
FLEXMEASURES_PLANNING_TTL: timedelta = timedelta(days=7)

ttl=int(
            current_app.config.get(
                "FLEXMEASURES_JOB_TTL", timedelta(-1)
            ).total_seconds()
        ),
 result_ttl=int(
            current_app.config.get(
                "FLEXMEASURES_PLANNING_TTL", timedelta(-1)
            ).total_seconds()
        ),

Should we do just the same for ingestion? There might be more (smaller and less interesting) jobs for this.

We could simply reuse FLEXMEASURES_JOB_TTL for both right?

Ill work on this configuration

[Flix6x]

I did the test. Here's the difference, exactly as I tried to describe.

With keep_only_most_recent_belief=True:

most-recent-beliefs-only.mp4

Without (keep_only_most_recent_belief defaults to False):

all-beliefs.mp4

[nhoening]

I did the test. Here's the difference, exactly as I tried to describe.

With keep_only_most_recent_belief=True:

most-recent-beliefs-only.mp4
Without (keep_only_most_recent_belief defaults to False):

all-beliefs.mp4

So I'm out of the loop by now - what is your suggested course of action seeing this?

[Flix6x]

Before making a decision, I investigated further. The filter was introduced in 513a81e specifically mentioning slow file uploads without it. I verified this issue with a file upload of 1200 rows (2-hour upsampled to 6-hour). So for the sake of UX we should add back the filter.

My initial worry was actually that we'd lose support for uploading datasets containing multiple forecasts per event. However, the file upload already did not support that use case, so I consider it out of scope.

More context: I envision our time series file import should at some point happily accept our own time series file exports.

Follow-ups: I'll re-introduce the filter, fix the failing test and merge.