Alibabacloud Dms Data Agent Platform Setup

import re
from typing import List, Any, Optional, Tuple, Dict
from copy import deepcopy


class void_arg_t:
    def __init__(self):
        pass


DEFAULT_USERNAME = "dify_user"

# 敏感字段名集合，错误消息中对其值进行脱敏
_SENSITIVE_KEYS = {"DbInstancePassword", "KvStorePassword", "VectordbPassword"}

# 字符串参数的正则格式约束：{参数名: (pattern, 说明)}
_STR_FORMAT_RULES: Dict[str, Tuple[str, str]] = {
    "VpcId":            (r"^vpc-[a-z0-9]+$",          "should match ^vpc-[a-z0-9]+$"),
    "VSwitchId":        (r"^vsw-[a-z0-9]+$",          "should match ^vsw-[a-z0-9]+$"),
    "BackupVSwitchId":  (r"^vsw-[a-z0-9]+$",          "should match ^vsw-[a-z0-9]+$"),
    "SecurityGroupId":  (r"^sg-[a-z0-9]+$",           "should match ^sg-[a-z0-9]+$"),
    "WorkspaceId":      (r"^[a-zA-Z0-9_-]{1,64}$",    "alphanumeric/underscore/hyphen, 1-64 chars"),
    "WorkspaceName":    (r"^[a-zA-Z0-9_-]{1,64}$",    "alphanumeric/underscore/hyphen, 1-64 chars"),
    "DbInstanceAccount":(r"^[a-zA-Z][a-zA-Z0-9_]{0,63}$", "starts with letter, alphanumeric/underscore, 1-64 chars"),
    "DbInstancePassword":(r"^.{8,32}$",               "8-32 characters"),
    "KvStoreAccount":   (r"^[a-zA-Z][a-zA-Z0-9_]{0,63}$", "starts with letter, alphanumeric/underscore, 1-64 chars"),
    "KvStorePassword":  (r"^.{8,32}$",                "8-32 characters"),
    "VectordbAccount":  (r"^[a-zA-Z][a-zA-Z0-9_]{0,63}$", "starts with letter, alphanumeric/underscore, 1-64 chars"),
    "VectordbPassword": (r"^.{8,32}$",                "8-32 characters"),
}


class bodyArgItem_t:
    def __init__(
        self,
        name: str,
        val_type: Any,
        default_val: Any,
        allowed_val_list: Optional[List[Any]],
        required: bool,
    ):
        assert isinstance(name, str)
        self.key_name = name
        self.val_type = val_type
        self.val = default_val
        self.allowed_val_list = allowed_val_list
        self.required = required
        self.check_val(strict=False)

    def check_val(self, strict: bool = True):
        if isinstance(self.val, void_arg_t):
            if strict and self.required:
                raise ValueError(f"{self.key_name}: required")
            return
        # 1. 检查类型
        if self.val_type is not Any and not isinstance(self.val, self.val_type):
            # 对敏感字段脱敏，避免密码出现在错误消息中
            display_val = "***" if self.key_name in _SENSITIVE_KEYS else self.val
            raise ValueError(
                f"{self.key_name}: val <{display_val}> type mis-match, expect: {self.val_type}, got: {type(self.val)}"
            )
        # 2. 检查枚举值
        if self.allowed_val_list and self.val not in self.allowed_val_list:
            display_val = "***" if self.key_name in _SENSITIVE_KEYS else self.val
            raise ValueError(
                f"{self.key_name}: val <{display_val}> not in allowed list <{self.allowed_val_list}>"
            )
        # 3. 对字符串参数进行正则格式校验
        if isinstance(self.val, str) and self.key_name in _STR_FORMAT_RULES:
            pattern, description = _STR_FORMAT_RULES[self.key_name]
            if not re.fullmatch(pattern, self.val):
                raise ValueError(
                    f"{self.key_name}: invalid format, {description}"
                )

    def set_val(self, new_val: Any):
        self.val = new_val
        self.check_val()


