Tomakhina Olha
02/07/2023, 8:42 PMdagster._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:
@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 ?Tomakhina Olha
02/08/2023, 10:23 AMsean
02/08/2023, 1:14 PMfoo_job
here successfully?
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},
)
Emil Sundman
02/08/2023, 1:30 PM{"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".Emil Sundman
02/08/2023, 1:33 PMcontext.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):
slack_resource = context.resources.slack.chat_postMessage(channel=SLACK_PIPELINE_CHANNEL, text=msg)
Tomakhina Olha
02/25/2023, 12:15 PM