I ve written a unit test for a sensor that requires a resour dagster #ask-community

I've written a unit test for a sensor that require...

Philippe Laflamme

08/09/2023, 5:13 PM

I've written a unit test for a sensor that requires a resource (more or less) like so:

Copy code

def test_the_sensor():
  ctx = build_sensor_context(resources={"my_api": mock_api})
  my_sensor(ctx)

with

my_sensor

defined like so

Copy code

@sensor(required_resource_keys={"my_api"})
def my_sensor(ctx):
  my_api = ctx.resources.my_api
  ...

Running

pytest

results in this:

Copy code

TypeError: my_sensor() got an unexpected keyword argument 'my_api'

If I add

my_api: SomeApi

my_sensor

arguments, the test passes, but then at runtime in the real dagster instance, I get this:

Copy code

TypeError: my_sensor() missing 1 required positional argument: 'my_api'

I'm running 1.13.13, is this a known issue or am I doing something wrong?

🤖 1

yuhan

08/09/2023, 6:01 PM

Copy code

from dagster import sensor, build_sensor_context, ConfigurableResource


class MyResource(ConfigurableResource):
    value: str


@sensor()
def my_sensor(my_api: MyResource):
    print(my_api.value)


def test_the_sensor():
    ctx = build_sensor_context(resources={"my_api": MyResource(value="foo")})
    my_sensor(ctx)

this should work

yuhan

08/09/2023, 6:02 PM

https://docs.dagster.io/concepts/partitions-schedules-sensors/sensors#using-resources-in-sensors sensor with resources works with Pythonic Resources which are passed in as function inputs

Philippe Laflamme

08/09/2023, 6:03 PM

Right, when I make my sensor function look like this:

Copy code

def my_sensor(ctx, my_api: MyResource):
  pass

the test works, but in the real runtime, I get this:

Copy code

TypeError: my_sensor() missing 1 required positional argument: 'my_api'

yuhan

08/09/2023, 6:04 PM

ah got it. let me try repro

🙏 1

yuhan

08/09/2023, 6:31 PM

Copy code

from dagster import (
    sensor,
    build_sensor_context,
    ConfigurableResource,
    asset,
    Definitions,
    AssetSelection,
)


class MyResource(ConfigurableResource):
    value: str


@asset
def my_asset():
    return


@sensor(asset_selection=AssetSelection.all())
def my_sensor(context, my_api: MyResource):
    print(my_api.value)


def test_the_sensor():
    ctx = build_sensor_context(resources={"my_api": MyResource(value="foo")})
    my_sensor(ctx)


defs = Definitions(
    assets=[my_asset], sensors=[my_sensor], resources={"my_api": MyResource(value="foo")}
)

this is the full code snippet that i tested and it works in a real dagster instance

Philippe Laflamme

08/09/2023, 6:35 PM

Ok, presumably I made a mistake somewhere. I'll dig around to see if I can find what I did. Thanks for looking into this. On a related note, this made my sensor fail every single tick which is only visible in the UI. Is there an existing mechanism to allow detecting sensors that fail repeatedly like this (and potentially do something like send a slack message or whatever)?

chris

08/09/2023, 7:45 PM

in cloud, there’s alerting based on schedule / sensor state change (IE, when your sensor starts failing, you’ll get an alert, but won’t get spammed with alerts for every single failed sensor tick)

Philippe Laflamme

08/09/2023, 7:48 PM

Got it, yeah that makes sense. Is this feature based on things we can build from the OSS version?

chris

08/09/2023, 7:58 PM

In this particular instance, I don’t think there’s an analogue functionality in OSS - you could maybe ingest the logs from sensors / schedules and build out some alerting on that

👍 1

4 Views

Open in Slack

Previous Next