本文以查询Redis实例的慢日志为例介绍如何使用Python调用数据库自治服务DAS(Database Autonomy Service)的SDK。

使用Aliyun的SDK进行调用

import json

from alibabacloud_r_kvstore20150101 import models as r_kvstore_20150101_models
from alibabacloud_r_kvstore20150101.client import Client as R_kvstore20150101Client
from alibabacloud_tea_openapi import models as open_api_models
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from django.conf import settings


class RedisClient:
def __init__(
self,
access_key_id: str = settings.ALIYUN['ACCESS_KEY_ID'],
access_key_secret: str = settings.ALIYUN['ACCESS_KEY_SECRET'],
region_id: str = settings.ALIYUN['REGION_ID']
):
self.region_id = region_id
config = open_api_models.Config(
# 您的AccessKey ID,
access_key_id=access_key_id,
# 您的AccessKey Secret,
access_key_secret=access_key_secret
)
# 访问的域名
config.endpoint = 'r-kvstore.aliyuncs.com'
self.client = R_kvstore20150101Client(config)

def get_hot_big_keys(
self, instance_id
) -> dict:
"""
查询实时热点Key和大Key
"""
client = AcsClient(settings.ALIYUN['ACCESS_KEY_ID'], settings.ALIYUN['ACCESS_KEY_SECRET'], 'cn-shanghai')
request = CommonRequest()
request.set_domain('das.cn-shanghai.aliyuncs.com')
request.set_version('2020-01-16')
request.set_action_name('DescribeHotBigKeys')
request.add_query_param('InstanceId', instance_id)
response = client.do_action_with_exception(request)
data = {}
if response:
try:
data = json.loads(response).get('Data', {})
except (Exception, ):
return {}
else:
return data
return data

def describe_slow_log_records(
self, instance_id, start_time: str = '', end_time: str = '',
page_size: int = 100, page_number: int = 1
) -> dict:
"""
查询实例的慢查询日志
"""
client = self.client
describe_available_slow_record_request = r_kvstore_20150101_models.DescribeSlowLogRecordsRequest(
start_time=start_time, end_time=end_time, instance_id=instance_id, page_size=page_size,
page_number=page_number)
data = client.describe_slow_log_records(describe_available_slow_record_request)
data = data.to_map()
return data.get('body', {})