Skip to content

AWS python sdk

api 文档地址:https://boto3.amazonaws.com/v1/documentation/api/latest/index.html

安装与权限配置

https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html

  • 安装pip install boto3
  • 配置:先运行 aws configure ,然后按照提示输入 aws_access_key_idaws_secret_access_key 的值即可。默认会保存到 ~/.aws/credentials 文件中

AWSAPI 集成

python
import boto3
from utils.common.decorator import retry


class AWSAPI():
    """AWS api
    https://boto3.amazonaws.com/v1/documentation/api/latest/index.html
    """
    def __init__(self, product_code, region='us-east-1'):
        self.client = boto3.client(product_code, region_name=region)
        self.region = region

    def __getattr__(self, attr):
        @retry(10, 5, func_name=attr)
        def func(**kwargs):
            res = getattr(self.client, attr)(**kwargs)
            return res

        def func_no_retry(**kwargs):
            res = getattr(self.client, attr)(**kwargs)
            return res

        if attr in {'get_user'}:
            # 用户不存在会直接报异常,retry 不再适用,所以单独提取出来了
            return func_no_retry
        else:
            return func
python
def retry(times=20, sleep=2, func_name=None):
    """重试装饰器
    用于重试一些非代码问题的异常,比如调用阿里云等等
    """
    def decorator(func):
        def wrapper(*args, **kw):
            if func_name:
                func.__name__ = func_name
            for i in range(times):
                try:
                    return func(*args, **kw)
                except Exception as e:
                    time.sleep(sleep)
                    logger.info(
                        f'报错信息:{e};重试 {func.__name__} 函数'
                        f'第 {i + 1}/{times} 次...'
                    )
            else:
                raise Exception(f'{func.__name__} 函数超时失败!')
        return wrapper
    return decorator
  • 使用子账号API 密钥进行鉴权即可,需要给子账号分配相应的产品权限

如何使用 AWSAPI

python
# 获取地域列表
AWSAPI('ec2').describe_regions()
python
from .api import AWSAPI


class EC2(AWSAPI):
    """云服务器
    https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2.html
    """
    def __init__(self, region='us-east-1'):
        super().__init__('ec2', region=region)

    def get_all_instance_type(self):
        """获取所有实例类型"""
        instance_types = []
        kwargs = {'MaxResults': 100}
        while True:
            res = self.describe_instance_types(**kwargs)
            for it in res['InstanceTypes']:
                it['Region'] = self.region
            instance_types.extend(res['InstanceTypes'])
            next_token = res.get('NextToken')
            if next_token is None:
                break
            kwargs['NextToken'] = next_token
        return instance_types
  • 如果对同一个产品要使用较多的 api 进行管理,则可以将此产品使用继承方式独立出一个子类,方便代码调用及管理