Run reports via CLI and API

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\|ingestion
+      flexmeasures jobs run-worker --name flexmeasures-worker --queue forecasting\|scheduling\|ingestion\|reporting
   test-db:
     image: postgres
     expose:

documentation/api/change_log.rst

@@ -11,6 +11,7 @@ v3.0-34 | September 2, 2026
 
 v3.0-33 | September 1, 2026
 """""""""""""""""""""""""""
+- Added ``POST /api/v3_0/assets/<id>/reports/trigger`` to queue a one-off report as a background job. It returns ``202 Accepted`` with the canonical ``job`` and ``job-url`` fields, and shares the trigger rate limit with forecast and schedule endpoints.
 - Added ``GET /api/v3_0/assets/<id>/automations`` and ``GET /api/v3_0/assets/<id>/automations/<automation_id>`` for listing and inspecting forecast automations, including the sensors an automation reads from and writes to. Each automation shows the IANA ``timezone`` in which its cron expression is interpreted, and a ``cursor``: the offset-aware UTC time of the most recent run it committed to. The cursor advances just before queueing, so it does not indicate that queueing or the forecast itself succeeded. Asset job entries now include ``created_via`` provenance; automation identity is included only when the caller may read that automation.
 - Added ``GET /api/v3_0/sources/<id>`` to show the full record of one data source, including the attributes in which data generators store their configuration.
 

documentation/changelog.rst

@@ -17,6 +17,7 @@ v1.1.0 | September XX, 2026
 New features
 -------------
 
+* Run one-off reports as background jobs from the CLI or the asset API, with sensor-level authorization and a dedicated reporting worker queue [see `PR #2298 <https://github.com/FlexMeasures/flexmeasures/pull/2298>`_]
 * A single automation can now be run on demand, from the CLI (``flexmeasures jobs run-automation``), the API (``POST /assets/<id>/automations/<automation_id>/trigger``) and the asset's *Automations* page (a *Run now* button), which is useful to try out a new automation, to re-run one after fixing what made it fail, or to refresh its results after late input data arrived [see `PR #2460 <https://www.github.com/FlexMeasures/flexmeasures/pull/2460>`_]
 * Changing the selected time range on an asset or sensor chart now only loads the data that is actually new, instead of reloading the whole range, which makes stepping through or extending a long period much faster; reloading the page, or leaving it open for five minutes, still fetches everything afresh [see `PR #2433 <https://www.github.com/FlexMeasures/flexmeasures/pull/2433>`_]
 

documentation/cli/change_log.rst

@@ -4,6 +4,11 @@
 FlexMeasures CLI Changelog
 **********************
 
+since v1.1.0 | September XX, 2026
+=================================
+
+* Add ``flexmeasures add report --as-job`` and the ``reporting`` worker queue for asynchronous one-off reports.
+
 since v1.0.0 | August 11, 2026
 =================================
 

documentation/configuration.rst

@@ -346,7 +346,7 @@ Default: ``timedelta(days=1)``
 FLEXMEASURES_DEFAULT_JOB_TIMEOUT
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-Default timeout for jobs (e.g. forecasting, scheduling and ingestion), expressed as a fixed ISO 8601 duration.
+Default timeout for jobs (e.g. forecasting, scheduling, ingestion and reporting), expressed as a fixed ISO 8601 duration.
 Jobs that exceed this timeout are moved to RQ's failed queue.
 
 Default: ``timedelta(seconds=180)`` (``"PT180S"``)
@@ -356,9 +356,9 @@ FLEXMEASURES_JOB_TIMEOUT
 
 Timeouts per queue, expressed as fixed ISO 8601 durations.
 Queue-specific values override ``FLEXMEASURES_DEFAULT_JOB_TIMEOUT``.
-Supported queue names are ``forecasting``, ``scheduling`` and ``ingestion``.
+Supported queue names are ``forecasting``, ``scheduling``, ``ingestion`` and ``reporting``.
 
-Example: ``{"forecasting": "PT2M", "scheduling": "PT5M", "ingestion": "PT30S"}``
+Example: ``{"forecasting": "PT2M", "scheduling": "PT5M", "ingestion": "PT30S", "reporting": "PT10M"}``
 
 Default: ``{}``
 
@@ -925,9 +925,8 @@ Default: ``"500 per minute"``
 FLEXMEASURES_API_TRIGGER_RATE_LIMIT
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 
-How often a client may trigger a schedule or a forecast. This is the expensive work, so this limit is stricter
-than the default one. The trigger endpoints share this budget, so triggering a forecast and triggering a schedule
-draw on the same one.
+How often a client may trigger a schedule, forecast or report. This is the expensive work, so this limit is stricter
+than the default one. The trigger endpoints share this budget, so all three kinds of computation draw on the same one.
 
 Default: ``"10 per 5 minutes"``
 

documentation/features/reporting.rst

@@ -17,6 +17,12 @@ We added an infrastructure that allows us to define computation pipelines and CL
 - ``flexmeasures show reporters``
 - ``flexmeasures add report``
 
+Reports can be queued for asynchronous processing with ``flexmeasures add report --as-job``.
+Run ``flexmeasures jobs run-worker --queue reporting`` to process these jobs. A one-off report
+can also be queued through ``POST /api/v3_0/assets/<id>/reports/trigger``. The caller needs read
+access to every input and configuration sensor and permission to record data on each output;
+outputs are limited to the asset in the URL and its descendants.
+
 The reporter classes we are designing are using pandas under the hood and can be sub-classed, allowing us to build new reporters from stable simpler ones, and even pipelines. Remember: re-use is developer power!
 
 We believe this infrastructure will become very powerful and enable FlexMeasures hosts and plugin developers to implement exciting new features.
@@ -122,4 +128,4 @@ The input sensor stores the power/energy flow, and the output sensor will store
 Here, the ``ProfitOrLossReporter`` used as source (with Id 6) is the one we configured above.
 With the offsets, we control the timing ― we indicate that we want the new report to encompass the day of tomorrow (see Pandas offset strings).
 
-The report sensor will now store all costs which we know will be made tomorrow by the  schedule.
+The report sensor will now store all costs which we know will be made tomorrow by the  schedule.

documentation/host/installation.rst

@@ -382,7 +382,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|ingestion"
+   $ flexmeasures jobs run-worker --queue "scheduling|forecasting|ingestion|reporting"
 
 
 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,7 +24,7 @@ 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|ingestion
+   $ flexmeasures jobs run-worker --name our-only-worker --queue "forecasting|scheduling|ingestion|reporting"
 
 Running multiple workers in parallel might be a great idea.
 
@@ -33,6 +33,7 @@ Running multiple workers in parallel might be a great idea.
    $ flexmeasures jobs run-worker --name forecaster --queue forecasting
    $ flexmeasures jobs run-worker --name scheduler --queue scheduling
    $ flexmeasures jobs run-worker --name ingester --queue ingestion
+   $ flexmeasures jobs run-worker --name reporter --queue reporting
 
 You can also clear the job queues:
 
@@ -41,6 +42,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
+   $ flexmeasures jobs clear-queue --queue reporting
 
 
 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/common/schemas/tests/test_sensor_data_schema.py

@@ -2,6 +2,7 @@
 import json
 import pytest
 import pytz
+from rq.job import JobStatus
 
 from marshmallow import ValidationError
 import pandas as pd
@@ -465,7 +466,7 @@ def test_asset_sensors_metadata(
 
 
 def test_build_asset_jobs_data(db, app, add_battery_assets, clean_redis):
-    """Check that we get both types of jobs for a battery asset."""
+    """Check that we get scheduling, forecasting and reporting jobs."""
     battery_asset = add_battery_assets["Test battery"]
     battery = battery_asset.sensors[0]
     tz = pytz.timezone("Europe/Amsterdam")
@@ -495,26 +496,47 @@ def test_build_asset_jobs_data(db, app, add_battery_assets, clean_redis):
         },
     )
     forecasting_job = app.queues["forecasting"].fetch_job(pipeline_returns["job_id"])
+    reporting_job = app.queues["reporting"].enqueue(sum, [1, 2])
+    reporting_job.meta["exception"] = "report failed"
+    reporting_job.save_meta()
+    reporting_job.set_status(JobStatus.FAILED)
+    app.job_cache.add(
+        battery.id,
+        reporting_job.id,
+        queue="reporting",
+        asset_or_sensor_type="sensor",
+    )
 
     jobs_data = build_asset_jobs_data(battery_asset)
     forecasting_jobs_data = [j for j in jobs_data if j["queue"] == "forecasting"]
     scheduling_jobs_data = [j for j in jobs_data if j["queue"] == "scheduling"]
+    reporting_jobs_data = [j for j in jobs_data if j["queue"] == "reporting"]
     assert len(forecasting_jobs_data) == 1
     assert scheduling_jobs_data
+    assert len(reporting_jobs_data) == 1
+    assert (
+        reporting_jobs_data[0]["err"] == "Reporting job failed with str: report failed"
+    )
     scheduling_job_ids = set()
     for job_data in jobs_data:
         metadata = json.loads(job_data["metadata"])
         if job_data["queue"] == "forecasting":
             assert metadata["job_id"] == forecasting_job.id
             assert job_data["entity"] == f"sensor: {battery.name} (Id: {battery.id})"
-        else:
+            assert job_data["status"] == "queued"
+        elif job_data["queue"] == "scheduling":
             scheduling_job_ids.add(metadata["job_id"])
-        assert job_data["status"] == "queued"
+            assert job_data["status"] == "queued"
+        else:
+            assert metadata["job_id"] == reporting_job.id
+            assert job_data["status"] == JobStatus.FAILED
 
     assert scheduling_job.id in scheduling_job_ids
 
     # Clean up queues
     app.queues["scheduling"].empty()
     app.queues["forecasting"].empty()
+    app.queues["reporting"].empty()
     assert app.queues["scheduling"].count == 0
     assert app.queues["forecasting"].count == 0
+    assert app.queues["reporting"].count == 0

flexmeasures/api/common/utils/api_utils.py

@@ -25,8 +25,8 @@
     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.queries.generic_assets import asset_is_in_subtree
+from flexmeasures.data.models.time_series import Sensor
 from flexmeasures.data.utils import (
     SAVE_TO_DB_SUCCESS,
     SAVE_TO_DB_SUCCESS_BUT_NOTHING_NEW,

flexmeasures/api/v3_0/__init__.py

@@ -42,6 +42,7 @@
 )
 from flexmeasures.data.schemas.annotations import AnnotationSchema
 from flexmeasures.data.schemas.generic_assets import GenericAssetSchema as AssetSchema
+from flexmeasures.data.schemas.reporting import ReportTriggerSchema
 from flexmeasures.data.schemas.sensors import QuantitySchema, TimeSeriesSchema
 from flexmeasures.data.schemas.account import (
     AccountSchema,
@@ -222,6 +223,7 @@ def create_openapi_specs(app: Flask):
         ("AssetAPIQuerySchema", AssetAPIQuerySchema),
         ("AssetSchema", AssetSchema),
         ("AnnotationSchema", AnnotationSchema),
+        ("ReportTriggerSchema", ReportTriggerSchema),
         ("CopyAssetSchema", CopyAssetSchema),
         ("DefaultAssetViewJSONSchema", DefaultAssetViewJSONSchema),
         ("AccountSchema", AccountSchema(partial=True)),

flexmeasures/api/v3_0/assets.py

@@ -57,13 +57,21 @@
     run_automation,
 )
 from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType
+from flexmeasures.data.models.reporting import Reporter
 from flexmeasures.data.queries.generic_assets import (
+    asset_is_in_subtree,
     filter_assets_under_root,
     query_assets_by_search_terms,
 )
 from flexmeasures.data.queries.utils import id_prefix_filter
 from flexmeasures.data.schemas import AwareDateTimeField
 from flexmeasures.data.schemas.annotations import AnnotationSchema
+from flexmeasures.data.schemas.reporting import ReportTriggerSchema
+from flexmeasures.data.services.data_generators import (
+    check_sensor_access,
+    resolve_data_generator_sensors,
+)
+from flexmeasures.data.services.data_sources import get_data_generator
 from flexmeasures.data.services.annotations import prepare_annotations_for_chart
 from flexmeasures.data.schemas.generic_assets import (
     GenericAssetSchema as AssetSchema,
@@ -1892,6 +1900,129 @@ def update_keep_legends_below_graphs(self, **kwargs):
             "message": "Default legend position updated successfully.",
         }, 200
 
+    @route("/<id>/reports/trigger", methods=["POST"])
+    @limit_triggers()
+    @use_kwargs({"asset": AssetIdField(data_key="id")}, location="path")
+    @permission_required_for_context("create-children", ctx_arg_name="asset")
+    @as_json
+    def trigger_report(self, id: int, asset: GenericAsset):
+        """
+        .. :quickref: Assets; Trigger a one-off reporting job for this asset.
+        ---
+        post:
+          summary: Trigger a one-off reporting job for this asset.
+          description: |
+            Queue a one-off report for a worker processing the `reporting` queue.
+            The caller must be able to read every input/configuration sensor and
+            record data on every output sensor. Each output must belong to the
+            asset in the URL or one of its descendants.
+          security:
+            - ApiKeyAuth: []
+          parameters:
+            - in: path
+              name: id
+              required: true
+              $ref: '#/components/parameters/AssetIdPath'
+          requestBody:
+            content:
+              application/json:
+                schema: ReportTriggerSchema
+          responses:
+            202:
+              description: ACCEPTED
+              content:
+                application/json:
+                  schema:
+                    type: object
+                    required:
+                      - status
+                      - message
+                      - job
+                      - job-url
+                    properties:
+                      status:
+                        type: string
+                        enum:
+                          - ACCEPTED
+                      message:
+                        type: string
+                      job:
+                        type: string
+                        description: UUID of the queued reporting job.
+                      job-url:
+                        type: string
+                        format: uri
+                        description: URL to query the generic job status API.
+                  example:
+                    status: ACCEPTED
+                    message: Request has been accepted for processing.
+                    job: 364bfd06-c1fa-430b-8d25-8f5a547651fb
+                    job-url: /api/v3_0/jobs/364bfd06-c1fa-430b-8d25-8f5a547651fb
+            400:
+              description: INVALID_REQUEST, REQUIRED_INFO_MISSING, UNEXPECTED_PARAMS
+            401:
+              description: UNAUTHORIZED
+            403:
+              description: INVALID_SENDER
+            422:
+              description: UNPROCESSABLE_ENTITY
+          tags:
+            - Assets
+        """
+        body = request.get_json(silent=True)
+        if not body:
+            return unprocessable_entity("No JSON data provided.")
+        try:
+            report_data = ReportTriggerSchema().load(body)
+        except ValidationError as exc:
+            return unprocessable_entity(exc.messages)
+
+        try:
+            reporter = get_data_generator(
+                source=None,  # pre-defined app.data_generators
+                model=report_data["reporter"],
+                config=report_data["config"],
+                save_config=True,
+                data_generator_type=Reporter,
+            )
+        except ValidationError as exc:
+            db.session.rollback()
+            return unprocessable_entity({"config": exc.messages})
+        if reporter is None:
+            db.session.rollback()
+            return unprocessable_entity(
+                f"Reporter class `{report_data['reporter']}` not available."
+            )
+
+        parameters = report_data["parameters"]
+        try:
+            deserialized_parameters = reporter._parameters_schema.load(parameters)
+            report_sensors = resolve_data_generator_sensors(
+                reporter, deserialized_parameters
+            )
+            check_sensor_access(
+                report_sensors["input_sensors"], report_sensors["output_sensors"]
+            )
+            for output_sensor in report_sensors["output_sensors"]:
+                if not asset_is_in_subtree(asset.id, output_sensor.generic_asset_id):
+                    raise ValueError(
+                        f"Report output sensor {output_sensor.id} must belong to asset"
+                        f" {asset.id} or one of its descendants."
+                    )
+            reporter.set_job_trigger("API")
+            result = reporter.compute(as_job=True, parameters=parameters)
+        except ValidationError as exc:
+            db.session.rollback()
+            return unprocessable_entity({"parameters": exc.messages})
+        except ValueError as exc:
+            db.session.rollback()
+            return unprocessable_entity(str(exc))
+        except Forbidden:
+            db.session.rollback()
+            raise
+
+        return request_accepted_for_processing(result["job_id"])
+
     @route("/<id>/schedules/trigger", methods=["POST"])
     @limit_triggers()
     @use_args(AssetTriggerSchemaV3(), location="args_and_json", as_kwargs=True)

flexmeasures/api/v3_0/tests/test_rate_limiting.py

@@ -474,6 +474,7 @@ def test_trigger_limited_views_are_registered():
     from flexmeasures.api.common.rate_limiting import TRIGGER_LIMITED_VIEWS
 
     assert {
+        "AssetAPI.trigger_report",
         "AssetAPI.trigger_schedule",
         "SensorAPI.trigger_schedule",
         "SensorAPI.trigger_forecast",