u110
8/15/2019 - 5:56 AM

test_SlackAPIPostOperator.py

from airflow.operators.slack_operator import SlackAPIPostOperator


SLACK_TOKEN = "xoxb-XXXX-YYYY-ZZZ"

# refs.
# https://api.slack.com/docs/message-attachments#action_fields

if __name__ == "__main__":
        op = SlackAPIPostOperator(
            task_id="slack_post_task",
            token=SLACK_TOKEN,
            text="post message test",
            channel='#bot',
            username='airflow-bot',
            attachments=[{
                    "text": "DANGER!",
                    "color": "danger"
            }]
        )
        op.execute()