Python 运维开发

Appearance

阿里云 python sdk ​

SDK 地址：https://help.aliyun.com/zh/sdk/developer-reference/python-sdk/
使用 CommonRequest 进行调用：https://help.aliyun.com/zh/sdk/developer-reference/use-commonrequest

AliyunAPI 集成 ​

python
import json
from sanic import Sanic

from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest

from utils.common import retry


class AliyunAPI():
    """阿里云 API
    https://help.aliyun.com/zh/sdk/developer-reference/use-commonrequest
    """
    def __init__(self, product_code, region='cn-beijing'):
        self.app = Sanic.get_app()
        self.region = region
        self.client = AcsClient(
            self.app.config.ALIYUN_ACCESSKEY_ID,
            self.app.config.ALIYUN_ACCESSKEY_SECRET,
            self.region,
            timeout=300,
        )
        self.reqeust = self._init_request(product_code)

    def _init_request(self, product_code):
        request = CommonRequest()
        request.set_accept_format('json')
        request.set_method('POST')
        res = self.app.ctx.pymongo_client['aliyun']['product_type'].find_one({
            'code': product_code,
        })
        domain = res['domain']
        if 'region' in domain:
            domain = domain.replace('region', self.region)
        request.set_domain(domain)
        request.set_version(res['version'])
        return request

    def __getattr__(self, attr):
        @retry(10, 5, func_name=attr)
        def func(**kwargs):
            request = self.reqeust
            request.set_action_name(attr)
            for k, v in kwargs.items():
                request.add_query_param(k, v)
            response = json.loads(
                self.client.do_action_with_exception(request))
            return response
        return func
txt
code        domain                      name            version
------------------------------------------------------------------
ecs         ecs.aliyuncs.com	        云服务器	        2014-05-26
dns	        alidns.aliyuncs.com	        DNS 域名解析	    2015-01-09
alb	        alb.region.aliyuncs.com	    ALB	            2020-06-16
business	business.aliyuncs.com	    账单管理	        2017-12-14
cdn	        cdn.aliyuncs.com	        CDN	            2018-05-10
metrics	    metrics.aliyuncs.com	    监控数据	        2019-01-01
mongodb	    mongodb.aliyuncs.com	    MongoDB	        2015-12-01
vpc	        vpc.aliyuncs.com	        VPC	            2016-04-28
polardb	    polardb.aliyuncs.com	    PolarDB	        2017-08-01
ram	        ram.aliyuncs.com	        RAM 访问控制	    2015-05-01
rds	        rds.aliyuncs.com	        RDS	            2014-08-15
redis	    r-kvstore.aliyuncs.com	    Redis	        2015-01-01
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 密钥进行鉴权即可，需要给子账号分配相应的产品权限
  • 产品 code 和 version 数据存于数据库；是为了方便统一管理，以及使代码更加简洁。其数据可在各自的 api 文档中查询得到

如何使用 AliyunAPI ​

python
# 获取地域列表
AliyunAPI('ecs').DescribeRegions()
python
from .api import AliyunAPI


class ECS(AliyunAPI):
    def __init__(self):
        super().__init__('ecs')

    # ...
  • 如果对同一个产品要使用较多的 api 进行管理，则可以将此产品使用继承方式独立出一个子类，方便代码调用及管理