model_script/resource/service/RDSServicee.py

import uuid
from typing import Dict, List, Tuple

from alibabacloud_rds20140815.client import Client as RDSClient
from alibabacloud_rds20140815.models import DescribeDBInstancesResponseBodyItemsDBInstance, DescribeDBInstancesRequest, DescribeDBInstanceAttributeRequest, \
    DescribeDBInstanceAttributeResponseBodyItemsDBInstanceAttribute
from alibabacloud_tea_openapi import models

from resource.enums.region import Region


class RDSService(object):
    def __init__(self):
        self.access_key_id = "LTAI5tRwjztCCwQNBB6nW1dY"
        self.access_key_secret = "NaTnMxrGEJh64tLly7Kb5tr166Xpos"
        self.region_client_map: Dict[Region, RDSClient] = {}

    def create_client(self, region: Region) -> RDSClient:
        config = models.Config(
            access_key_id=self.access_key_id,
            access_key_secret=self.access_key_secret,
            endpoint=region.rds_endpoint
        )
        return RDSClient(config)

    def get_rds_client(self, region: Region) -> RDSClient:
        if region in self.region_client_map:
            return self.region_client_map[region]

        client = self.create_client(region)
        self.region_client_map[region] = client
        return client

    def get_all_instance(self, region: Region) -> List[DescribeDBInstancesResponseBodyItemsDBInstance]:
        page_number = 0
        all_instances: List[DescribeDBInstancesResponseBodyItemsDBInstance] = []
        while True:
            next_token, instances = self.get_instances(region=region, page_number=page_number)
            if not instances:
                break
            all_instances.extend(instances)
            page_number += 1
        return all_instances

    def get_instances(self, region: Region, page_size: int = 100, page_number: int = 0, next_token: str = "") -> Tuple[str, List[DescribeDBInstancesResponseBodyItemsDBInstance]]:
        client = self.get_rds_client(region)
        request = DescribeDBInstancesRequest(
            client_token=self.gen_request_id(),
            page_size=page_size,
            page_number=page_number,
            region_id=region.region,
            instance_level=1,
        )
        response = client.describe_dbinstances(request)
        body = response.body
        return body.next_token, body.items.dbinstance

    def get_instance_detail(self, region: Region, instance_id: str) -> List[DescribeDBInstanceAttributeResponseBodyItemsDBInstanceAttribute]:
        client = self.get_rds_client(region)
        request = DescribeDBInstanceAttributeRequest(
            dbinstance_id=instance_id
        )
        response = client.describe_dbinstance_attribute(request)
        return response.body.items.dbinstance_attribute

    @staticmethod
    def gen_request_id():
        return str(uuid.uuid4()).replace("-", "")