Convert event-replay back to batch API

Author: jlhood

pipelines/event-replay-pipeline/src/replay_messages.py

@@ -20,20 +20,24 @@ def handler(event, context):
  LOG.info('No records in event')
  return
 
- for record in event['Records']:
- request = _to_request(record)
- LOG.debug('Replaying event to SQS queue: queue URL=%s, request=%s', DESTINATION_SQS_QUEUE_URL, request)
- response = QUEUE.send_message(**request)
- LOG.debug('SQS response: %s', response)
+ entries = [_to_request_record(str(i), r) for i, r in enumerate(event['Records'])]
+ LOG.debug('Replaying events to SQS queue: queue URL=%s, entries=%s', DESTINATION_SQS_QUEUE_URL, entries)
+ response = QUEUE.send_messages(Entries=entries)
+ LOG.debug('SQS response: %s', response)
+ if response.get('Failed'):
+ raise RuntimeError('Failed to send messages to destination queue: queue URL={}, failures={}'.format(
+ DESTINATION_SQS_QUEUE_URL, response['Failed']
+ ))
 
 
-def _to_request(record):
+def _to_request_record(id, record):
  # SQS event lowercases message attribute property names so have to capitalize them again before sending to SQS
  message_attributes = {attr_name: {_capitalize(attr_prop_name): attr_prop_value
  for (attr_prop_name, attr_prop_value) in attr_props.items()
  if attr_prop_value}
  for (attr_name, attr_props) in record['messageAttributes'].items()}
  return {
+ 'Id': id,
  'MessageBody': record['body'],
  'MessageAttributes': message_attributes
  }

pipelines/event-replay-pipeline/template.yml

@@ -90,7 +90,7 @@ Resources:
  Effect: Allow
  Action:
  - sqs:GetQueueUrl
- - sqs:SendMessage
+ - sqs:SendMessage*
  Resource:
  - !Sub arn:${AWS::Partition}:sqs:${AWS::Region}:${AWS::AccountId}:${DestinationSQSQueueName}
 

pipelines/event-replay-pipeline/test/unit/test_replay_messages.py

@@ -15,19 +15,17 @@ def replay_messages(mocker):
  mocker.patch.object(boto3, 'resource')
  boto3.client.return_value = mock_sqs_client
  boto3.resource.return_value = mock_sqs_client
- mock_queue = MagicMock()
- mock_sqs_client.Topic.return_value = mock_queue
  import replay_messages
  return replay_messages
 
 
 def test_handler_no_records(mocker, replay_messages):
  replay_messages.handler({}, None)
- replay_messages.QUEUE.send_message.assert_not_called()
+ replay_messages.QUEUE.send_messages.assert_not_called()
 
 
 def test_handler_success(mocker, replay_messages):
- replay_messages.QUEUE.send_message.return_value = {}
+ replay_messages.QUEUE.send_messages.return_value = {}
 
  event = {
  'Records': [
@@ -69,29 +67,50 @@ def test_handler_success(mocker, replay_messages):
  }
  replay_messages.handler(event, None)
 
- replay_messages.QUEUE.send_message.assert_any_call(
- MessageBody='foo',
- MessageAttributes={
- 'fooAttr1': {
- 'DataType': 'String',
- 'StringValue': 'fooAttr1Value'
- },
- 'fooAttr2': {
- 'DataType': 'String',
- 'StringValue': 'fooAttr2Value'
+ expected_entries = [
+ {
+ 'Id': '0',
+ 'MessageBody': 'foo',
+ 'MessageAttributes': {
+ 'fooAttr1': {
+ 'DataType': 'String',
+ 'StringValue': 'fooAttr1Value'
+ },
+ 'fooAttr2': {
+ 'DataType': 'String',
+ 'StringValue': 'fooAttr2Value'
+ }
  }
- }
- )
- replay_messages.QUEUE.send_message.assert_any_call(
- MessageBody='bar',
- MessageAttributes={
- 'barAttr1': {
- 'DataType': 'String',
- 'StringValue': 'barAttr1Value'
- },
- 'barAttr2': {
- 'DataType': 'String',
- 'StringValue': 'barAttr2Value'
+ },
+ {
+ 'Id': '1',
+ 'MessageBody': 'bar',
+ 'MessageAttributes': {
+ 'barAttr1': {
+ 'DataType': 'String',
+ 'StringValue': 'barAttr1Value'
+ },
+ 'barAttr2': {
+ 'DataType': 'String',
+ 'StringValue': 'barAttr2Value'
+ }
  }
- }
- )
+ },
+ ]
+ replay_messages.QUEUE.send_messages.assert_called_with(Entries=expected_entries)
+
+
+def test_handler_failure(mocker, replay_messages):
+ replay_messages.QUEUE.send_messages.return_value = {'Failed': 'True'}
+
+ event = {
+ 'Records': [
+ {
+ 'body': 'foo',
+ 'messageAttributes': {}
+ }
+ ]
+ }
+
+ with pytest.raises(RuntimeError):
+ replay_messages.handler(event, None)