Apache Airflow |如何在使用SSHOperator时检查命令的响应？

from airflow.contrib.operators.ssh_operator import SSHOperator
from airflow.utils.decorators import apply_defaults

class CustomSSHOperator(SSHOperator):
    @apply_defaults
    def __init__(self, *args, **kwargs):
        super(CustomSSHOperator, self).__init__(*args, **kwargs)

    def execute(self, context):
        # 调用父类的execute方法执行SSH命令
        super(CustomSSHOperator, self).execute(context)

        # 检查命令的响应
        response = self.ssh_hook.remote_cmd_output
        if response:
            # 响应不为空，执行相应的操作
            # 例如，可以将响应保存到文件中
            with open('/path/to/response.txt', 'w') as f:
                f.write(response)
        else:
            # 响应为空，执行相应的操作
            # 例如，可以抛出异常或记录日志
            raise Exception("Command response is empty.")

task = CustomSSHOperator(
    task_id='ssh_task',
    ssh_conn_id='my_ssh_conn',
    command='echo "Hello, Airflow!"',
    dag=dag
)