Add unified job status endpoint GET /api/v3_0/jobs/<uuid>

documentation/api/change_log.rst

@@ -5,6 +5,11 @@ API change log
 
 .. note:: The FlexMeasures API follows its own versioning scheme. This is also reflected in the URL (e.g. `/api/v3_0`), allowing developers to upgrade at their own pace.
 
+v3.0-31 | 2026-04-28
+""""""""""""""""""""
+
+- Added a unified job status endpoint ``GET /api/v3_0/jobs/<uuid>`` that returns the current execution status and a human-readable result message for any background job (scheduling, forecasting, etc.) identified by its UUID.
+
 v3.0-30 | 2026-04-15
 """"""""""""""""""""
 - Added ``unit`` field to the `/sensors/<id>/schedules/<uuid>` (GET) endpoint for fetching a schedule, to get the schedule in a different unit still compatible to the sensor unit.

documentation/changelog.rst

@@ -11,6 +11,7 @@ New features
 * Added API and UI support for copying assets and their subtrees [see `PR #2017 <https://www.github.com/FlexMeasures/flexmeasures/pull/2017>`_ and `PR #2120 <https://www.github.com/FlexMeasures/flexmeasures/pull/2120>`_]
 * Improve UX after deleting a child asset through the UI [see `PR #2119 <https://www.github.com/FlexMeasures/flexmeasures/pull/2119>`_]
 * Improve source filtering in the sensor data GET endpoint by exposing the documented query parameters in Swagger and allowing filtering by the account linked to data sources [see `PR #2083 <https://www.github.com/FlexMeasures/flexmeasures/pull/2083>`_]
+* Added a unified job status endpoint ``GET /api/v3_0/jobs/<uuid>`` to retrieve the current execution status and result message for any background job [see `PR #2141 <https://www.github.com/FlexMeasures/flexmeasures/pull/2141>`_]
 * New ``GET /api/v3_0/sources`` endpoint to list accessible data sources and defined types, with ``only_latest=true`` by default to return only the most recent version per source [see `PR #2126 <https://www.github.com/FlexMeasures/flexmeasures/pull/2126>`_]
 
 Infrastructure / Support

flexmeasures/api/v3_0/__init__.py

@@ -29,6 +29,7 @@
     CopyAssetSchema,
 )
 from flexmeasures.api.v3_0.health import HealthAPI
+from flexmeasures.api.v3_0.jobs import JobAPI
 from flexmeasures.api.v3_0.public import ServicesAPI
 from flexmeasures.api.v3_0.deprecated import SensorEntityAddressAPI
 from flexmeasures.api.v3_0.sources import SourceAPI
@@ -57,6 +58,7 @@ def register_at(app: Flask):
     AssetAPI.register(app, route_prefix=v3_0_api_prefix)
     AssetTypesAPI.register(app, route_prefix=v3_0_api_prefix)
     HealthAPI.register(app, route_prefix=v3_0_api_prefix)
+    JobAPI.register(app, route_prefix=v3_0_api_prefix)
     ServicesAPI.register(app)
     SensorEntityAddressAPI.register(app, route_prefix=v3_0_api_prefix)
     SourceAPI.register(app, route_prefix=v3_0_api_prefix)

flexmeasures/api/v3_0/jobs.py

