RemediationAction

{
    "action_name": "slack_action",
    "category": "messaging",
    "description": "send slack message for TCSA Notification",
    "file_data": "import json\nfrom datetime import datetime\nfrom airflow.models import DAG\nfrom airflow.models import Variable\nfrom airflow.models.param import Param\nfrom airflow.operators.python import PythonOperator\nfrom airflow.providers.slack.operators.slack import SlackAPIPostOperator\nimport logging\nfrom Success import *\nfrom Failure import *\n\nmyParams = {\n    \"notification_obj\": {}\n}\n\ndef extract_action_params(**context) -> None:\n    return context[\"params\"][\"notification_obj\"][\"Name\"] + \" Source: \" + context[\"params\"][\"notification_obj\"][\"Source\"]\n\nwith DAG(\n\tdag_id='{{tcsa.dag_name}}',\n\tschedule_interval=None,\n\tstart_date=datetime(2023,8,16),\n    params=myParams,\n\tis_paused_upon_creation=False,\n\tcatchup=False,\n    on_success_callback =success,\n    on_failure_callback =failure\n) as dag:\n\n    # 1. extract action params into a dict\n    extract_action_params = PythonOperator(task_id='extract_action_params',\n        python_callable=extract_action_params,\n        dag=dag,\n        on_success_callback = success,\n        on_failure_callback =failure\n    )\n\n    # 2. send slack message with JIRA info and summary\n    task_post_slack = SlackAPIPostOperator(\n        task_id='task_post_slack',\n        slack_conn_id='{{tcsa.task_post_slack_conn_id}}',\n        text='{{ tcsa.slack_message }}' + ' ' + '[[ ti.xcom_pull(task_ids=[\"extract_action_params\"][0]) ]]',\n        channel='{{ tcsa.slack_channel }}',\n        on_success_callback = success,\n        on_failure_callback =failure\n    )\n\n    extract_action_params >> task_post_slack\n",
    "file_type": "PYTHON",
    "params": [
        {
            "parameterName": "slack_message",
            "option": "mandatory"
        },
        {
            "parameterName": "slack_channel",
            "option": "mandatory"
        }
    ]
}