airflow/contrib/sensors/aws_sqs_sensor.py
def poke(self, context):
        """
        Check for message on subscribed queue and write to xcom the message with key ``messages``

        :param context: the context object
        :type context: dict
        :return: ``True`` if message is available or ``False``
        """

        sqs_hook = SQSHook(aws_conn_id=self.aws_conn_id)
        sqs_conn = sqs_hook.get_conn()

        self.log.info('SQSSensor checking for message on queue: %s', self.sqs_queue)

        messages = sqs_conn.receive_message(QueueUrl=self.sqs_queue,
                                            MaxNumberOfMessages=self.max_messages,
                                            WaitTimeSeconds=self.wait_time_seconds)

        self.log.info("reveived message %s", str(messages))

        if 'Messages' in messages and len(messages['Messages']) > 0:

            entries = [{'Id': message['MessageId'], 'ReceiptHandle': message['ReceiptHandle']}
                       for message in messages['Messages']]

            result = sqs_conn.delete_message_batch(QueueUrl=self.sqs_queue,
                                                   Entries=entries)

            if 'Successful' in result:
                context['ti'].xcom_push(key='messages', value=messages)
                return True
            else:
                raise AirflowException(
                    'Delete SQS Messages failed ' + str(result) + ' for messages ' + str(messages))

        return False
def poke(self, context):
        """
        Check for message on subscribed queue and write to xcom the message with key ``messages``

        :param context: the context object
        :type context: dict
        :return: ``True`` if message is available or ``False``
        """

        sqs_hook = SQSHook(aws_conn_id=self.aws_conn_id)
        sqs_conn = sqs_hook.get_conn()

        self.log.info('SQSSensor checking for message on queue: %s', self.sqs_queue)

        messages = sqs_conn.receive_message(QueueUrl=self.sqs_queue,
                                            MaxNumberOfMessages=self.max_messages,
                                            WaitTimeSeconds=self.wait_time_seconds)

        self.log.info("reveived message %s", str(messages))

        if 'Messages' in messages and len(messages['Messages']) > 0:

            entries = [{'Id': message['MessageId'], 'ReceiptHandle': message['ReceiptHandle']}
                       for message in messages['Messages']]

            result = sqs_conn.delete_message_batch(QueueUrl=self.sqs_queue,
                                                   Entries=entries)

            if 'Successful' in result:
                context['ti'].xcom_push(key='messages', value=messages)
                return True
            else:
                raise AirflowException(
                    'Delete SQS Messages failed ' + str(result) + ' for messages ' + str(messages))

        return False
Check for message on subscribed queue and write to xcom the message with key ``messages``

        :param context: the context object
        :type context: dict
        :return: ``True`` if message is available or ``False``