diff --git a/documentation/api/change_log.rst b/documentation/api/change_log.rst index 66dde555d0..bb98ba154a 100644 --- a/documentation/api/change_log.rst +++ b/documentation/api/change_log.rst @@ -5,6 +5,10 @@ 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-34 | September 2, 2026 +""""""""""""""""""""""""""" +- Added ``POST /api/v3_0/assets//automations//trigger``, to run one automation now, once, on top of its recurring runs. The response is the standard job response, extended with ``n_jobs``: how many jobs the run queued. An on-demand run does not affect the automation's recurrence, and inactive automations can be triggered, too. Triggering requires the same permission as writing data under the asset, and falls under the stricter rate limit that the other triggering endpoints share. + v3.0-33 | September 1, 2026 """"""""""""""""""""""""""" - Added ``GET /api/v3_0/assets//automations`` and ``GET /api/v3_0/assets//automations/`` 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. diff --git a/documentation/changelog.rst b/documentation/changelog.rst index 02c1e5eb58..5adbca1655 100644 --- a/documentation/changelog.rst +++ b/documentation/changelog.rst @@ -17,6 +17,7 @@ v1.1.0 | September XX, 2026 New features ------------- +* A single automation can now be run on demand, from the CLI (``flexmeasures jobs run-automation``), the API (``POST /assets//automations//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 `_] * 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 `_] Infrastructure / Support diff --git a/documentation/cli/change_log.rst b/documentation/cli/change_log.rst index c3a1769089..e77643ded9 100644 --- a/documentation/cli/change_log.rst +++ b/documentation/cli/change_log.rst @@ -14,6 +14,7 @@ since v1.0.0 | August 11, 2026 * Add ``flexmeasures delete secret`` to remove an encrypted secret from an account or asset. * Add ``flexmeasures add automation``, ``flexmeasures edit automation`` and ``flexmeasures delete automation`` to manage automations (recurring tasks on an asset; for now, computing forecasts). Each automation carries its own IANA timezone (``--timezone``), in which its cron expression is interpreted. * Add ``flexmeasures jobs run-automations`` to queue jobs for all automations that are due to run this minute from standard five-field cron expressions. Run this command once per minute. It makes at most one queueing attempt per automation per minute, including when an attempt fails after partially queueing jobs. Runs missed while the runner was down are caught up once, with several missed forecast runs coalesced into the latest useful forecast, and a run at a skipped or repeated daylight-saving-time hour happens exactly once. +* Add ``flexmeasures jobs run-automation --automation `` to queue the jobs for a single run of one automation, now, on top of its recurring runs. This leaves the automation's cursor alone, so its next recurring run still happens as scheduled, and inactive automations can be run this way, too. * ``flexmeasures delete sensor`` now warns which automations read from or write to a sensor before it is deleted, as an automation refers to its sensors by ID and would fail on its next run. * ``flexmeasures show data-sources`` now shows the account a data source belongs to, and lists the sensors holding data recorded by a single source with ``--show-sensors``. diff --git a/documentation/features/automations.rst b/documentation/features/automations.rst index 7a441226e7..ae8d1a74cb 100644 --- a/documentation/features/automations.rst +++ b/documentation/features/automations.rst @@ -56,6 +56,24 @@ If the process crashes, or queueing fails after creating some jobs, that run is The jobs record how they were created, which is shown on the asset's status page (UI), where recent jobs are listed. +Running one automation on demand +-------------------------------- + +Besides its recurring runs, a single automation can be run now, once. +This 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. + +.. code-block:: bash + + flexmeasures jobs run-automation --automation 4 + +The same is available in the API, as `[POST] /assets/(id)/automations/(automation_id)/trigger <../api/v3_0.html#post--api-v3_0-assets-id-automations-automation_id-trigger>`_, and in the UI, as the *Run now* button on the asset's *Automations* page. + +The automation runs with the parameters it was created with, and the jobs it queues are recorded as its jobs, just like the jobs of a recurring run. +An on-demand run does not affect the automation's recurrence: its cursor (see :ref:`automation_cursor`) stays where it was, so the next recurring run still happens as scheduled, and a run missed while the runner was down is still caught up. +Inactive automations can be run this way, too, which is how you can try one out before activating it. + +Unlike a recurring run, an on-demand run is not protected against being started twice: asking for two runs in a row queues two runs. + Viewing automations ------------------- diff --git a/flexmeasures/api/v3_0/assets.py b/flexmeasures/api/v3_0/assets.py index 33fe92af26..a51cd37cec 100644 --- a/flexmeasures/api/v3_0/assets.py +++ b/flexmeasures/api/v3_0/assets.py @@ -54,6 +54,7 @@ describe_cronstr, get_automation_job_stats, resolve_automation_sensors, + run_automation, ) from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType from flexmeasures.data.queries.generic_assets import ( @@ -1575,6 +1576,114 @@ def get_automation(self, id: int, automation_id: int, asset: GenericAsset): automation_data["redis_connection_err"] = redis_connection_err return automation_data, 200 + @route("//automations//trigger", methods=["POST"]) + @limit_triggers() + @use_kwargs( + { + "asset": AssetIdField(data_key="id"), + "automation_id": fields.Int(), + }, + location="path", + ) + # Running an automation writes data under the asset, which is what create-children means here. + # The sensors it writes to were checked against the same permission when the automation was created, + # and its output scope is checked again on each run (see validate_forecast_output_scope). + @permission_required_for_context("create-children", ctx_arg_name="asset") + @as_json + def trigger_automation(self, id: int, automation_id: int, asset: GenericAsset): + """ + .. :quickref: Assets; Trigger a single run of an automation. + + --- + post: + summary: Trigger a single run of an automation. + description: | + Run one automation now, once, in addition to its recurring runs. + This 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. + + The automation runs with the parameters it was created with, + and the jobs it queues are recorded as its jobs, just like the jobs of a recurring run. + + An on-demand run does not affect the automation's recurrence: its cursor stays where it was, + so the next recurring run still happens as scheduled, and a missed run is still caught up. + Inactive automations can be triggered, too, which is how you can try one out before activating it. + security: + - ApiKeyAuth: [] + parameters: + - in: path + name: id + required: true + description: ID of the asset the automation is defined on. + schema: + type: integer + - in: path + name: automation_id + required: true + description: ID of the automation to run. + schema: + type: integer + responses: + 202: + description: PROCESSING + content: + application/json: + examples: + triggered: + summary: Automation run accepted + description: | + The automation queued its jobs, which will be picked up by a worker. + The `job` field holds the Universally Unique Identifier (UUID) of the job to follow, + and `n_jobs` says how many jobs the run queued in total. + value: + status: ACCEPTED + job: "364bfd06-c1fa-430b-8d25-8f5a547651fb" + job-url: "/api/v3_0/jobs/364bfd06-c1fa-430b-8d25-8f5a547651fb" + n_jobs: 2 + message: "Request has been accepted for processing." + 401: + description: UNAUTHORIZED + 403: + description: INVALID_SENDER + 404: + description: NOT_FOUND + 422: + description: UNPROCESSABLE_ENTITY + tags: + - Assets + """ + automation = db.session.get(Automation, automation_id) + if automation is None or automation.asset_id != asset.id: + return { + "message": f"Asset {asset.id} has no automation with id {automation_id}." + }, 404 + try: + returns = run_automation(automation) + except (NotImplementedError, ValueError, ValidationError) as e: + db.session.rollback() + return unprocessable_entity( + e.messages if isinstance(e, ValidationError) else str(e) + ) + job_id = (returns or {}).get("job_id") + if job_id is None: + db.session.rollback() + current_app.logger.error( + "Automation %s ran on demand, but reported no job: %r", + automation.id, + returns, + ) + return unprocessable_entity( + f"Automation {automation.id} did not queue any job." + ) + AssetAuditLog.add_record( + asset, + f"Triggered a run of automation '{automation.name}' ({automation.id}).", + ) + db.session.commit() + response, status_code = request_accepted_for_processing(job_id) + response["n_jobs"] = returns.get("n_jobs") + return response, status_code + @route("//jobs", methods=["GET"]) @use_kwargs( {"asset": AssetIdField(data_key="id")}, diff --git a/flexmeasures/api/v3_0/tests/test_automations_api.py b/flexmeasures/api/v3_0/tests/test_automations_api.py index 39652038b8..e7ef018575 100644 --- a/flexmeasures/api/v3_0/tests/test_automations_api.py +++ b/flexmeasures/api/v3_0/tests/test_automations_api.py @@ -172,3 +172,141 @@ def test_get_nonexistent_automation( url_for("AssetAPI:get_automation", id=battery.id, automation_id=9999), ) assert response.status_code == 404 + + +@pytest.mark.parametrize( + "requesting_user, expected_status_code", + [ + (None, 401), # not logged in + ("test_prosumer_user@seita.nl", 202), # same account + ("test_dummy_user_3@seita.nl", 403), # different account + ], + indirect=["requesting_user"], +) +def test_trigger_automation_auth( + app, + add_battery_assets, + add_automations, + requesting_user, + expected_status_code, + mocker, +): + battery = add_battery_assets["Test battery"] + automation = add_automations[0] + run_automation = mocker.patch( + "flexmeasures.api.v3_0.assets.run_automation", + return_value={"job_id": "364bfd06-c1fa-430b-8d25-8f5a547651fb", "n_jobs": 2}, + ) + with app.test_client() as client: + response = client.post( + url_for( + "AssetAPI:trigger_automation", + id=battery.id, + automation_id=automation.id, + ), + ) + assert response.status_code == expected_status_code + if expected_status_code == 202: + assert run_automation.call_args.args[0].id == automation.id + else: + run_automation.assert_not_called() + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_trigger_automation( + app, + db, + add_battery_assets, + add_automations, + requesting_user, + mocker, +): + """Triggering a run reports the queued job, and leaves the automation's recurrence alone.""" + battery = add_battery_assets["Test battery"] + automation = add_automations[1] # inactive automations can be triggered, too. + cursor_before = automation.cursor + mocker.patch( + "flexmeasures.api.v3_0.assets.run_automation", + return_value={"job_id": "364bfd06-c1fa-430b-8d25-8f5a547651fb", "n_jobs": 2}, + ) + with app.test_client() as client: + response = client.post( + url_for( + "AssetAPI:trigger_automation", + id=battery.id, + automation_id=automation.id, + ), + ) + assert response.status_code == 202 + assert response.json["status"] == "ACCEPTED" + assert response.json["job"] == "364bfd06-c1fa-430b-8d25-8f5a547651fb" + assert response.json["n_jobs"] == 2 + db.session.expire_all() + assert automation.cursor == cursor_before + assert automation.active is False + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +def test_trigger_automation_that_cannot_run( + app, + add_battery_assets, + add_automations, + requesting_user, + mocker, +): + """A run which cannot be set up is reported as such, rather than as a queued job.""" + battery = add_battery_assets["Test battery"] + automation = add_automations[0] + mocker.patch( + "flexmeasures.api.v3_0.assets.run_automation", + side_effect=ValueError( + "Forecast automation output sensor 3 must belong to asset 1 or one of its descendants." + ), + ) + with app.test_client() as client: + response = client.post( + url_for( + "AssetAPI:trigger_automation", + id=battery.id, + automation_id=automation.id, + ), + ) + assert response.status_code == 422 + assert "must belong to asset" in str(response.json["message"]) + + +@pytest.mark.parametrize( + "requesting_user", ["test_prosumer_user@seita.nl"], indirect=True +) +@pytest.mark.parametrize("via_other_asset", [True, False]) +def test_trigger_unknown_automation( + app, + add_battery_assets, + add_automations, + requesting_user, + via_other_asset, + mocker, +): + """Triggering an automation the asset does not have returns 404, without running anything.""" + run_automation = mocker.patch("flexmeasures.api.v3_0.assets.run_automation") + if via_other_asset: + # an existing automation, requested through an asset it does not belong to. + asset = add_battery_assets["Test small battery"] + automation_id = add_automations[0].id + else: + asset = add_battery_assets["Test battery"] + automation_id = 9999 + with app.test_client() as client: + response = client.post( + url_for( + "AssetAPI:trigger_automation", + id=asset.id, + automation_id=automation_id, + ), + ) + assert response.status_code == 404 + run_automation.assert_not_called() diff --git a/flexmeasures/cli/jobs.py b/flexmeasures/cli/jobs.py index 541a1e83a6..f4833fe554 100644 --- a/flexmeasures/cli/jobs.py +++ b/flexmeasures/cli/jobs.py @@ -26,12 +26,16 @@ ScheduledJobRegistry, StartedJobRegistry, ) +from marshmallow import ValidationError from sqlalchemy.orm import configure_mappers from tabulate import tabulate import pandas as pd from flexmeasures.data import db +from flexmeasures.data.models.audit_log import AssetAuditLog +from flexmeasures.data.models.automations import Automation from flexmeasures.data.schemas import AssetIdField, SensorIdField +from flexmeasures.data.schemas.automations import AutomationIdField from flexmeasures.data.services.automations import ( claim_due_automation, floor_to_minute, @@ -127,6 +131,59 @@ def run_automations(): raise click.exceptions.Exit(1) +@fm_jobs.command("run-automation") +@with_appcontext +@click.option( + "--automation", + "automation", + type=AutomationIdField(), + required=True, + help="ID of the automation to run.", +) +def run_one_automation(automation: Automation): + """ + Queue the jobs for a single run of one automation, now. + + \b + flexmeasures jobs run-automation --automation 4 + + Use this 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. + The automation runs with the parameters it was created with, + and the jobs it queues are recorded as its jobs, just like the jobs of a recurring run. + + This does not affect the automation's recurrence: its cursor stays where it was, + so the next recurring run still happens as scheduled (see `flexmeasures jobs run-automations`). + Inactive automations can be run this way, too, which is how you can try one out before activating it. + """ + try: + returns = run_automation(automation) + except (NotImplementedError, ValueError, ValidationError) as e: + db.session.rollback() + click.secho( + f"Automation {automation.id} ('{automation.name}') failed to queue jobs: {e}", + **MsgStyle.ERROR, + ) + raise click.Abort() + if not returns or returns.get("job_id") is None: + db.session.rollback() + click.secho( + f"Automation {automation.id} ('{automation.name}') did not queue any job.", + **MsgStyle.ERROR, + ) + raise click.Abort() + n_jobs = returns.get("n_jobs") + AssetAuditLog.add_record( + automation.asset, + f"Triggered a run of automation '{automation.name}' ({automation.id}) via CLI.", + ) + db.session.commit() + click.secho( + f"Automation {automation.id} ('{automation.name}') queued {n_jobs} forecasting job(s) for asset {automation.asset_id}.", + **MsgStyle.SUCCESS, + ) + + @fm_jobs.command("stats") @with_appcontext @click.option( diff --git a/flexmeasures/cli/tests/test_automations.py b/flexmeasures/cli/tests/test_automations.py index 14e26569d4..511a570990 100644 --- a/flexmeasures/cli/tests/test_automations.py +++ b/flexmeasures/cli/tests/test_automations.py @@ -916,3 +916,129 @@ def test_run_automation_revalidates_output_scope( assert run_result.exit_code == 1 assert "must belong to asset" in run_result.output assert app.queues["forecasting"].count == 0 + + +def test_run_one_automation_on_demand( + app, fresh_db, setup_dummy_data, clean_redis, freeze_server_now +): + """An on-demand run queues the automation's jobs without claiming its next scheduled run.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.jobs import run_automations, run_one_automation + + freeze_server_now(datetime(2026, 1, 15, 8, 58, 30, tzinfo=timezone.utc)) + runner = app.test_cli_runner() + cli_input = { + "asset": 1, + "name": "Daily forecasts", + "cron": "0 10 * * *", # not due at the time we trigger it by hand + "timezone": "UTC", + "sensor": setup_dummy_data[0], + } + add_result = runner.invoke(add_automation, to_flags(cli_input)) + assert add_result.exit_code == 0, add_result.output + automation = fresh_db.session.scalars(select(Automation)).one() + cursor_before = automation.cursor + + run_result = runner.invoke(run_one_automation, ["--automation", str(automation.id)]) + + assert run_result.exit_code == 0, run_result.output + assert "queued" in run_result.output + n_jobs = app.queues["forecasting"].count + assert n_jobs > 0 + # the jobs are recorded as this automation's jobs, just like those of a recurring run. + assert all( + job.meta["trigger"]["origin"] == "automation" + and job.meta["trigger"]["automation_id"] == automation.id + for job in app.queues["forecasting"].jobs + ) + assert fresh_db.session.execute( + select(AssetAuditLog).filter( + AssetAuditLog.event.like("Triggered a run of automation%") + ) + ).scalar_one_or_none() + + # the cursor stayed put, so the scheduled run still happens. + fresh_db.session.expire_all() + assert automation.cursor == cursor_before + freeze_server_now(datetime(2026, 1, 15, 10, 0, tzinfo=timezone.utc)) + scheduled_result = runner.invoke(run_automations) + assert scheduled_result.exit_code == 0, scheduled_result.output + assert scheduled_result.output.count("queued") == 1, scheduled_result.output + assert app.queues["forecasting"].count > n_jobs + + +def test_run_one_automation_runs_inactive_automation( + app, fresh_db, setup_dummy_data, clean_redis +): + """An inactive automation can be tried out on demand, while it stays out of the recurring runs.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.jobs import run_automations, run_one_automation + + runner = app.test_cli_runner() + add_result = runner.invoke( + add_automation, + to_flags( + { + "asset": 1, + "name": "Not active yet", + "cron": "* * * * *", + "sensor": setup_dummy_data[0], + } + ) + + ["--inactive"], + ) + assert add_result.exit_code == 0, add_result.output + automation = fresh_db.session.scalars(select(Automation)).one() + + scheduled_result = runner.invoke(run_automations) + assert "No automations due" in scheduled_result.output, scheduled_result.output + + run_result = runner.invoke(run_one_automation, ["--automation", str(automation.id)]) + + assert run_result.exit_code == 0, run_result.output + assert "queued" in run_result.output + assert app.queues["forecasting"].count > 0 + + +def test_run_one_automation_without_queued_job_is_an_error( + app, fresh_db, setup_dummy_data, clean_redis, mocker +): + """A run which reports no job is an error, rather than a success which recorded nothing.""" + from flexmeasures.cli.data_add import add_automation + from flexmeasures.cli.jobs import run_one_automation + + runner = app.test_cli_runner() + add_result = runner.invoke( + add_automation, + to_flags( + { + "asset": 1, + "name": "Reports no job", + "cron": "0 6 * * *", + "sensor": setup_dummy_data[0], + } + ), + ) + assert add_result.exit_code == 0, add_result.output + automation = fresh_db.session.scalars(select(Automation)).one() + mocker.patch("flexmeasures.cli.jobs.run_automation", return_value=None) + + result = runner.invoke(run_one_automation, ["--automation", str(automation.id)]) + + assert result.exit_code == 1, result.output + assert "did not queue any job" in result.output + assert not fresh_db.session.execute( + select(AssetAuditLog).filter( + AssetAuditLog.event.like("Triggered a run of automation%") + ) + ).scalar_one_or_none() + + +def test_run_one_automation_reports_unknown_automation(app, fresh_db, clean_redis): + from flexmeasures.cli.jobs import run_one_automation + + result = app.test_cli_runner().invoke(run_one_automation, ["--automation", "9999"]) + + assert result.exit_code == 2, result.output + assert "No automation found with id 9999" in result.output + assert app.queues["forecasting"].count == 0 diff --git a/flexmeasures/ui/static/openapi-specs.json b/flexmeasures/ui/static/openapi-specs.json index 255b097cb7..4f156b8c06 100644 --- a/flexmeasures/ui/static/openapi-specs.json +++ b/flexmeasures/ui/static/openapi-specs.json @@ -4471,6 +4471,77 @@ ] } }, + "/api/v3_0/assets/{id}/automations/{automation_id}/trigger": { + "post": { + "summary": "Trigger a single run of an automation.", + "description": "Run one automation now, once, in addition to its recurring runs.\nThis is useful to try out a new automation, to re-run one after fixing what made it fail,\nor to refresh its results after late input data arrived.\n\nThe automation runs with the parameters it was created with,\nand the jobs it queues are recorded as its jobs, just like the jobs of a recurring run.\n\nAn on-demand run does not affect the automation's recurrence: its cursor stays where it was,\nso the next recurring run still happens as scheduled, and a missed run is still caught up.\nInactive automations can be triggered, too, which is how you can try one out before activating it.\n", + "security": [ + { + "ApiKeyAuth": [] + } + ], + "parameters": [ + { + "in": "path", + "name": "id", + "required": true, + "description": "ID of the asset the automation is defined on.", + "schema": { + "type": "integer" + } + }, + { + "in": "path", + "name": "automation_id", + "required": true, + "description": "ID of the automation to run.", + "schema": { + "type": "integer" + } + } + ], + "responses": { + "202": { + "description": "PROCESSING", + "content": { + "application/json": { + "examples": { + "triggered": { + "summary": "Automation run accepted", + "description": "The automation queued its jobs, which will be picked up by a worker.\nThe `job` field holds the Universally Unique Identifier (UUID) of the job to follow,\nand `n_jobs` says how many jobs the run queued in total.\n", + "value": { + "status": "ACCEPTED", + "job": "364bfd06-c1fa-430b-8d25-8f5a547651fb", + "job-url": "/api/v3_0/jobs/364bfd06-c1fa-430b-8d25-8f5a547651fb", + "n_jobs": 2, + "message": "Request has been accepted for processing." + } + } + } + } + } + }, + "401": { + "description": "UNAUTHORIZED" + }, + "403": { + "description": "INVALID_SENDER" + }, + "404": { + "description": "NOT_FOUND" + }, + "422": { + "description": "UNPROCESSABLE_ENTITY" + }, + "429": { + "description": "TOO_MANY_REQUESTS - You called the API, or triggered computation, more often than your rate limits allow. Triggering endpoints share a stricter limit than the rest of the API. Wait for as long as the Retry-After header says, then try again." + } + }, + "tags": [ + "Assets" + ] + } + }, "/api/v3_0/assets/{id}/schedules/trigger": { "post": { "summary": "Trigger scheduling job for any number of devices", diff --git a/flexmeasures/ui/templates/assets/asset_automations.html b/flexmeasures/ui/templates/assets/asset_automations.html index 2646a9bbfc..455b09760e 100644 --- a/flexmeasures/ui/templates/assets/asset_automations.html +++ b/flexmeasures/ui/templates/assets/asset_automations.html @@ -45,6 +45,7 @@

+
@@ -53,6 +54,8 @@

diff --git a/flexmeasures/ui/views/assets/views.py b/flexmeasures/ui/views/assets/views.py index ef2b068b82..0ccb92233b 100644 --- a/flexmeasures/ui/views/assets/views.py +++ b/flexmeasures/ui/views/assets/views.py @@ -256,6 +256,7 @@ def automations(self, id: str): return render_flexmeasures_template( "assets/asset_automations.html", asset=asset, + user_can_create_children=user_can_create_children(asset), current_page="Automations", )