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

@@ -7,6 +7,7 @@ API change log
 
 v3.0-32 | August 11, 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.
 - API endpoints are now rate-limited. A request which exceeds a limit is answered with a ``429 (Too Many Requests)`` status code and a ``Retry-After`` header stating how many seconds to wait. Responses also carry ``X-RateLimit-*`` headers, describing the limit that applied, how much of it is left, and when it resets. A stricter limit applies to ``POST /assets/<id>/schedules/trigger``, ``POST /sensors/<id>/schedules/trigger`` and ``POST /sensors/<id>/forecasts/trigger`` than to other endpoints; the health endpoints are exempt. Per-account overrides are set by assigning the account a plan (a ``Plan`` database row), rather than through an account attribute.
 - Introduced the ``inflexible-consumption`` and ``inflexible-production`` flex-context fields, which make explicit how the sign of each inflexible device's power data should be read: positive values denote consumption resp. production. Each entry is a sensor reference (``{"sensor": <id>}``), optionally with source filters (``source-types``, ``exclude-source-types``, ``sources``, ``source-account``). Deprecated the ``inflexible-device-sensors`` field (a list of bare sensor IDs, whose sign convention is read from each sensor's ``consumption_is_positive`` attribute); it remains supported, but cannot be combined with the new fields in one flex-context.
 - Added a ``role`` query parameter to ``GET /api/v3_0/accounts`` for filtering accessible organisations by account role.

documentation/changelog.rst

@@ -11,6 +11,8 @@ 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>`_ and `issue #2426 <https://github.com/FlexMeasures/flexmeasures/issues/2426>`_]
+
 Infrastructure / Support
 -------------------------
 

documentation/cli/change_log.rst

@@ -7,6 +7,7 @@ FlexMeasures CLI Changelog
 since v1.0.0 | August 11, 2026
 =================================
 
+* Add ``flexmeasures add report --as-job`` and the ``reporting`` worker queue for asynchronous one-off reports.
 * ``flexmeasures add user --roles`` now parses comma-separated role lists and accepts repeated ``--roles`` options.
 * ``flexmeasures db upgrade`` now runs ``VACUUM ANALYZE`` after upgrading (refreshing the query planner's statistics); opt out with ``--no-vacuum``.
 * Add ``flexmeasures add plan``, ``flexmeasures show plans`` and ``flexmeasures edit plan``, to manage the rate limits and quotas which apply to the accounts on a plan.

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,6 +25,7 @@
     add_beliefs_to_db_and_enqueue_forecasting_jobs,
 )
 from flexmeasures.data.models.generic_assets import GenericAsset
+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,
@@ -570,23 +571,6 @@ def _determine_copy_name(
     return f"{source_name} (Copy {max_index + 1})"
 
 
-def _asset_is_in_subtree(root_asset_id: int, candidate_asset_id: int) -> bool:
-    """Return True if candidate_asset_id is root or a descendant of root_asset_id."""
-    current_asset_id = candidate_asset_id
-    visited: set[int] = set()
-
-    while current_asset_id is not None and current_asset_id not in visited:
-        if current_asset_id == root_asset_id:
-            return True
-        visited.add(current_asset_id)
-        current_asset = db.session.get(GenericAsset, current_asset_id)
-        if current_asset is None:
-            return False
-        current_asset_id = current_asset.parent_asset_id
-
-    return False
-
-
 def copy_asset(
     asset: GenericAsset,
     account=None,
@@ -634,7 +618,7 @@ def copy_asset(
             target_account_id = int(account.id)
             target_parent_asset_id = int(parent_asset.id)
 
-        if target_parent_asset_id is not None and _asset_is_in_subtree(
+        if target_parent_asset_id is not None and asset_is_in_subtree(
             root_asset_id=asset.id,
             candidate_asset_id=target_parent_asset_id,
         ):

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

@@ -49,13 +49,21 @@
 from flexmeasures.data.models.user import Account
 from flexmeasures.data.models.audit_log import AssetAuditLog
 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,
@@ -1571,6 +1579,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,
+                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")
+            returns = 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(returns["job_id"])
+
     @route("/<id>/schedules/trigger", methods=["POST"])
     @limit_triggers()
     @use_args(AssetTriggerSchemaV3(), location="args_and_json", as_kwargs=True)