The cloud-native orchestrator for the whole development lifecycle, with integrated lineage and observability.

dagster

Hey. I was trying to implement the "error detecting sensor" in the dagster docs, but I'm getting a "PipelineRunsFilter" error during its execution. Any idea why this is happening?

Here is the code:
```def CAMP_restart_sensor(context):
    runs = context.instance.get_runs(
        filters=PipelineRunsFilter(

            # Pipeline to check using the sensor
            pipeline_name="CAMP_pipeline",
            statuses=[PipelineRunStatus.FAILURE],
        ),
    )
    for run in runs:
        # Use the id of the failed run as run_key to avoid duplicate alerts.
        yield RunRequest(...)```
When I looked into it, it seems like the PipelineRunsFilter is a GraphQL related function. This also made me curious. How exactly can I integrate GraphQL queries inside python? Is there a specific way or is it like any other GraphQL implementation?

Hi Eduardo!  Can you also include the error message and the import statement?

Thanks! I suspected that an additional import was needed. What is the package that I need to include?

The error is basically: PipelineRunsFilter is not defined. So yes, missing import from my part

I think probably you’ll need something like this: `from dagster.core.storage.pipeline_run import PipelineRunStatus, PipelineRunsFilter`

I’ll update the sensor example documentation to include the import statement

We should also consider exporting those from the main dagster package so you can do something like `from dagster import PipelineRunStatus, PipelineRunsFilter`, but need to think about that a bit more

Thank you Phil! This was exactly what I needed. And yes, my first instinct was to use `from dagster import PipelineRunStatus, PipelineRunsFilter`  but Python refused to recognize this, so I came to the conclusion that maybe the import statement was not needed.