FULL_PARAM_LIST: Tuple[bodyArgItem_t, ...] = (
    bodyArgItem_t("VpcId", str, void_arg_t(), None, True),
    bodyArgItem_t("VSwitchId", str, void_arg_t(), None, True),
    bodyArgItem_t("BackupVSwitchId", str, void_arg_t(), None, True),
    bodyArgItem_t("SecurityGroupId", str, void_arg_t(), None, True),
    bodyArgItem_t("NatGatewayOption", str, "NoNeed", ("NoNeed", "Enable"), True),
    bodyArgItem_t("ZoneId", str, void_arg_t(), None, True),
    bodyArgItem_t("DataRegion", str, void_arg_t(), None, True),
    bodyArgItem_t("ResourceQuota", str, "12CU", None, True),
    bodyArgItem_t("Replicas", int, 1, None, True),
    bodyArgItem_t(
        "WorkspaceOption",
        str,
        "CreateNewInstance",
        ("UseExistingInstance", "CreateNewInstance"),
        True,
    ),
    bodyArgItem_t("WorkspaceId", str, void_arg_t(), None, False),
    bodyArgItem_t("WorkspaceName", str, void_arg_t(), None, False),
    bodyArgItem_t(
        "DatabaseOption",
        str,
        "CreateNewInstance",
        ("CreateNewInstance", "UseExistingInstance"),
        True,
    ),
    bodyArgItem_t("DbResourceId", int, void_arg_t(), None, False),
    bodyArgItem_t("DbEngineType", str, "PostgreSQL", None, True),
    bodyArgItem_t("DbInstanceClass", str, "pg.n2.2c.1m", None, True),
    bodyArgItem_t("DbEngineVersion", str, "15.0", None, True),
    bodyArgItem_t("DbInstanceCategory", str, "Basic", None, True),
    bodyArgItem_t("DbStorageType", str, "general_essd", None, True),
    bodyArgItem_t("DbStorageSize", str, "60", None, True),
    bodyArgItem_t("DbInstanceAccount", str, DEFAULT_USERNAME, None, True),
    bodyArgItem_t("DbInstancePassword", str, void_arg_t(), None, True),
    bodyArgItem_t(
        "KvStoreOption",
        str,
        "CreateNewInstance",
        ("CreateNewInstance", "UseExistingInstance"),
        True,
    ),
    bodyArgItem_t("KvStoreResourceId", int, void_arg_t(), None, False),
    bodyArgItem_t("KvStoreType", str, "REDIS", None, True),
    bodyArgItem_t("KvStoreInstanceClass", str, "redis.shard.large.ce", None, True),
    bodyArgItem_t("KvStoreEngineVersion", str, "6.0", None, True),
    bodyArgItem_t("KvStoreNodeType", str, "MASTER_SLAVE", None, True),
    bodyArgItem_t("KvStoreAccount", str, DEFAULT_USERNAME, None, True),
    bodyArgItem_t("KvStorePassword", str, void_arg_t(), None, True),
    bodyArgItem_t(
        "VectordbOption",
        str,
        "CreateNewInstance",
        ("CreateNewInstance", "UseExistingInstance"),
        True,
    ),
    bodyArgItem_t("VectordbResourceId", int, void_arg_t(), None, False),
    bodyArgItem_t("VectordbType", str, void_arg_t(), None, False),
    bodyArgItem_t("VectordbCategory", str, "Basic", None, True),
    bodyArgItem_t("VectordbEngineVersion", str, "7.0", None, True),
    bodyArgItem_t("VectordbInstanceSpec", str, "4C16G", None, True),
    bodyArgItem_t("SegNodeNum", int, 2, None, True),
    bodyArgItem_t("VectordbStorageType", str, "cloud_essd", None, True),
    bodyArgItem_t("SegDiskPerformanceLevel", str, "pl1", None, True),
    bodyArgItem_t("VectordbStorageSize", str, "50", None, True),
    bodyArgItem_t("VectordbAccount", str, DEFAULT_USERNAME, None, True),
    bodyArgItem_t("VectordbPassword", str, void_arg_t(), None, True),
    bodyArgItem_t("StorageType", str, "cloud_essd", None, True),
    bodyArgItem_t("PayType", str, "PrePaid", ("PrePaid", "PostPaid"), True),
    bodyArgItem_t("PayPeriodType", str, "Month", ("Month", "Year"), True),
    bodyArgItem_t("PayPeriod", int, 1, None, True),
    bodyArgItem_t("DryRun", bool, True, (True, False), True),
    bodyArgItem_t("MajorVersion", str, "1.13.x", None, True),
    bodyArgItem_t(
        "Edition",
        str,
        "OpenCommunity",
        ("OpenCommunity", "Community", "Enterprise"),
        True,
    ),
    bodyArgItem_t("EnableExtraEndpoint", bool, True, (True, False), True),
    bodyArgItem_t("OnlyIntranet", bool, False, (True, False), True),
)


def fill_in_param_body(customer_input: Dict[str, Any]) -> Dict[str, Any]:
    param_list: List[bodyArgItem_t] = deepcopy(FULL_PARAM_LIST)

    def reset_param_val(key_name: str, val: Any):
        for param in param_list:
            if param.key_name == key_name:
                param.set_val(val)
                return
        raise ValueError(f"unknown param key name: {key_name}")

    for key_name in customer_input:
        reset_param_val(key_name, customer_input[key_name])

    result = {}
    for param in param_list:
        param: bodyArgItem_t
        param.check_val()
        if not isinstance(param.val, void_arg_t):
            result[param.key_name] = param.val
    post_check(result)
    return result