@@ -0,0 +1,255 @@
+from __future__ import annotations
+
+from datetime import datetime
+
+from flask import current_app
+from flask_classful import FlaskView, route
+from flask_json import as_json
+from flask_security import auth_required
+from redis.exceptions import ConnectionError as RedisConnectionError
+from rq.job import Job, JobStatus, NoSuchJobError
+from webargs.flaskparser import use_kwargs
+from marshmallow import fields
+
+from flexmeasures.api.common.utils.api_utils import job_status_description
+from flexmeasures.auth.policy import check_access
+from flexmeasures.data import db
+from flexmeasures.data.models.time_series import Sensor
+from flexmeasures.data.services.utils import get_asset_or_sensor_from_ref
+
+
+def _isoformat_or_none(dt: datetime | None) -> str | None:
+    """Return an ISO-8601 string for *dt*, or ``None`` when *dt* is absent."""
+    return dt.isoformat() if dt is not None else None
+
+
+def _failed_job_exc_info(job: Job) -> str | None:
+    """Return traceback text for failed jobs when RQ stored it."""
+    if not job.is_failed:
+        return None
+
+    latest_result = job.latest_result()
+    if latest_result is None:
+        return None
+
+    return latest_result.exc_string
+
+
+def _job_read_context(job: Job):
+    """Resolve the asset or sensor whose read access governs this job."""
+    asset_or_sensor_ref = job.meta.get("asset_or_sensor") or job.kwargs.get(
+        "asset_or_sensor"
+    )
+    if asset_or_sensor_ref is not None:
+        return get_asset_or_sensor_from_ref(asset_or_sensor_ref)
+
+    sensor_id = job.meta.get("sensor_id")
+    if sensor_id is None:
+        forecast_kwargs = job.meta.get("forecast_kwargs", {})
+        if isinstance(forecast_kwargs, dict):
+            sensor_id = forecast_kwargs.get("sensor_id")
+    if sensor_id is None:
+        sensor_id = job.kwargs.get("sensor_id")
+
+    if sensor_id is None:
+        return None
+
+    return db.session.get(Sensor, sensor_id)
+
+
+def _job_queue_unavailable_response():
+    return (
+        dict(
+            status="ERROR",
+            message="Job queues are currently unavailable.",
+        ),
+        503,
+    )
+
+
+class JobAPI(FlaskView):
+    """
+    Endpoint for querying the status of background jobs by UUID.
+    """
+
+    route_base = "/jobs"
+    trailing_slash = False
+
+    @route("/<uuid>", methods=["GET"])
+    @auth_required()
+    @use_kwargs({"job_id": fields.Str(data_key="uuid", required=True)}, location="path")
+    @as_json
+    def get_job_status(self, job_id: str, **kwargs):
+        """
+        .. :quickref: Jobs; Get the status of a background job
+
+        ---
+        get:
+          summary: Get the status of a background job
+          description: |
+            Look up a background job by its UUID and see whether it is
+            queued, running, finished, or failed.
+
+            The response includes a status message plus job metadata such
+            as the queue name, function name, timestamps, and the job
+            result when available.
+
+            Failed jobs also include traceback information when the worker
+            stored it with the job result.
+          security:
+            - ApiKeyAuth: []
+          parameters:
+            - in: path
+              name: uuid
+              required: true
+              description: UUID of the background job.
+              example: b3d26a8a-7a43-4a9f-93e1-fc2a869ea97b
+              schema:
+                type: string
+          responses:
+            200:
+              description: Job status retrieved successfully.
+              content:
+                application/json:
+                  schema:
+                    type: object
+                    properties:
+                      status:
+                        type: string
+                        enum:
+                          - QUEUED
+                          - STARTED
+                          - FINISHED
+                          - FAILED
+                          - DEFERRED
+                          - SCHEDULED
+                          - STOPPED
+                          - CANCELED
+                        description: Current status of the job.
+                      message:
+                        type: string
+                        description: Human-readable description of the job status.
+                      result:
+                        description: Return value of the job function, or null when not yet available.
+                        nullable: true
+                      func_name:
+                        type: string
+                        description: Fully-qualified name of the function executed by this job.
+                      origin:
+                        type: string
+                        description: Name of the queue the job was placed on.
+                      enqueued_at:
+                        type: string
+                        format: date-time
+                        nullable: true
+                        description: ISO-8601 timestamp of when the job was enqueued.
+                      started_at:
+                        type: string
+                        format: date-time
+                        nullable: true
+                        description: ISO-8601 timestamp of when the job started executing.
+                      ended_at:
+                        type: string
+                        format: date-time
+                        nullable: true
+                        description: ISO-8601 timestamp of when the job finished executing.
+                      exc_info:
+                        type: string
+                        nullable: true
+                        description: Traceback information for failed jobs, or null otherwise.
+                  examples:
+                    queued:
+                      summary: Queued job
+                      value:
+                        status: QUEUED
+                        message: "Scheduling job waiting to be processed."
+                        result: null
+                        func_name: "flexmeasures.data.services.scheduling.create_schedule"
+                        origin: scheduling
+                        enqueued_at: "2026-04-28T10:00:00+00:00"
+                        started_at: null
+                        ended_at: null
+                        exc_info: null
+                    finished:
+                      summary: Finished job
+                      value:
+                        status: FINISHED
+                        message: "Scheduling job has finished."
+                        result: null
+                        func_name: "flexmeasures.data.services.scheduling.create_schedule"
+                        origin: scheduling
+                        enqueued_at: "2026-04-28T10:00:00+00:00"
+                        started_at: "2026-04-28T10:00:01+00:00"
+                        ended_at: "2026-04-28T10:00:05+00:00"
+                        exc_info: null
+                    failed:
+                      summary: Failed job
+                      value:
+                        status: FAILED
+                        message: "Scheduling job failed with ValueError: ..."
+                        result: null
+                        func_name: "flexmeasures.data.services.scheduling.create_schedule"
+                        origin: scheduling
+                        enqueued_at: "2026-04-28T10:00:00+00:00"
+                        started_at: "2026-04-28T10:00:01+00:00"
+                        ended_at: "2026-04-28T10:00:02+00:00"
+                        exc_info: "Traceback (most recent call last): ..."
+            404:
+              description: NOT_FOUND
+            401:
+              description: UNAUTHORIZED
+            403:
+              description: INVALID_SENDER
+            503:
+              description: SERVICE_UNAVAILABLE
+          tags:
+            - Jobs
+        """
+        connection = current_app.redis_connection
+
+        try:
+            connection.ping()
+            job = Job.fetch(job_id, connection=connection)
+            read_context = _job_read_context(job)
+            if read_context is not None:
+                check_access(read_context, "read")
+        except NoSuchJobError:
+            return (
+                dict(
+                    status="ERROR",
+                    message=f"Job {job_id} not found.",
+                ),
+                404,
+            )
+        except RedisConnectionError:
+            return _job_queue_unavailable_response()
+
+        try:
+            job_status = job.get_status()
+            status_name = (
+                job_status.name
+                if isinstance(job_status, JobStatus)
+                else str(job_status).upper()
+            )
+
+            # job.return_value is None when the job has not finished successfully
+            result = job.return_value()
+        except RedisConnectionError:
+            return _job_queue_unavailable_response()
+        except Exception:  # noqa: BLE001
+            result = None
+
+        return (
+            dict(
+                status=status_name,
+                message=job_status_description(job),
+                result=result,
+                func_name=job.func_name,
+                origin=job.origin,
+                enqueued_at=_isoformat_or_none(job.enqueued_at),
+                started_at=_isoformat_or_none(job.started_at),
+                ended_at=_isoformat_or_none(job.ended_at),
+                exc_info=_failed_job_exc_info(job),
+            ),
+            200,
+        )