Sensors¶

@Sensor decorator¶

Creates a sensor from a function. The function receives a SensorEvaluationContext and returns a SensorResult, RunRequest, SkipReason, or None.

import rivers as rs

@rs.Sensor(job_name="my_job")
def file_sensor(context: rs.SensorEvaluationContext):
    # Check for new data
    if has_new_files():
        return rs.SensorResult(
            run_requests=[rs.RunRequest()],
            cursor=str(latest_timestamp()),
        )
    return rs.SkipReason("No new files")

Parameters:

Returns: Sensor

Sensor¶

A sensor that polls for external conditions and triggers runs. Can be created directly or via the @Sensor decorator.

sensor = rs.Sensor(
    name="my_sensor",
    job_name="my_job",
    minimum_interval="1m",
    default_status=rs.SensorStatus.Running,
    description="Checks for new data",
    tags={"team": "data"},
    asset_selection=["raw_data"],
)

Constructor:

Properties:

SensorEvaluationContext¶

Context passed to a sensor's evaluation function on each tick.

Properties:

SensorResult¶

Structured return type from a sensor evaluation function.

result = rs.SensorResult(
    run_requests=[rs.RunRequest(tags={"batch": "123"})],
    cursor="offset_42",
)

Parameters:

Providing both run_requests and skip_reason raises ValueError.

SensorStatus¶

Enum controlling whether a sensor is active.

SensorTickResult¶

Result of evaluating a sensor tick via CodeRepository.evaluate_sensor().

Properties:

Registration and evaluation¶

@rs.Sensor(job_name="my_job")
def my_sensor(context: rs.SensorEvaluationContext):
    return rs.SensorResult(
        run_requests=[rs.RunRequest()],
        cursor="new_cursor",
    )

@rs.Asset
def my_asset() -> int:
    return 42

repo = rs.CodeRepository(assets=[my_asset], sensors=[my_sensor])

# Evaluate with optional cursor from previous tick
result = repo.evaluate_sensor("my_sensor", cursor="old_cursor")
print(result.run_requests)  # [RunRequest(...)]
print(result.cursor)        # "new_cursor"

Cursor-based sensors¶

Sensors can track state across ticks using cursors. The cursor from the previous tick is passed into the next evaluation:

@rs.Sensor(job_name="process_events")
def event_sensor(context: rs.SensorEvaluationContext):
    last_id = int(context.cursor) if context.cursor else 0
    new_events = fetch_events_after(last_id)
    if new_events:
        return rs.SensorResult(
            run_requests=[rs.RunRequest(tags={"event": e.id}) for e in new_events],
            cursor=str(new_events[-1].id),
        )
    return rs.SkipReason("No new events")

Evaluation return types¶

The evaluation function can return:

• SensorResult — full result with run requests, skip reason, and cursor
• RunRequest — shorthand for a single run request
• SkipReason — skip with a reason
• None — skip silently

Run-status sensors¶

A run-status sensor calls its function once for each run of its code location that reaches a given status. Use it to alert on failed runs, or to act when a run succeeds.

import rivers as rs

@rs.Sensor.run_failure(default_status=rs.SensorStatus.Running)
def on_failure(context: rs.RunStatusSensorContext) -> None:
    for failure in context.step_failures:
        context.log.error("%s failed in run %s: %s", failure.asset_name, context.run.run_id, failure.error)

@rs.Sensor.run_status(
    rs.RunStatus.Success,
    monitored_jobs=["nightly"],
    default_status=rs.SensorStatus.Running,
)
def on_nightly_done(context: rs.RunStatusSensorContext) -> None:
    ...

repo = rs.CodeRepository(assets=[...], jobs=[...], sensors=[on_failure, on_nightly_done])

Sensor.run_failure(...) is Sensor.run_status(RunStatus.Failure, ...), and also works as a bare decorator: @rs.Sensor.run_failure.

The daemon evaluates only sensors with default_status=rs.SensorStatus.Running.

Sensor.run_status¶

The function takes a RunStatusSensorContext as its first parameter, and resources by parameter name. It must return None.

How runs are picked¶

• First tick: the sensor starts watching at that time. Runs that ended earlier are not reported. The tick is Skipped with "Watching for … runs from now on".
• Each later tick: the sensor calls the function once per new run, oldest first, for at most 5 runs. Further runs wait for the next tick.
• Errors: if the function raises for a run, the tick is Failed with run <run_id>: <error>, and the sensor moves on. That run is not tried again. If the whole tick times out, the tick fails and its runs are tried again on the next tick.
• Restarts: the cursor is stored on the tick, so a restarted daemon reports each run once.
• Late writes: each read starts 60 s back, so a run is reported when its record is written within 60 s of the end time it carries.
• Scope: only runs of the sensor's own code location. With monitored_jobs set, runs without a job, such as repo.materialize() runs, are skipped.

A run counts as finished when it gets its end time. The status on that run record decides whether the sensor reports it.

CodeRepository.evaluate_sensor() raises SensorDefinitionError for a run-status sensor: only the daemon evaluates it.

RunStatusSensorContext¶

StepFailure¶

RunStatus¶

RunRecord.status holds the same names as strings.