Skip to main content

Scenarios and jobs

A scenario is a project's scheduled or on-demand sequence of steps; a job run is one execution recorded by the platform. Scenarios are managed through Scenario handles; runs, pipeline triggers and logs through client.jobs. Scenario endpoints need org-admin permission. See Scheduler for the concepts.

Scenarios​

project.list_scenarios()
sc = project.get_scenario("daily_refresh")    # by name or display name
sc = project.get_scenario(42)                 # or by id

sc = project.create_scenario(
    "daily_refresh",
    display_name="Daily refresh",
    trigger_type="manual",
    steps=[{"type": "run_scenario", "config": {"scenario_id": 41}}],
)

create_scenario(name, *, display_name=None, description="", trigger_type="manual", steps=None, is_active=False, **extra) sends any extra keyword as a field of the scenario. sc.update(**fields) changes fields (for example steps=[...] or is_active=True), sc.delete() removes it, and sc.get_settings() returns the cached record.

Run and wait​

run = sc.run()                                # {run_id, status, steps}
final = client.jobs.wait(run["run_id"], poll_interval=10, timeout=3600)

final = sc.run_and_wait(poll_interval=10.0, timeout=3600.0)   # the same, in one call
print(final["status"])

sc.get_last_runs(limit=5)                     # recent runs from the scenario record

run_and_wait triggers the scenario and polls the run until it reaches a terminal status (below). When the timeout passes first, it returns the last-seen run, still running; the run continues on the server either way.

Triggers​

A scenario can carry several triggers. Adding the first stored trigger supersedes the scenario's legacy single trigger_type field.

trig = sc.add_trigger("temporal", config={"cron": "0 6 * * *"}, name="every morning")
sc.add_trigger("sql_query", config={
    "sql": "SELECT 1 FROM staging.orders_raw WHERE customer_id IS NULL HAVING COUNT(*) > 100",
})
for t in sc.list_triggers():
    print(t["trigger_id"], t["trigger_type"], t["active"])

sc.update_trigger(trig["trigger_id"], "temporal", config={"cron": "0 7 * * *"})  # full replace
sc.delete_trigger(trig["trigger_id"])

Trigger types include temporal (cron), sql_query, dataset_modified and after_scenario; config is specific to the type.

Reporters​

Reporters send a notification when a run starts or ends.

rep = sc.add_reporter(
    "mail", "notify ops",
    send_on="end",                 # start | end
    trigger="failure",             # always | success | failure | custom
    channel_config={"to": "ops@your-domain.com"},
    template_body="${scenarioName}: ${outcome}",
)
sc.test_reporter(rep["id"])        # sends a sample success notification now
sc.list_reporters()
sc.update_reporter(rep["id"], active=False)
sc.delete_reporter(rep["id"])

kind is mail, slack, teams or webhook; channel_config carries that channel's settings. With trigger="custom", the reporter evaluates run_condition_formula.

Jobs​

client.jobs is a JobsAPI object.

Runs​

runs = client.jobs.list_runs(status="failed", limit=50, offset=0)   # newest first
run = client.jobs.get_run(run_id)          # full detail, including log_output
client.jobs.abort_run(run_id)

final = client.jobs.wait(run_id, poll_interval=5.0, timeout=1800.0,
                         raise_on_timeout=True)

list_runs accepts job_type and status filters; limit must be 1–200 (the server's page size cap; anything else raises ValueError, 0.1.8+) and offset pages further.

wait polls get_run until the status is terminal. On timeout it returns the last-seen detail, or raises TimeoutError with raise_on_timeout=True (0.1.8+).

Terminal statuses​

A run is finished when its status is one of:

success · failed · aborted · error · done · skipped

skipped is what the still-open child steps of an aborted run become (0.1.8+ adds it to the terminal set). The set is available as JobsAPI.TERMINAL_STATUSES.

Pipeline triggers​

client.jobs.trigger("dbt_run", select="stg_sales+")
# {"status": "triggered", "job_name": "dbt_run", "triggered_by": ...}

run = client.jobs.trigger_and_wait("dbt_run", select="stg_sales+", timeout=3600)
print(run["status"], run.get("duration_sec"))

print(client.jobs.logs("dbt_run", lines=100))    # the tail of the job's log

trigger(job_name, *, project_id=None, **opts) starts extract_incremental, extract_organization or dbt_run, passing options such as select, mode, tables, organization, start_date, end_date or schedule_at. It returns as soon as the job is started; the response carries no run id. With schedule_at, the response is "scheduled" with the scheduled time.

trigger_and_wait (0.1.8+) closes that gap: it notes the existing pipeline runs, triggers, finds the new run of the job started by the same user, and polls it until it is no longer running, queued or pending. It raises TimeoutError when no run appears within start_timeout (default 120 s), and refuses schedule_at. Both need org admin.

list_pipeline_runs(status=None, limit=20) and get_pipeline_run(run_id) (0.1.8+) read the pipeline-run rows directly.

Verified against honeyframeapi 0.1.8.