airflow.providers.amazon.aws.sensors.sqs

Reads and then deletes the message from SQS queue.

Module Contents

Classes

SqsSensor

Get messages from an Amazon SQS queue and then delete the messages from the queue.
class airflow.providers.amazon.aws.sensors.sqs.SqsSensor(*, sqs_queue, aws_conn_id='aws_default', max_messages=5, num_batches=1, wait_time_seconds=1, visibility_timeout=None, message_filtering=None, message_filtering_match_values=None, message_filtering_config=None, delete_message_on_reception=True, deferrable=conf.getboolean('operators', 'default_deferrable', fallback=False), **kwargs)[source]

Bases: airflow.sensors.base.BaseSensorOperator

Get messages from an Amazon SQS queue and then delete the messages from the queue.

If deletion of messages fails, an AirflowException is thrown. Otherwise, the messages are pushed through XCom with the key messages.

By default,the sensor performs one and only one SQS call per poke, which limits the result to a maximum of 10 messages. However, the total number of SQS API calls per poke can be controlled by num_batches param.

See also

For more information on how to use this sensor, take a look at the guide: Read messages from an Amazon SQS queue

Parameters

  • aws_conn_id (str) – AWS connection id

  • sqs_queue – The SQS queue url (templated)

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

  • num_batches (int) – The number of times the sensor will call the SQS API to receive messages (default: 1)

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

  • visibility_timeout (int | None) – Visibility timeout, a period of time during which Amazon SQS prevents other consumers from receiving and processing the message.

  • message_filtering (typing_extensions.Literal[literal, jsonpath] | None) – 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.

  • delete_message_on_reception (bool) – Default to True, the messages are deleted from the queue as soon as being consumed. Otherwise, the messages remain in the queue after consumption and should be deleted manually.

  • deferrable (bool) – If True, the sensor will operate in deferrable more. This mode requires aiobotocore module to be installed. (default: False, but can be overridden in config file by setting default_deferrable to True)

template_fields: Sequence[str] = ('sqs_queue', 'max_messages', 'message_filtering_config')[source]
execute(context)[source]

Derive when creating an operator.

Context is the same dictionary used as when rendering jinja templates.

Refer to get_template_context for more context.

execute_complete(context, event=None)[source]
poll_sqs(sqs_conn)[source]

Poll SQS queue to retrieve messages.

Parameters

sqs_conn (airflow.providers.amazon.aws.hooks.base_aws.BaseAwsConnection) – SQS connection

Returns

A list of messages retrieved from SQS

Return type

Collection

poke(context)[source]

Check subscribed queue for messages and write them to xcom with the messages key.

Parameters

context (airflow.utils.context.Context) – the context object

Returns

True if message is available or False

get_hook()[source]

Create and return an SqsHook.

hook()[source]

Was this entry helpful?