Is there a way during a

run_on_failure_sensor

to take a look at the failed pipelines last run? We had a question come up to where we want to do alert monitoring but in some cases we want to only alert if the prior run also failed and not just the individual run.

Hey Scott. This is possible, you can query the instance directly from within the sensor. Here's a snippet of untested code:

@run_failure_sensor
def check_multiple_runs_sensor(context: RunFailureSensorContext):
    latest_runs = context.instance.get_runs(RunsFilter(job_name="my_job"), limit=2)
    for run in latest_runs:
        if run.status == DagsterRunStatus.FAILURE:
             # your action here