ferruzzi commented on code in PR #24554:
URL: https://github.com/apache/airflow/pull/24554#discussion_r901921811
##########
tests/system/providers/amazon/aws/example_sqs.py:
##########
@@ -66,6 +66,20 @@ def delete_queue(queue_url):
)
# [END howto_sensor_sqs]
+ # [START howto_sensor_sqs_batch]
+ # batch multiple messages from SQS.
+ # each SQS poll can retrieve no more than 10 messages
+ # due to requirements by AWS SQS
+ read_from_queue_in_batch = SqsBatchSensor(
+ task_id='read_from_queue_in_batch',
+ sqs_queue=create_queue,
+ # get maximum 10 messages each poll
+ max_messages=10,
+ # perform 3 polls before returning results
Review Comment:
```suggestion
# Combine 3 polls before returning results
```
##########
airflow/providers/amazon/aws/sensors/sqs.py:
##########
@@ -215,3 +226,65 @@ def __init__(self, *args, **kwargs):
stacklevel=2,
)
super().__init__(*args, **kwargs)
+
+
+class SqsBatchSensor(SqsSensor):
+ """
+ Get messages from an Amazon SQS queue in batches and then delete the
retrieved messages from the queue.
+ If deletion of messages fails an AirflowException is thrown. Otherwise,
all messages
+ are pushed through XCom with the key ``messages``.
+ The total number of messages retrieved at maxium will be equal to the
number of messages retrived for each
+ SQS's API call multiplies with total number of call. Each SQS
receive_message can get a max 10 messages.
+ This sensor is identical to SQSSensor, except the fact that SQSSensor
performs one and only one SQS call
+ per poke, while SQSBatchSensor performs multiple SQS API calls per poke.
+ .. seealso::
+ For more information on how to use this sensor, take a look at the
guide:
+ :ref:`howto/sensor:SqsBatchSensor`
+ :param batch: The number of time the sensor will call the SQS to receive
messages (default: 1)
Review Comment:
I'm not crazy about the variable name `batch` here. The name doesn't
really indicate what it is for. Perhaps `batches` or `num_batches` or
`num_pokes` might be better?
##########
airflow/providers/amazon/aws/sensors/sqs.py:
##########
@@ -215,3 +226,65 @@ def __init__(self, *args, **kwargs):
stacklevel=2,
)
super().__init__(*args, **kwargs)
+
+
+class SqsBatchSensor(SqsSensor):
+ """
+ Get messages from an Amazon SQS queue in batches and then delete the
retrieved messages from the queue.
+ If deletion of messages fails an AirflowException is thrown. Otherwise,
all messages
+ are pushed through XCom with the key ``messages``.
+ The total number of messages retrieved at maxium will be equal to the
number of messages retrived for each
+ SQS's API call multiplies with total number of call. Each SQS
receive_message can get a max 10 messages.
+ This sensor is identical to SQSSensor, except the fact that SQSSensor
performs one and only one SQS call
+ per poke, while SQSBatchSensor performs multiple SQS API calls per poke.
+ .. seealso::
+ For more information on how to use this sensor, take a look at the
guide:
+ :ref:`howto/sensor:SqsBatchSensor`
+ :param batch: The number of time the sensor will call the SQS to receive
messages (default: 1)
Review Comment:
Suggested rewording of the docstring. It's still a bit confusing, but may
be a bit better?
```suggestion
Get messages from an Amazon SQS queue in batches and then delete the
retrieved messages from the queue.
If deletion of messages fails an AirflowException is thrown. Otherwise,
all messages
are pushed through XCom with the key ``messages``.
This sensor is identical to SqsSensor, except that SqsSensor performs
one and only one SQS call
per poke, which limits the result to a maximum of 10 messages, while
SqsBatchSensor performs multiple SQS API calls per poke and combines the
results into one list.
.. seealso::
For more information on how to use this sensor, take a look at the
guide:
:ref:`howto/sensor:SqsBatchSensor`
:param batch: The number of times the sensor will call the SQS API to
receive messages (default: 1)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
