Skip to main content

:::note Generated reference Generated from honeyframeapi 0.1.8 — do not edit by hand. :::

honeyframeapi.scenario

Scenario handle — run, schedule and monitor a project's scenarios.

Replaces a hand-rolled scenario runner script:

sc = project.get_scenario("daily_pipeline")
run = sc.run()                     # {run_id, status, steps}
final = sc.wait(run["run_id"])     # polls /api/jobs/{run_id} to terminal state
print(final["status"], final.get("steps_failed"))

Scenarios live under /api/pipeline/scenarios and require org-admin permission on the platform.

Steps are a plain field on the scenario, set via the generic update(**fields):

# chain another scenario in the same project (cycle-guarded at runtime)
sc.update(steps=[{"type": "run_scenario", "config": {"scenario_id": 42}}])

Triggers: a scenario can carry MANY triggers (cron + sql + after_scenario …). Use the trigger wrappers — the first one you add supersedes the legacy single-trigger_type columns (which the scheduler then ignores):

sc.add_trigger("temporal", config={"cron": "0 6 * * *"})
sc.add_trigger("sql_query", config={"sql": "SELECT 1 FROM staging.orders_raw "
                                           "WHERE customer_id IS NULL HAVING count(*) > 100"})
sc.list_triggers()

The legacy single-trigger field still works for a one-shot scenario (sc.update(trigger_type="manual")) but is inert once any stored trigger row exists.

class Scenario​

Scenario.name​

property

name: str

Scenario.reload()​

reload() -> Scenario

Scenario.get_settings()​

get_settings() -> dict

Scenario.update()​

update(**fields) -> dict

Scenario.delete()​

delete() -> dict

Scenario.run()​

run() -> dict

Trigger the scenario; returns {run_id, status, steps}.

Scenario.run_and_wait()​

run_and_wait(*, poll_interval: float = 10.0, timeout: float = 3600.0) -> dict

Trigger and block until the run reaches a terminal state.

Scenario.get_last_runs()​

get_last_runs(*, limit: int = 5) -> list[dict]

Recent runs of this scenario (from the scenario detail's recent_runs).

Scenario.list_triggers()​

list_triggers() -> list[dict]

Stored trigger rows for this scenario ({trigger_id, name, trigger_type, config, active, last_fired_at, created_at}).

Scenario.add_trigger()​

add_trigger(trigger_type: str, *, config: Optional[dict] = None, name: str = '', active: bool = True) -> dict

Add a trigger. trigger_type ∈ temporal (cron) | sql_query | dataset_modified | after_scenario …; config is type-specific (e.g. {"cron": "0 6 * * *"} or {"sql": "…"}). Returns {trigger_id, ok}.

Scenario.update_trigger()​

update_trigger(trigger_id: int, trigger_type: str, *, config: Optional[dict] = None, name: str = '', active: bool = True) -> dict

Replace a trigger's definition (full-body PUT).

Scenario.delete_trigger()​

delete_trigger(trigger_id: int) -> dict

Scenario.list_reporters()​

list_reporters() -> list[dict]

This scenario's reporters (email / webhook / … notification rules).

Scenario.add_reporter()​

add_reporter(kind: str, name: str, *, send_on: str = 'end', trigger: str = 'always', run_condition_formula: Optional[str] = None, channel_config: Optional[dict] = None, template_subject: Optional[str] = None, template_body: str = '${scenarioName}: ${outcome}', active: bool = True) -> dict

Add a reporter. kind ∈ mail | slack | teams | webhook; send_on ∈ start | end; trigger ∈ always | success | failure | custom (custom reads run_condition_formula). channel_config carries the channel's settings (e.g. {"to": "ops@x.com"} for mail, a webhook/Teams/Slack URL for the others). Returns {id}.

Scenario.update_reporter()​

update_reporter(reporter_id: int, **fields) -> dict

Patch a reporter (only the fields you pass change).

Scenario.delete_reporter()​

delete_reporter(reporter_id: int) -> dict

Scenario.test_reporter()​

test_reporter(reporter_id: int) -> dict

Fire a synthetic SUCCESS notification now (bypasses the predicate) so you can see the real message. Returns the send result.

Verified against honeyframeapi 0.1.8.