python 在AWS CLI调用上获取Throttling Exception作为ThrottlingException错误

gopyfrb3  于 2023-06-20  发布在  Python
关注(0)|答案(1)|浏览(107)

我试图通过AWS CLI探索一个名为Workspaces的新AWS服务,它似乎有能力满足1个请求/秒。当我尝试同时命中多次时,它会抛出ThrottlingException错误。由于工作区还没有在boto包中,所以我通过子进程调用在python中使用CLI。

def describe_workspaces():
        process=subprocess.Popen(access_endpoint.split(), stdout=subprocess.PIPE)
        output=process.communicate()[0]

所以,如果我调用这个函数>=1/sec,我将有ThrottlingException。怎么处理?并且会有多个用户同时调用此函数。
我想做批处理和异步调用,但如何适应这种架构?

n3h0vuf2

n3h0vuf21#

新方案:

您可以使用Boto3Python中的AWS SDK)提供的botocore.config中的Config来解决此问题,方法是指定重试配置,如以下代码片段所示:

import boto3 
from botocore.config import Config

config=Config(connect_timeout=5, read_timeout=60, retries={'max_attempts': 5})

def describe_workspaces():
    return boto3.client('workspaces', config).describe_workspaces()

参考文献:

您可以使用Boto3Python中的AWS SDK)解决此问题,方法是为ThrottlingException添加异常处理并重试,如下面的代码片段所示:

import boto3
from botocore.exceptions import ClientError

def describe_workspaces(tries=1):
    try:
        return boto3.client('workspaces').describe_workspaces()
    except ClientError as exception_obj:
        if exception_obj.response['Error']['Code'] == 'ThrottlingException':
            if tries <= 3:
                print("Throttling Exception Occured.")
                print("Retrying.....")
                print("Attempt No.: " + str(tries))
                time.sleep(3*tries)
                return describe_workspaces(tries + 1)
            else:
                print("Attempted 3 Times But No Success.")
                print("Raising Exception.....")
                raise
        else:
            raise

您可以在函数外部创建AWS client,并可以根据需要修改逻辑。
您也可以通过AWS CLI处理ThrottlingException,但如果您编写一些Bash/Shell脚本而不是Python脚本,则会更有意义。对于Python,建议使用Boto3
有关更多详细信息,请查看以下内容:AWS Workspaces APIs

通用旧方案:

如果您的代码有多个(不止一个)AWS client调用,并且您不想为ThrottlingException处理添加相同的代码,并且为了实现可重用性而一次又一次地重试每个调用(DRY;不要重复自己)并避免不必要的代码重复,可以使用以下代码片段:

import time
import boto3
from botocore.exceptions import ClientError

def generic_aws_client_method(client, method_name, method_arguments, tries=1):
    """Generic AWS client method with throttling exception handling"""
    try:
        client_method = getattr(client, method_name)
        return client_method(**method_arguments)
    except ClientError as exception_obj:
        if exception_obj.response['Error']['Code'] == 'ThrottlingException':
            if tries <= 3:
                print("Throttling Exception Occured.")
                print("Retrying.....")
                print("Attempt No.: " + str(tries))
                time.sleep(3*tries)
                return generic_aws_client_method(
                    client,
                    method_name,
                    method_arguments,
                    tries + 1
                )
            else:
                print("Attempted 3 Times But No Success.")
                print("Raising Exception.....")
                raise
        else:
            raise

下面你可以找到一些关于如何使用它的例子:

# AWS clients for AWS S3 and AWS SSM respectively
s3_client = boto3.client('s3')
ssm_client = boto3.client('ssm')

# Example 1: With multiple method arguments
mapping = {'Name': '/prod/aws-eks-vpc-id', 'WithDecryption': True}
returned_object = generic_aws_client_method(ssm_client, "get_parameter", mapping)
print(returned_object)

# Example 2: With single method argument
mapping = {'operation_name': 'describe_parameters'}
returned_object = generic_aws_client_method(ssm_client, "get_paginator", mapping)
print(returned_object)

# Example 3: With no method argument
mapping = {}
returned_object = generic_aws_client_method(s3_client, "list_buckets", mapping)
print(returned_object)

测试ThrottlingException处理:

为了测试上述代码片段中的ThrottlingException处理,请使用以下代码片段自行引发一个自定义ClientError

raise ClientError(
    {'Error': {'Code': 'ThrottlingException', 'Message': 'Rate exceeded'}},
    'generic_aws_client_method'
)

相关问题