Python 运维开发

Appearance

腾讯云 python sdk ​

API 文档地址：https://cloud.tencent.com/document/api
我们使用更加方便的 Common Client 调用方式，点击查看 CommonClient API 示例

TencentCloudAPI 集成 ​

python
from sanic import Sanic
from tencentcloud.common.common_client import CommonClient
from tencentcloud.common import credential

from utils.common import retry


class TencentCloudAPI():
    """腾讯云 API"""
    def __init__(self, product_code, region='ap-beijing'):
        self.app = Sanic.get_app()
        account_si = self.app.config.TENCENTCLOUD_SECRET_ID
        account_sk = self.app.config.TENCENTCLOUD_SECRET_KEY

        cred = credential.Credential(account_si, account_sk)
        client = self.app.ctx.pymongo_client
        res = client['tencentcloud']['product_type'].find_one({
            'code': product_code,
        })
        version = res['version']
        self.client = CommonClient(product_code, version, cred, region)

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

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

        if attr in {'GetUser', 'DescribeRecordList'}:
            # 用户不存在会直接报异常，retry 不再适用，所以单独提取出来了
            return func_no_retry
        else:
            return func
txt
code        name                    version
----------------------------------------------
cvm	        云服务器	                2017-03-12
cdb	        云数据库 MySQL	        2017-03-20
cdn	        腾讯云 CDN	            2018-06-06
clb	        负载均衡	                2018-03-17
cam	        子账号	                2019-01-16
tcb	        云托管	                2018-06-08
tcbr	    云托管 RUN	            2022-02-17
cls	        日志服务	                2020-10-16
cynosdb	    TDSQL-C MySQL	        2019-01-07
dnspod	    DNSPod	                2021-03-23
postgres	云数据库 PostgreSQL	    2017-03-12
redis	    云数据库 Redis	        2018-04-12
mongodb	    云数据库 MongoDB	        2019-07-25
monitor	    云监控	                2018-07-24
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 文档中查询得到

如何使用 TencentCloudAPI ​

python
# 获取服务器列表
TencentCloudAPI('cvm').DescribeInstances(Offset=0, Limit=100)
python
import threading

from .api import TencentCloudAPI


class CVM(TencentCloudAPI):
    def __init__(self):
        super().__init__('cvm')

    def get_all_cvm_multi_thread(self):
        """获取所有服务器"""
        cvm_list = []
        limit = 100

        def _get_cvm_total_count():
            kwargs = {
                'Offset': 0,
                'Limit': 1,
            }
            res = self.DescribeInstances(**kwargs)
            total_count = res['Response']['TotalCount']
            return total_count

        def _get_cvm_info_thread(offset):
            kwargs = {
                'Offset': offset,
                'Limit': limit,
            }
            res = self.DescribeInstances(**kwargs)
            cvm_list.extend(res['Response']['InstanceSet'])

        total_count = _get_cvm_total_count()

        threads = []
        for per_count in range(0, total_count, limit):
            t = threading.Thread(
                    target=_get_cvm_info_thread, args=(per_count,))
            t.start()
            threads.append(t)
        for th in threads:
            th.join()
        return cvm_list
  • 如果对同一个产品要使用较多的 api 进行管理，则可以将此产品使用继承方式独立出一个子类，方便代码调用及管理