airflow.providers.amazon.aws.sensors.sqs

Reads and then deletes the message from SQS queue

Module Contents

class airflow.providers.amazon.aws.sensors.sqs.SQSSensor(*, sqs_queue, aws_conn_id: str = 'aws_default', max_messages: int = 5, wait_time_seconds: int = 1, visibility_timeout: Optional[int] = None, message_filtering: Optional[Literal['literal', 'jsonpath']] = None, message_filtering_match_values: Any = None, message_filtering_config: Any = None, **kwargs)[source]

Bases: airflow.sensors.base.BaseSensorOperator

Get messages from an SQS queue and then deletes the message from the SQS queue. If deletion of messages fails an AirflowException is thrown otherwise, the message is pushed through XCom with the key messages.

Parameters

  • aws_conn_id (str) -- AWS connection id

  • sqs_queue (str) -- The SQS queue url (templated)

  • max_messages (int) -- The maximum number of messages to retrieve for each poke (templated)

  • wait_time_seconds (int) -- The time in seconds to wait for receiving messages (default: 1 second)

  • visibility_timeout (Optional[Int]) -- Visibility timeout, a period of time during which Amazon SQS prevents other consumers from receiving and processing the message.

  • message_filtering (Optional[Literal["literal", "jsonpath"]]) -- Specified how received messages should be filtered. Supported options are: None (no filtering, default), 'literal' (message Body literal match) or 'jsonpath' (message Body filtered using a JSONPath expression). You may add further methods by overriding the relevant class methods.

  • message_filtering_match_values (Any) -- Optional value/s for the message filter to match on. For example, with literal matching, if a message body matches any of the specified values then it is included. For JSONPath matching, the result of the JSONPath expression is used and may match any of the specified values.

  • message_filtering_config (Any) -- Additional configuration to pass to the message filter. For example with JSONPath filtering you can pass a JSONPath expression string here, such as 'foo[*].baz'. Messages with a Body which does not match are ignored.

template_fields = ['sqs_queue', 'max_messages', 'message_filtering_config'][source]
poke(self, context)[source]

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

Parameters

context (dict) -- the context object

Returns

True if message is available or False

get_hook(self)[source]

Create and return an SQSHook

filter_messages(self, messages)[source]
filter_messages_literal(self, messages)[source]
filter_messages_jsonpath(self, messages)[source]

Was this entry helpful?