使用 Boto3 创建 SQS 队列时指定死信队列

Question

我正在尝试使用 boto3 创建一对 SQS 队列。第一个是第二个的死信队列。

def create_queues(qname):

    # create a fifo queue with a deadletter queue
    dl_queue = sqs.create_queue(
        QueueName=f'{qname}-dead-letter.fifo',
        Attributes={'FifoQueue': "true"}
    )
    dl_arn = dl_queue.attributes['QueueArn']
    print(dl_arn)


    policy = {
      "maxReceiveCount" : '3',
      "deadLetterTargetArn": dl_arn
    }
    policy = json.dumps(policy)

    task_queue = sqs.create_queue(
        QueueName=f'{qname}.fifo',
        Attributes={'RedrivePolicy': policy}
    )

create_queues('test')

我可以使用其他属性创建队列，但指定死信队列 ARN 所需的 RedrivePolicy 属性是嵌套属性，而所有其他属性都是简单的键值对。 boto 文档页面 不清楚如何处理这个嵌套属性。

我尝试过 boto.resource 和 boto.client，以及 JSON、字符串和 Python 字典的几种变体。我还看到了一堆相关的错误消息和问题，但还没有找到一个简单的解决方案（我认为事后设置属性是一种解决方法，但我想弄清楚如何在创建时设置此 RedrivePolicy .）

我收到的上述代码的错误消息如下。我不确定它是否在抱怨 ARN 中的冒号或 JSON 策略中的引号。

Traceback (most recent call last):
  File "sqshelper.py", line 57, in <module>
    create_test_queue()
  File "sqshelper.py", line 29, in create_test_queue
    queue_url = create_sqs('testbot.fifo', redrive_policy=redrive_policy)
  File "sqshelper.py", line 16, in create_sqs
    response = sqs_client.create_queue(
  File "/Users/g/git/sqstasks/venv/lib/python3.8/site-packages/botocore/client.py", line 395, in _api_call
    return self._make_api_call(operation_name, kwargs)
  File "/Users/g/git/sqstasks/venv/lib/python3.8/site-packages/botocore/client.py", line 725, in _make_api_call
    raise error_class(parsed_response, operation_name)
botocore.exceptions.ClientError: An error occurred (InvalidParameterValue) when calling the CreateQueue operation: Can only include alphanumeric characters, hyphens, or underscores. 1 to 80 in length

Answer 1

正如 jarmod 在评论中暗示的那样，这是因为没有为死信队列指定 Fifo 属性。

以下代码按预期工作。

import boto3
import json

from dotenv import load_dotenv

load_dotenv()

sqs = boto3.resource("sqs", region_name="eu-west-2")


def create_queues(qname):

    # create a fifo queue with a deadletter queue
    dl_queue = sqs.create_queue(
        QueueName=f"{qname}-dead-letter.fifo", Attributes={"FifoQueue": "true"}
    )
    dl_arn = dl_queue.attributes["QueueArn"]

    print("successfully created dead letter queue")
    print(dl_arn)

    policy = {"maxReceiveCount": "3", "deadLetterTargetArn": dl_arn}
    policy = json.dumps(policy)

    task_queue = sqs.create_queue(
        QueueName=f"{qname}.fifo",
        Attributes={"RedrivePolicy": policy, "FifoQueue": "true"},
    )


create_queues("testqueue")