https://dagster.io/ logo
#ask-community
Title
# ask-community
t

Tomakhina Olha

02/07/2023, 8:42 PM
hI! Help, please. I try to add slack notification but something went wrong. It says:
Copy code
dagster._core.errors.DagsterInvalidDefinitionError: resource with key 'slack' required by op 'slack_op' was not provided. Please provide a <class 'dagster._core.definitions.resource_definition.ResourceDefinition'> to key 'slack', or change the required key to one of the following keys which points to an <class 'dagster._core.definitions.resource_definition.ResourceDefinition'>: ['io_manager']
I did this:
Copy code
@op(required_resource_keys={'slack'})
def slack_op(context, msg=':wave: hey there!'):
    context.resources.slack.chat_postMessage(channel=SLACK_PIPELINE_CHANNEL, text=msg)

@graph
def my_graph():
    ...
    slack_op(...)

def failure_message_fn(context: RunFailureSensorContext) -> str:
    return (
        f"Job {context.pipeline_run.pipeline_name} failed!"
        f"Error: {context.failure_event.message}"
    )


slack_on_run_failure = make_slack_on_run_failure_sensor(
    channel=SLACK_ALERTS_CHANNEL,
    slack_token=SLACK_TOKEN,
    text_fn=failure_message_fn
)

update_job = my_graph.to_job(name="update_job",
                             resource_defs={'slack': slack_resource},
                             config={'resources': {'slack': {'config': {'token': SLACK_TOKEN}}}})

defs = Definitions(
    schedules=[...],
    sensors=[slack_on_run_failure],
    jobs=[...],
    resources={"slack": slack_resource}
)
obviously I miss something…. but what ?
Can anybody help with this, please?
s

sean

02/08/2023, 1:14 PM
Hi Tomakhina, Thanks for the question. Here is a simple example of resource binding-- are you able to run
foo_job
here successfully?
Copy code
from dagster import Definitions
from dagster._core.definitions.decorators.graph_decorator import graph
from dagster._core.definitions.decorators.op_decorator import op
from dagster._core.definitions.resource_definition import ResourceDefinition

foo_resource = ResourceDefinition.hardcoded_resource("foo")

@op(required_resource_keys={"foo_resource"})
def foo_op(context):
    return context.resources.foo_resource

@graph
def foo_graph():
    foo_op()

foo_job = foo_graph.to_job(resource_defs={"foo_resource": foo_resource})

defs = Definitions(
    jobs=[foo_job],
    resources={"foo_resource": foo_resource},
)
e

Emil Sundman

02/08/2023, 1:30 PM
Yeah, you seem to specify the config as
{"slack": slack_resource}
but have never defined slack_resource anywhere in your code. Hence why the error message says that you haven't defined the resource with key "slack".
Without having worked with the the dagster slack integration myself, perhaps the row
context.resources.slack.chat_postMessage(channel=SLACK_PIPELINE_CHANNEL, text=msg)
is the very resource that should be defined as slack_resource? So change that to (and I could be wrong):
Copy code
slack_resource = context.resources.slack.chat_postMessage(channel=SLACK_PIPELINE_CHANNEL, text=msg)
t

Tomakhina Olha

02/25/2023, 12:15 PM
Thank you 🙂
59 Views