通知不断发布到slack频道

2024-03-28 09:39:41 发布

您现在位置:Python中文网/ 问答频道 /正文

我正在使用slack Webhook将消息发送到slack频道。 问题是,它每隔几分钟就会发布一次消息。 这就是我所做的

我在util文件夹下创建了一个简单的函数

def send_to_slack(text):
    conn_id = "https://hooks.slack.com/services/your/slack/URL"    
    task_slack_alert(text, url, is_error=False, args=None)

def task_slack_alert(msg, url, is_error=False, args=None):
     slack_msg = ":red_circle: Task Failed" if is_error else ":green_heart: Task Message"
     """*Task*: {task}  
        *Dag*: {dag} 
        *Execution Time*: {exec_ts}""".format(
                                          task=args["task"],
                                          dag=args["dag"],
                                          exec_ts=args["ts"],
        ) if args else ""
        message = {'text': + msg}
        response = requests.post(url=url, data=json.dumps(message))
        time.sleep(1)
        print(f"Slack response {response}")
        if response.status_code != 200:
            print(f"Error sending chat message. Got: {response.status_code}")

在我的dag(在另一个文件夹下)中,我调用该函数 dag将数据从oracle复制到snowflake db,这项工作没有松弛部分。 在我的dag中,我执行以下操作:

 x = {‘key1’: [‘value1’, ‘value 2’, … ‘value10]}
    
    send_to_slack('My test message from python')

default_args = {... 
'on_failure_callback': send_to_slack, }

with DAG(‘my_dag’,
         default_args=default_args,
         catchup=False) as dag:
    parallel = 4
    start = DummyOperator(task_id='start')
    tasks = []
    i = 0
    for s in x.keys():
        for t in x.get(s):
            task = OracleToSnowflakeOperator(                
                task_id=s + '_' + t,
                source_oracle_conn_id=source_oracle_conn_id,
                source_schema=schema,
                source_table=table,…
            )
            if i <= parallel:
                task.set_upstream(start)
            else:
                task.set_upstream(tasks[i - (parallel + 1)])
            i = i + 1
            tasks.append(task)

我知道如果我在同一个dag中定义函数,每次解析dag时都会调用它。 不是我的情况,怎么了? 谢谢


Tags: to函数textsendidurlsourcemessage
1条回答
网友
1楼 · 发布于 2024-03-28 09:39:41

您正在DAG文件中调用函数send_to_slack,这意味着它将在计划程序每次评估DAG时(每隔几分钟)运行

你应该:

  • 使用气流附带的slack operator,将其置于OracleToSnowflakeOperator的下游,并像对待任何其他操作员一样对待它
  • 编辑您的OracleToSnowflakeOperator,我假设它是一个自定义的,并将调用Slack的逻辑放在其中(使用Slack钩子)

基本上,您应该将对Slack的调用封装在自定义操作符中,或者使用提供的标准Slack操作符,不要将其放入DAG定义中

相关问题 更多 >