DagsterDocs

Slack (dagster-slack)


This library provides an integration with Slack, to support posting messages in your company’s Slack workspace.


Presently, it provides a thin wrapper on the Slack client API chat.postMessage.


To use this integration, you’ll first need to create a Slack App for it.

  1. Create App: Go to https://api.slack.com/apps and click “Create New App”:

  2. Install App: After creating an app, on the left-hand side of the app configuration, click “Bot Users”, and then create a bot user. Then, click “Install App” on the left hand side, and finally “Install App to Workspace”.

  3. Bot Token: Once finished, this will create a new bot token for your bot/workspace:

Copy this bot token and put it somewhere safe; see Safely Storing Credentials for more on this topic.

dagster_slack.slack_resource ResourceDefinition[source]

This resource is for connecting to Slack.

The resource object is a slack.WebClient.

By configuring this Slack resource, you can post messages to Slack from any Dagster solid:

Examples:

import os

from dagster import solid, execute_pipeline, ModeDefinition
from dagster_slack import slack_resource


@solid(required_resource_keys={'slack'})
def slack_solid(context):
    context.resources.slack.chat_postMessage(channel='#noise', text=':wave: hey there!')

@pipeline(
    mode_defs=[ModeDefinition(resource_defs={'slack': slack_resource})],
)
def slack_pipeline():
    slack_solid()

execute_pipeline(
    slack_pipeline, {'resources': {'slack': {'config': {'token': os.getenv('SLACK_TOKEN')}}}}
)
dagster_slack.slack_on_failure HookDefinition[source]

Create a hook on step failure events that will message the given Slack channel.

Parameters
  • channel (str) – The channel to send the message to (e.g. “#my_channel”)

  • message_fn (Optional(Callable[[HookContext], str])) – Function which takes in the HookContext outputs the message you want to send.

  • dagit_base_url – (Optional[str]): The base url of your Dagit instance. Specify this to allow messages to include deeplinks to the specific pipeline run that triggered the hook.

Examples

@slack_on_failure("#foo", dagit_base_url="http://localhost:3000")
@pipeline(...)
def my_pipeline():
    pass
def my_message_fn(context: HookContext) -> str:
    return "Solid {solid_name} failed!".format(
        solid_name=context.solid
    )

@solid
def a_solid(context):
    pass

@pipeline(...)
def my_pipeline():
    a_solid.with_hooks(hook_defs={slack_on_failure("#foo", my_message_fn)})
dagster_slack.slack_on_success HookDefinition[source]

Create a hook on step success events that will message the given Slack channel.

Parameters
  • channel (str) – The channel to send the message to (e.g. “#my_channel”)

  • message_fn (Optional(Callable[[HookContext], str])) – Function which takes in the HookContext outputs the message you want to send.

  • dagit_base_url – (Optional[str]): The base url of your Dagit instance. Specify this to allow messages to include deeplinks to the specific pipeline run that triggered the hook.

Examples

@slack_on_success("#foo", dagit_base_url="http://localhost:3000")
@pipeline(...)
def my_pipeline():
    pass
def my_message_fn(context: HookContext) -> str:
    return "Solid {solid_name} worked!".format(
        solid_name=context.solid
    )

@solid
def a_solid(context):
    pass

@pipeline(...)
def my_pipeline():
    a_solid.with_hooks(hook_defs={slack_on_success("#foo", my_message_fn)})
dagster_slack.make_slack_on_pipeline_failure_sensor(channel, slack_token, text_fn=<function _default_failure_message>, blocks_fn=None, pipeline_selection=None, name=None, dagit_base_url=None)[source]

Create a sensor on pipeline failures that will message the given Slack channel.

Parameters
  • channel (str) – The channel to send the message to (e.g. “#my_channel”)

  • slack_token (str) – The slack token. Tokens are typically either user tokens or bot tokens. More in the Slack API documentation here: https://api.slack.com/docs/token-types

  • text_fn (Optional(Callable[[PipelineFailureSensorContext], str])) – Function which takes in the PipelineFailureSensorContext and outputs the message you want to send. Defaults to a text message that contains error message, pipeline name, and run ID. The usage of the text_fn changes depending on whether you’re using blocks_fn. If you are using blocks_fn, this is used as a fallback string to display in notifications. If you aren’t, this is the main body text of the message. It can be formatted as plain text, or with mrkdwn. See more details in https://api.slack.com/methods/chat.postMessage#text_usage

  • blocks_fn (Callable[[PipelineFailureSensorContext], List[Dict]]) – Function which takes in the PipelineFailureSensorContext and outputs the message blocks you want to send. See information about Blocks in https://api.slack.com/reference/block-kit/blocks

  • pipeline_selection (Optional[List[str]]) – Names of the pipelines that will be monitored by this failure sensor. Defaults to None, which means the alert will be sent when any pipeline in the repository fails.

  • name – (Optional[str]): The name of the sensor. Defaults to “slack_on_pipeline_failure”.

  • dagit_base_url – (Optional[str]): The base url of your Dagit instance. Specify this to allow messages to include deeplinks to the failed pipeline run.

Examples

slack_on_pipeline_failure = make_slack_on_pipeline_failure_sensor(
    "#my_channel",
    os.getenv("MY_SLACK_TOKEN")
)

@repository
def my_repo():
    return [my_pipeline + slack_on_pipeline_failure]
def my_message_fn(context: PipelineFailureSensorContext) -> str:
    return "Pipeline {pipeline_name} failed! Error: {error}".format(
        pipeline_name=context.pipeline_run.pipeline_name,
        error=context.failure_event.message,
    )

slack_on_pipeline_failure = make_slack_on_pipeline_failure_sensor(
    channel="#my_channel",
    slack_token=os.getenv("MY_SLACK_TOKEN"),
    message_fn=my_message_fn,
    dagit_base_url="http://mycoolsite.com",
)