def post_check(param_body: Dict[str, Any]):
    # 1. 新建或使用已有工作空间
    assert "WorkspaceOption" in param_body, "WorkspaceOption is required"
    if param_body["WorkspaceOption"] == "CreateNewInstance":
        assert (
            "WorkspaceName" in param_body
        ), "WorkspaceName wanted when CreateNewInstance"
    else:
        assert (
            "WorkspaceId" in param_body
        ), "WorkspaceId needed when UseExistingInstance"

    # 2. 新建或使用已有数据库实例
    assert "DatabaseOption" in param_body, "DatabaseOption is required"
    if param_body["DatabaseOption"] == "UseExistingInstance":
        assert (
            "DbResourceId" in param_body
        ), "DbResourceId needed when UseExistingInstance"

    # 3. 新建或使用已有 KV 存储实例
    assert "KvStoreOption" in param_body, "KvStoreOption is required"
    if param_body["KvStoreOption"] == "UseExistingInstance":
        assert (
            "KvStoreResourceId" in param_body
        ), "KvStoreResourceId needed when UseExistingInstance"

    # 4. 新建或使用已有向量库实例
    assert "VectordbOption" in param_body, "VectordbOption is required"
    if param_body["VectordbOption"] == "UseExistingInstance":
        assert (
            "VectordbResourceId" in param_body
        ), "VectordbResourceId needed when UseExistingInstance"


if __name__ == "__main__":
    print(f"共有 {len(FULL_PARAM_LIST)} 个参数需要填写")

import hashlib
import sys
import json

from typing import List, Dict, Any

from alibabacloud_tea_openapi.client import Client as OpenApiClient
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_openapi_util.client import Client as OpenApiUtilClient
from fill_in_param_body import fill_in_param_body


class callAliyunOpenAPI:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> OpenApiClient:
        """
        使用凭据初始化账号Client，凭据通过默认凭据链自动获取。
        凭据链配置方式请参见：https://help.aliyun.com/zh/sdk/developer-reference/v2-manage-python-access-credentials#3ca299f04bw3c
        @return: Client
        @throws Exception
        """
        # 使用默认凭据链，无需手动传入 AK/SK
        credentialsClient = CredentialClient()

        config = open_api_models.Config(credential=credentialsClient)
        config.user_agent = "AlibabaCloud-Agent-Skills/alibabacloud-dms-data-agent-platform-setup"
        # Endpoint 请参考 https://api.aliyun.com/product/dms-enterprise
        config.endpoint = "dms-enterprise.aliyuncs.com"
        return OpenApiClient(config)

    @staticmethod
    def create_api_info() -> open_api_models.Params:
        """
        API 相关
        @param path: string Path parameters
        @return: OpenApi.Params
        """
        params = open_api_models.Params(
            # 接口名称,
            action="CreateDifyInstance",
            # 接口版本,
            version="2018-11-01",
            # 接口协议,
            protocol="HTTPS",
            # 接口 HTTP 方法,
            method="POST",
            auth_type="AK",
            style="RPC",
            # 接口 PATH,
            pathname="/",
            # 接口请求体内容格式,
            req_body_type="json",
            # 接口响应体内容格式,
            body_type="json",
        )
        return params

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        if len(args) != 1:
            print("请传入一个json格式的字符串，作为命令行参数")
            exit(1)
        json_body: Dict[str, Any] = json.loads(args[0])
        json_body = fill_in_param_body(json_body)

        # 基于请求参数生成幂等 ClientToken，确保相同参数的重试不会创建重复实例
        client_token = hashlib.sha256(
            json.dumps(json_body, sort_keys=True).encode()
        ).hexdigest()
        json_body["ClientToken"] = client_token

        client = callAliyunOpenAPI.create_client()
        params = callAliyunOpenAPI.create_api_info()

        # runtime options
        runtime = util_models.RuntimeOptions(
            connect_timeout=10000,  # 连接超时 10 秒（单位：毫秒）
            read_timeout=30000,  # 读取超时 30 秒（单位：毫秒）
        )
        request = open_api_models.OpenApiRequest(
            query=OpenApiUtilClient.query(json_body)
        )
        # 返回值实际为 Map 类型，可从 Map 中获得三类数据：响应体 body、响应头 headers、HTTP 返回的状态码 statusCode。
        resp: Dict[str, Any] = client.call_api(params, request, runtime)
        # 只输出 body 中对用户有用的字段，避免暴露 HTTP headers 等非必要信息
        body: Dict[str, Any] = resp.get("body", {})

        # 对 Data 中已知敏感字段进行脱敏处理
        _SENSITIVE_DATA_KEYS = {"Password", "password", "AccessKey", "SecretKey", "ConnectionString", "connectionString"}
        raw_data = body.get("Data")
        if isinstance(raw_data, dict):
            sanitized_data = {
                k: ("***" if k in _SENSITIVE_DATA_KEYS else v)
                for k, v in raw_data.items()
            }
        else:
            sanitized_data = raw_data

        output = {
            "Success": body.get("Success"),
            "HttpStatusCode": body.get("HttpStatusCode"),
            "RequestId": body.get("RequestId"),
            "Data": sanitized_data,
        }
        print(json.dumps(output, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    callAliyunOpenAPI.main(sys.argv[1:])
    exit(0)

alibabacloud-tea-openapi==0.4.4
alibabacloud-credentials==1.0.8
alibabacloud-tea-util==0.3.14
alibabacloud-openapi-util==0.2.4