from typing import Dict, Any, List, Set, Optional import json from pydantic import BaseModel from llmops.environment.executor import CompoundExecutor from llmops.config import business_path from llmops.base import CliRequest LESSONS = """ Lessons: 1. valid parameters ```json "/clusters/{tidbNodeGroup.clusterId}/tidbNodeGroups/{tidbNodeGroup.tidbNodeGroupId}": { "patch": { "summary": "Update a TiDB Node Group", "operationId": "TidbNodeGroupService_UpdateTidbNodeGroup", "responses": { "200": { "description": "A successful response.", "schema": { "$ref": "#/definitions/dedicatedv1beta1TidbNodeGroup" } }, "400": { "description": "Bad Request", "schema": { "$ref": "#/definitions/googlerpcStatus" } } "parameters": [ { "name": "tidbNodeGroup.clusterId", "in": "path", "required": true, "type": "string" }, { "name": "tidbNodeGroup.tidbNodeGroupId", "in": "path", "required": true, "type": "string" }, { "name": "tidbNodeGroup", "in": "body", "required": true, "schema": { "type": "object", "properties": { "displayName": { "type": "string" }, "nodeCount": { "type": "integer", "format": "int32", "x-nullable": true } } } } ] } ``` The valid body is: ```json { "displayName": "test", "nodeCount": 1 } ``` not: ```json { "tidbNodeGroup": { "displayName": "test", "nodeCount": 1 } } ``` """ class SutAttr(BaseModel): region: str cloud_provider: str project_id: Optional[str] = None org_id: Optional[str] = None # environment: Optional[str] = 'dev' BUSINESS_PATH = business_path / "tidbcloud_dedicated_sut" / "configs" / "sut" class TidbCloudDedicatedExpert: def __init__( self, swagger_path: str, executor: CompoundExecutor, sut_attr: SutAttr ): """ Args: swagger_path: OpenAPI/Swagger file path executor: CompoundExecutor instance sut_attr: SUT attribute """ self.abs_swagger_path = swagger_path self.executor = executor self.sut_attr = sut_attr if sut_attr.cloud_provider == "azure": cloud_cli_spec_path = BUSINESS_PATH / "azure_cli.json" elif sut_attr.cloud_provider == "aws": cloud_cli_spec_path = BUSINESS_PATH / "aws_cli.json" elif sut_attr.cloud_provider == "gcp": cloud_cli_spec_path = BUSINESS_PATH / "gcp_cli.json" else: cloud_cli_spec_path = None try: with open(self.abs_swagger_path, "r") as f: self.spec = json.load(f) if cloud_cli_spec_path: with open(cloud_cli_spec_path, "r") as f: self.cloud_cli_spec = json.load(f) with open(BUSINESS_PATH / "mysqlsh.json", "r") as f: self.mysqlsh_spec = json.load(f) except ValueError as e: raise ValueError(f"Failed to load OpenAPI/Swagger file: {e}") @property def operation_ids(self) -> List[str]: openapi_operation_ids = self._openapi_operation_ids() cloud_cli_operation_ids = self._cloud_cli_operation_ids() mysqlsh_operation_ids = self._mysqlsh_operation_ids() return openapi_operation_ids + cloud_cli_operation_ids + mysqlsh_operation_ids @property def extract_ids_and_descriptions(self) -> Dict[str, str]: openapi_id_descriptions = self._extract_openapi_ids_and_descriptions() cloud_cli_id_descriptions = self._extract_cloud_cli_ids_and_descriptions() mysqlsh_id_descriptions = self._extract_mysqlsh_ids_and_descriptions() return openapi_id_descriptions | cloud_cli_id_descriptions | mysqlsh_id_descriptions def _openapi_operation_ids(self) -> List[str]: """获取所有operationId的列表""" operation_ids = [] for path in self.spec.get("paths", {}).values(): for operation in path.values(): if "operationId" in operation: operation_ids.append(operation["operationId"]) return operation_ids def _extract_openapi_ids_and_descriptions(self) -> Dict[str, str]: """获取所有operationId及其描述的映射""" id_descriptions = {} for path in self.spec.get("paths", {}).values(): for operation in path.values(): if "operationId" in operation: id_descriptions[operation["operationId"]] = { "description": operation.get("summary", ""), "type": "TiDB Cloud Dedicated OpenAPI", } return id_descriptions def _cloud_cli_operation_ids(self) -> List[str]: return list(self.cloud_cli_spec.keys()) def _mysqlsh_operation_ids(self) -> List[str]: return list(self.mysqlsh_spec.keys()) def _extract_cloud_cli_ids_and_descriptions(self) -> Dict[str, str]: """获取所有operationId及其描述的映射""" id_descriptions = {} for _id, attr in self.cloud_cli_spec.items(): id_descriptions[_id] = { "description": attr.get("description", ""), "type": f"{self.sut_attr.cloud_provider}_cli", } return id_descriptions def _extract_mysqlsh_ids_and_descriptions(self) -> Dict[str, str]: """获取所有operationId及其描述的映射""" id_descriptions = {} for _id, attr in self.mysqlsh_spec.items(): id_descriptions[_id] = { "description": attr.get("description", ""), "type": "mysqlsh", } return id_descriptions def _collect_refs(self, obj: Any, refs: Set[str]): """递归收集对象中的所有$ref引用 Args: obj: 要检查的对象 refs: 收集到的引用集合 """ if isinstance(obj, dict): if "$ref" in obj: refs.add(obj["$ref"]) for value in obj.values(): self._collect_refs(value, refs) elif isinstance(obj, list): for item in obj: self._collect_refs(item, refs) def _get_schema_name(self, ref: str) -> str: """从$ref字符串中提取schema名称""" return ref.split("/")[-1] def _extract_required_schemas(self, refs: Set[str]) -> Dict[str, Any]: """提取所需的schemas Args: refs: 收集到的$ref引用集合 Returns: 包含所需schemas的字典 """ schemas = {} processed_refs = set() while refs: ref = refs.pop() if ref in processed_refs: continue processed_refs.add(ref) if not ref.startswith("#/definitions/"): continue schema_name = self._get_schema_name(ref) schema = self.spec["definitions"].get(schema_name) if schema: schemas[schema_name] = schema # 收集这个schema中的新引用 new_refs = set() self._collect_refs(schema, new_refs) # 添加新的未处理的引用 refs.update(new_refs - processed_refs) if "googlerpcStatus" in schemas: schemas.pop("googlerpcStatus") if "protobufAny" in schemas: schemas.pop("protobufAny") return schemas def _list_available_nodespecs(self): # TODO: to be deleted and implement in resource manager # HACK: compatibility with old web api if isinstance(self.sut_attr.org_id, str) and len(self.sut_attr.org_id) > 0: path = f"/regions/{self.sut_attr.cloud_provider}-{self.sut_attr.region}/nodeSpecs?pageSize=100&orgId={self.sut_attr.org_id}&projectId={self.sut_attr.project_id}" else: path = f"/regions/{self.sut_attr.cloud_provider}-{self.sut_attr.region}/nodeSpecs?pageSize=100" list_nodespecs_request = { "method": "GET", "path": path, } nodespecs_response = self.executor.execute_http_request(**list_nodespecs_request) nodespecs = nodespecs_response.json.get("nodeSpecs") return nodespecs def extract_operation(self, operation_id: str) -> Dict[str, Any]: if self.mysqlsh_spec.get(operation_id) is not None: cli_spec = { operation_id: { "specification": self.mysqlsh_spec[operation_id], "cli_type": "mysqlsh" } } return cli_spec if self.cloud_cli_spec.get(operation_id) is None: return self._extract_openapi_operation(operation_id) cli_spec = { operation_id: { "specification": self.cloud_cli_spec[operation_id], "cli_type": f"{self.sut_attr.cloud_provider}_cli" } } return cli_spec def _extract_openapi_operation(self, operation_id: str) -> Dict[str, Any]: """提取指定operationId的操作信息和相关schemas Args: operation_id: 要提取的操作ID Returns: Dict包含两部分: - paths: 包含该操作的路径信息 - components: 包含该操作依赖的所有schemas """ # HACK: inject region # 查找操作 operation_path = None operation_method = None operation_info = None for path, methods in self.spec.get("paths", {}).items(): for method, operation in methods.items(): if operation.get("operationId") == operation_id: operation_path = path operation_method = method operation_info = operation break if operation_info: break if not operation_info: raise ValueError(f"Operation {operation_id} not found") # 收集所有引用 refs: Set[dict] = set() self._collect_refs(operation_info, refs) # 构建返回结果 result = { "paths": {operation_path: {operation_method: operation_info}}, "components": {"schemas": self._extract_required_schemas(refs)}, } return result # def list_all_resources(self) -> Dict[str, Any]: # """mamually list all resources""" # resources: Dict[str, Any] = { # "clusters": {}, # "tidbnodegroups": {}, # "private_endpoint_connections": {}, # "public_endpoint_setting": {}, # "private_link_services": {}, # "vpcs": {}, # "vpc_endpoints": {}, # } # list_clusters_request = { # "method": "GET", # "path": f"/clusters?projectId={self.sut_attr.project_id}", # "auth_type": "digest", # } # clusters_response = self.executor.execute_http_request(**list_clusters_request) # clusters = clusters_response.json.get("clusters") # for cluster in clusters: # if cluster["displayName"].startswith("llm-e2e-cluster"): # # if not cluster["displayName"].startswith("v1beta1-dedicated-e2e-") and not cluster["displayName"].startswith("manual-scenario"): # # or use createdBy to filter # resources["clusters"][cluster["clusterId"]] = {} # resources["clusters"][cluster["clusterId"]]["attributes"] = cluster # resources["clusters"][cluster["clusterId"]]["state"] = cluster.get("state") # resources["clusters"][cluster["clusterId"]]["references"] = [] # resources["clusters"][cluster["clusterId"]]["parent_id"] = None # list_nodegroups_request = { # "method": "GET", # "path": f"/clusters/{cluster['clusterId']}/tidbNodeGroups", # "auth_type": "digest", # } # nodegroups_response = self.executor.execute_http_request( # **list_nodegroups_request # ) # nodegroups = nodegroups_response.json.get("tidbNodeGroups", []) # for nodegroup in nodegroups: # resources["tidbnodegroups"][nodegroup["tidbNodeGroupId"]] = {} # resources["tidbnodegroups"][nodegroup["tidbNodeGroupId"]][ # "attributes" # ] = nodegroup # resources["tidbnodegroups"][nodegroup["tidbNodeGroupId"]][ # "state" # ] = nodegroup.get("state") # resources["tidbnodegroups"][nodegroup["tidbNodeGroupId"]][ # "references" # ] = [] # resources["tidbnodegroups"][nodegroup["tidbNodeGroupId"]][ # "parent_id" # ] = cluster["clusterId"] # # public endpoint setting # public_endpoint_setting_request = { # "method": "GET", # "path": f"/clusters/{cluster['clusterId']}/tidbNodeGroups/{nodegroup['tidbNodeGroupId']}/publicEndpointSetting", # "auth_type": "digest", # } # public_endpoint_setting_response = self.executor.execute_http_request( # **public_endpoint_setting_request # ) # public_endpoint_setting = public_endpoint_setting_response.json # resources["public_endpoint_setting"] = { # "attributes": public_endpoint_setting, # "references": [], # "parent_id": nodegroup["tidbNodeGroupId"], # } # # private_link_services # list_private_link_services_request = { # "method": "GET", # "path": f"/clusters/{cluster['clusterId']}/tidbNodeGroups/{nodegroup['tidbNodeGroupId']}/privateLinkService", # "auth_type": "digest", # } # private_link_services_response = self.executor.execute_http_request( # **list_private_link_services_request # ) # private_link_service = private_link_services_response.json # resources["private_link_services"] = { # "attributes": private_link_service, # "references": [], # "parent_id": nodegroup["tidbNodeGroupId"], # } # # private endpoint connections # list_endpoint_connections_request = { # "method": "GET", # "path": f"/clusters/{cluster['clusterId']}/tidbNodeGroups/{nodegroup['tidbNodeGroupId']}/privateEndpointConnections", # "auth_type": "digest", # } # endpoint_connections_response = self.executor.execute_http_request( # **list_endpoint_connections_request # ) # endpoint_connections = endpoint_connections_response.json.get( # "privateEndpointConnections", [] # ) # for endpoint in endpoint_connections: # resources["private_endpoint_connections"][ # endpoint["privateEndpointConnectionId"] # ] = {} # resources["private_endpoint_connections"][ # endpoint["privateEndpointConnectionId"] # ]["attributes"] = endpoint # resources["private_endpoint_connections"][ # endpoint["privateEndpointConnectionId"] # ]["state"] = endpoint.get("state") # resources["private_endpoint_connections"][ # endpoint["privateEndpointConnectionId"] # ]["references"] = [] # resources["private_endpoint_connections"][ # endpoint["privateEndpointConnectionId"] # ]["parent_id"] = nodegroup["tidbNodeGroupId"] # cloud_resources = self.list_cloud_resources() # return dict( # clusters=resources["clusters"], # tidbnodegroups=resources["tidbnodegroups"], # private_endpoint_connections=resources["private_endpoint_connections"], # public_endpoint_setting=resources["public_endpoint_setting"], # private_link_services=resources["private_link_services"], # vpcs=cloud_resources["vpcs"], # vpc_endpoints=cloud_resources["vpc_endpoints"], # ) def list_all_resources(self): """mamually initialize a list of resource types""" custom_vpc = { "aws": { "vpc-0393f150b4eb1a955": { "attributes": { "OwnerId": "024848481520", "InstanceTenancy": "default", "CidrBlockAssociationSet": [ { "AssociationId": "vpc-cidr-assoc-0c6dbca23e8325b3c", "CidrBlock": "172.31.0.0/16", "CidrBlockState": { "State": "associated" } } ], "IsDefault": True, "BlockPublicAccessStates": { "InternetGatewayBlockMode": "off" }, "VpcId": "vpc-0393f150b4eb1a955", "State": "available", "CidrBlock": "172.31.0.0/16", "DhcpOptionsId": "dopt-0a7b733856ff5db55" }, "references": [], "state": "available" } }, "gcp": {}, "azure": { "/subscriptions/85b598b9-a080-4476-bc96-ba89e64f92d1/resourceGroups/openapi-test/providers/Microsoft.Network/virtualNetworks/openapi-test-vnet/subnets/openapi-test-subnet": { "attributes": { "addressPrefix": "10.0.1.0/24", "delegations": [], "etag": "W/\"cfe220db-8483-4dc3-a7c5-f29f864d8479\"", "id": "/subscriptions/85b598b9-a080-4476-bc96-ba89e64f92d1/resourceGroups/openapi-test/providers/Microsoft.Network/virtualNetworks/openapi-test-vnet/subnets/openapi-test-subnet", "name": "openapi-test-subnet", "privateEndpointNetworkPolicies": "Disabled", "privateLinkServiceNetworkPolicies": "Enabled", "provisioningState": "Succeeded", "resourceGroup": "openapi-test", "type": "Microsoft.Network/virtualNetworks/subnets", }, "state": "Succeeded", "references": [], "parent_id": "/subscriptions/85b598b9-a080-4476-bc96-ba89e64f92d1/resourceGroups/openapi-test/providers/Microsoft.Network/virtualNetworks/openapi-test-vnet" } } } return { "clusters": {}, "tidbnodegroups": {}, "private_endpoint_connections": {}, "public_endpoint_setting": {}, "private_link_services": {}, "vpcs": custom_vpc[self.sut_attr.cloud_provider], "vpc_endpoints": {}, } # def list_cloud_resources(self) -> Dict[str, Any]: # request_type = f"{self.sut_attr.cloud_provider}_cli" # vpcs = {} # vpc_endpoints = {} # if request_type == "aws_cli": # list_vpcs_request = CliRequest( # service="ec2", # command="aws", # args=["ec2", "describe-vpcs", "--region", self.sut_attr.region], # args_params="{}", # ) # vpcs_response = self.executor.execute_cli_with_retry( # request_type=request_type, # request=list_vpcs_request, # ) # for vpc in json.loads(vpcs_response.json["output"])["Vpcs"]: # vpcs[vpc["VpcId"]] = { # "attributes": vpc, # "references": [], # "state": vpc["State"], # } # list_vpc_endpoints_request = CliRequest( # service="ec2", # command="aws", # args=["ec2", "describe-vpc-endpoints", "--region", self.sut_attr.region], # args_params="{}", # ) # vpc_endpoints_response = self.executor.execute_cli_with_retry( # request_type=request_type, # request=list_vpc_endpoints_request, # ) # for vpc_endpoint in json.loads(vpc_endpoints_response.json["output"])["VpcEndpoints"]: # vpc_endpoints[vpc_endpoint["VpcEndpointId"]] = { # "attributes": vpc_endpoint, # "references": [], # "state": vpc_endpoint["State"], # } # elif request_type == "azure_cli": # list_subnets_request = CliRequest( # service="network", # command="az", # args=["network", "vnet", "subnet", "list", "-g", "openapi-test", "--vnet-name", "openapi-test-vnet"], # args_params="{}", # ) # subnets_response = self.executor.execute_cli_with_retry( # request_type=request_type, # request=list_subnets_request, # ) # for subnet in json.loads(subnets_response.json["output"]): # vpcs[subnet["id"]] = { # "attributes": subnet, # "references": [], # "state": subnet["provisioningState"], # } # list_vpc_endpoints_request = CliRequest( # service="network", # command="az", # args=["network", "private-endpoint", "list", "-g", "openapi-test"], # args_params="{}", # ) # vpc_endpoints_response = self.executor.execute_cli_with_retry( # request_type=request_type, # request=list_vpc_endpoints_request, # ) # for vpc_endpoint in json.loads(vpc_endpoints_response.json["output"]): # dns_ip_request = CliRequest( # service="network", # command="az", # args=["network", "nic", "show", "--ids", vpc_endpoint["networkInterfaces"][0]["id"]], # args_params="{}", # ) # dns_ip_response = self.executor.execute_cli_with_retry( # request_type=request_type, # request=dns_ip_request, # ) # vpc_endpoint["privateIpAddress"] = json.loads(dns_ip_response.json["output"])["ipConfigurations"][0]["privateIPAddress"] # vpc_endpoints[vpc_endpoint["id"]] = { # "attributes": vpc_endpoint, # "references": [], # "state": vpc_endpoint["provisioningState"], # } # elif request_type == "gcp_cli": # raise NotImplementedError("gcp_cli not implemented yet") # return { # "vpcs": vpcs, # "vpc_endpoints": vpc_endpoints, # } # def cleanup(self): # """cleanup resources # currently an empty implementation, can be extended later # """ # print("start to clean up resources") # current_state = self.list_all_resources() # print( # "length of current_state: ", # len(current_state["clusters"]), # ) # for cluster_id, cluster in current_state["clusters"].items(): # if cluster["attributes"]["displayName"].startswith("llm-e2e-cluster"): # # if not cluster["attributes"]["displayName"].startswith("v1beta1-dedicated-e2e-") and not cluster["attributes"]["displayName"].startswith("manual-scenario"): # self.executor.execute_http_request( # method="DELETE", path=f"/clusters/{cluster_id}", auth_type="digest" # ) # # clean up vpc_endpoints # for vpc_endpoint, _ in current_state["vpc_endpoints"].items(): # if self.sut_attr.cloud_provider == "aws": # delete_vpc_endpoints_request = CliRequest( # service="ec2", # command="aws", # args=["ec2", "delete-vpc-endpoints", "--vpc-endpoint-ids", vpc_endpoint, "--region", self.sut_attr.region], # args_params="{}", # ) # self.executor.execute_cli_with_retry( # request_type=f"{self.sut_attr.cloud_provider}_cli", # request=delete_vpc_endpoints_request, # ) # elif self.sut_attr.cloud_provider == "azure": # delete_vpc_endpoints_request = CliRequest( # service="network", # command="az", # args=["network", "private-endpoint", "delete", "--ids", vpc_endpoint], # args_params="{}", # ) # self.executor.execute_cli_with_retry( # request_type=f"{self.sut_attr.cloud_provider}_cli", # request=delete_vpc_endpoints_request, # ) # elif self.sut_attr.cloud_provider == "gcp": # raise NotImplementedError("gcp_cli not implemented yet") # print("resources cleanup completed") @property def biz_system_prompt_preprocessor(self) -> str: """general biz logic for preprocess""" return """Lessons: - scale in/out tikv or scale up/down tikv/tidb should choose ClusterService_UpdateCluster. """ @property def biz_system_prompt_operation_selector(self) -> str: """general biz logic for operation_selector""" return """TiDB Cloud Resource Specifications: Resource Types and Dependencies: - clusters: Root resource, no parent-type dependencies - tidbNodeGroups: Child of clusters - privateEndpointConnections: Child of tidbNodeGroups, references VPC State Transitions: - Transitional States: * CREATING * MODIFYING * RESUMING * PAUSING etc. Resource-Specific Rules: - If the target resource in a transitional state, the next operation should be a `GET` method operation related to the resource. LESSONS: - Scale in/out tikv or scale up/down tikv should choose ClusterService_UpdateCluster. - To add (NOT update) public endpoint, you have to get the current public endpoint first. So you should TidbNodeGroupService_GetPublicEndpointSetting first. - To create a private endpoint connection, you have to create an endpoint with an operation in a cloud cli type first. - The 'pendingAcceptance' state of a VPC endpoint indicates that the endpoint connection is ready to be created. - If the target operation is to create a private endpoint, you must first get private link service's state to be 'ACTIVE'. """ @property def biz_system_prompt_request_generator(self) -> str: """general biz logic for request_generator""" return f"""TiDB Cloud Request Requirements: - TiDB Cloud API is based on Swagger 2.0. - Available roles are {list(self.executor.api_token.keys())}. If 'target_operation' mentions a role explicitly, please use the closest role in the list. If there is only one available role and it's not mentioned in 'target_operation', please use it. If multiple available roles and 'target_operation' doesn't mention any role explicitly, please use the role in your last successful trial in `history`. - If the state of the target resource is transitional like "CREATING" or "MODIFYING", and the `target_operation` is "GET", please follow it. - If a `region` field is required, please set it as "{self.sut_attr.cloud_provider}-{self.sut_attr.region}". - If a `project_id` or `projectId` field is required, please set it as "{self.sut_attr.project_id}". Especially when creating a new cluster, put `"tidb.cloud/project": "{self.sut_attr.project_id}"` in the "labels" field. - If a `org_id` or `orgId` field is required, please set it as "{self.sut_attr.org_id}". - Available nodespecs for tidb/tikv/tiflash: {self._list_available_nodespecs()}. If not specified in `target_operation`, please use the smallest available nodespec. - If TiFlash is not mentioned in the `target_operation`, there's no need to add this component - it's not essential. - Leave `expect` field empty unless there is a field to describe the state like `state` or `status`. - The ID of TiDB NodeGroup may exist in the `attributes` of 'clusters' resource. - CLI type: {self.sut_attr.cloud_provider}_cli. - CLI examples: - {{"request": {{"command": "aws", "args": ["ec2", "describe-vpcs", "--region", "us-west-1"]}}, "request_type": "aws_cli", "description": "", "expect": ""}} - {{"request": {{"command": "gcloud", "args": ["compute", "describe", "regions", "us-west1"]}}, "request_type": "gcloud_cli", "description": "", "expect": ""}} - {{"request": {{"command": "az", "args": ["vm", "list", "--location", "westus"]}}, "request_type": "azure_cli", "description": "", "expect": ""}} {LESSONS} 2. "add 0.0.0.0 to pulic endpiont IP Address list" indicates literally ADD 0.0.0.0 to the existing IP Address list by TidbNodeGroupService_GetPublicEndpointSetting other than replacing the existing IP Address list to [0.0.0.0]. 3. if the operation is "az network private-endpoint create", add the parameter `--manual-request 1` to the end of the operation. 4. if the operation is "aws ec2 create-vpc-endpoint", don't use parameters of `subnet-ids`, `security-group-ids`, `tag-specifications` """ @property def biz_system_prompt_response_analyzer(self) -> str: """general biz logic for preprocess""" return f"""TiDB Cloud Response Characteristics: - rootPassword won't be shown in response even if succeeded, only check status_code for the changing password operation. - Most attributes in the `target_operation` may not be taken into consideration, such as `displayName`, etc. - If some states is different between `prev_state` and `current_state`, the GET operation is a required and necessary operation. {LESSONS} """ @property def biz_system_prompt_target_matcher(self) -> str: """general biz logic for preprocess""" return """Important fields to compare (may not be top-level fields, could be nested under other fields): - Node specifications (e.g. node_size, instance_type) - Node count and configuration (e.g. node_quantity, replicas) - Storage settings (e.g. storage_size_gib) - rootPassword will not be shown in response even if succeeded, only check status_code for the changing password operation. Fields that do not need comparison: - Display names or labels (e.g. displayName, description) - Metadata fields (e.g. created_at, updated_at) - Internal identifiers (e.g. cluster_id, request_id) Tips: - Some APIs may not be well-designed, so you need to use additional information to determine if the current operation is a target operation. For example, if the target operation is to delete a TiDB Nodegroup, the request might be a POST or DELETE. When it's a POST, a possible way to confirm is if the ID in the request no longer exists in the response. """ @property def biz_system_prompt_resource_manager(self) -> str: return """In Dedicate SUT, the nodegroup whose id is "0" is not a valid nodegroup. Don't put it in the resource inventory.""" if __name__ == "__main__": from pathlib import Path from environs import Env current_path = Path(__file__).parent config_path = current_path / "configs/sut" task_path = current_path / "tasks" env = Env() env.read_env(config_path / ".env") api_token = env.str("SUT_TOKEN") executor = CompoundExecutor( base_url="https://dedicated.staging.tidbapi.com/v1beta1", api_token=api_token, auth_type="digest", ) sut_attr = SutAttr( region="eastus2", cloud_provider="azure", project_id="1369847559694040595", ) tidbcloud_expert = TidbCloudDedicatedExpert( config_path / "openapi.json", executor, sut_attr ) # print(tidbcloud_expert.list_cloud_resources()) # print(tidbcloud_expert._list_available_nodespecs()) resources = tidbcloud_expert.list_all_resources() print("resources: ", resources) # print(f"operation id length: {len(tidbcloud_expert.operation_ids)}") # print(f"operation description length: {len(tidbcloud_expert.extract_ids_and_descriptions)}") # print(tidbcloud_expert.extract_ids_and_descriptions.get("ClusterService_PauseCluster")) # print(tidbcloud_expert.extract_ids_and_descriptions.get("ec2 replace-route")) # print(tidbcloud_expert.extract_operation("ClusterService_PauseCluster")) # print("len: ", len(tidbcloud_expert.operation_ids), "operation ids: ", tidbcloud_expert.operation_ids) # print("len: ", len(tidbcloud_expert.extract_ids_and_descriptions), "operation descriptions: ", tidbcloud_expert.extract_ids_and_descriptions) # print( # "operation: ", # json.dumps( # tidbcloud_expert.extract_operation("ClusterService_CreateCluster"), # indent=2, # ), # ) # print( # "operation: ", # json.dumps( # tidbcloud_expert.extract_operation("ClusterService_PauseCluster"), # indent=2, # ), # ) # tidbcloud_expert.cleanup()