https://dagster.io/ logo
#announcements
Title
# announcements
e

Eduardo Santizo

04/05/2021, 5:09 PM
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:
Copy 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?
p

prha

04/05/2021, 5:15 PM
Hi Eduardo! Can you also include the error message and the import statement?
e

Eduardo Santizo

04/05/2021, 5:19 PM
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
p

prha

04/05/2021, 5:21 PM
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
e

Eduardo Santizo

04/05/2021, 5:51 PM
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.