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):
        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 ) : 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``
Check for message on subscribed queue and write to xcom the message with key messages