Amazon SQS

​

Config

connector_type: amazon_sqs
queue_names:
  - producer-events
  - controller-events
# Optional. Defaults to false so startup fails if any queue can't be initialized.
skip_failed_queues: false
# Optional. Prefer IAM roles or environment variables for production.
# aws_access_key_id: "{{ env_var('AWS_ACCESS_KEY_ID') }}"
# aws_secret_access_key: "{{ env_var('AWS_SECRET_ACCESS_KEY') }}"
# aws_session_token: "{{ env_var('AWS_SESSION_TOKEN') }}"  # Optional for temporary credentials
# aws_region: "{{ env_var('AWS_DEFAULT_REGION') }}"
batch_size: 10
wait_time_seconds: 1
message_deletion_method: AFTER_RECEIVED   # AFTER_RECEIVED or MANUAL

serde_config:
  serialization_method: JSON

connector_type: amazon_sqs
queue_name: producer-events
batch_size: 10
wait_time_seconds: 1

skip_failed_queues: true

​

Authentication

1. Grant your Mage runtime an AWS IAM role with access to the configured queues. This is the recommended option for Mage deployments running on AWS infrastructure.
2. Add the following keys and values to your runtime environment variables:
   • AWS_ACCESS_KEY_ID
   • AWS_SECRET_ACCESS_KEY
   • AWS_SESSION_TOKEN if you’re using temporary AWS credentials
   • AWS_DEFAULT_REGION
3. Configure credentials in the source config. Use environment-variable references instead of pasting raw secrets into the block YAML:
   aws_access_key_id: "{{ env_var('AWS_ACCESS_KEY_ID') }}"
   aws_secret_access_key: "{{ env_var('AWS_SECRET_ACCESS_KEY') }}"
   aws_session_token: "{{ env_var('AWS_SESSION_TOKEN') }}"
   aws_region: us-east-1
   
4. If you deploy Mage on an AWS ECS cluster, you can use the ECS task execution role to authenticate. Grant the ECS task permissions to access SQS by attaching IAM policies to this role.

​

Pass raw message to transformer

serde_config:
  serialization_method: RAW_VALUE

​

Message deletion method

1. Delete the message in the data loader automatically after deserializing the message body.
   • Set message_deletion_method: AFTER_RECEIVED in the source config.
   • The input of the transformer is the list of deserialized message body.
2. Manually delete the message in transformer after processing the message.
   • Set message_deletion_method: MANUAL in the source config.
   • The input of the transformer is the list of dictionary with the structure
     {
       'parsed_msg_body': {'k1': 'v1', ...},
       'queue_name': 'producer-events',
       'queue_url': 'https://sqs...',
       'raw_message': sqs.Message(...)
     }
     
     You can use the following example code to process the message and delete it after.
     @transformer
     def transform(messages: List[Dict], *args, **kwargs):
         processed_msgs = []
         for msg in messages:
             processed_msg = msg['parsed_msg_body']
             # Add your own logic to process the message
             processed_msgs.append(processed_msg)
             msg['raw_message'].delete()
         return processed_msgs