-
Notifications
You must be signed in to change notification settings - Fork 2
/
main.py
195 lines (167 loc) · 6.71 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
# Copyright(C) 2022 PingCAP. All Rights Reserved.
"""
Purpose
Shows how to create a TiDB cluster and get cluster's detail.
"""
import json
import os
import time
import requests
from requests.auth import HTTPDigestAuth
# Basic config
HOST = "https://api.tidbcloud.com"
class CreateDeveloperCluster:
def __init__(self):
"""
Generate authorization by public key and private key (https://tidbcloud.com/console/clusters)
"""
self.digest_auth = _authorization()
def get_all_projects(self):
"""
Get all projects
:return: Projects detail
"""
url = f"{HOST}/api/v1beta/projects"
resp = requests.get(url=url, auth=self.digest_auth)
print(f"Method: {resp.request.method}, Request: {url}")
return _response(resp)
def get_provider_regions_specifications(self):
"""
Get cloud providers, regions and available specifications.
:return: List the cloud providers, regions and available specifications.
"""
url = f"{HOST}/api/v1beta/clusters/provider/regions"
resp = requests.get(url=url, auth=self.digest_auth)
print(f"Method: {resp.request.method}, Request: {url}")
return _response(resp)
def create_developer_cluster(self, project_id: str, region: str) -> dict:
"""
Create a developer cluster in your specified project
:param project_id: The project id
:param region: Available region for developer cluster
:return: The developer cluster id
"""
url = f"{HOST}/api/v1beta/projects/{project_id}/clusters"
ts = int(time.time())
data_config = \
{
"name": f"tidbcloud-sample-{ts}",
"cluster_type": "DEVELOPER",
"cloud_provider": "AWS",
"region": f"{region}",
"config":
{
"root_password": "input_your_password",
"ip_access_list":
[
{
"cidr": "0.0.0.0/0",
"description": "Allow Access from Anywhere."
}
]
}
}
data_config_json = json.dumps(data_config)
resp = requests.post(url=url,
auth=self.digest_auth,
data=data_config_json)
print(f"Method: {resp.request.method}, Request: {url}, Payload: {data_config_json}")
return _response(resp)
def get_cluster_by_id(self, project_id: str, cluster_id: str) -> dict:
"""
Get the cluster detail
You will get `connection_strings` from the response after the cluster's status is`AVAILABLE`.
Then, you can connect to TiDB using the default user, host, and port in `connection_strings`
:param project_id: The project id
:param cluster_id: The cluster id
:return: The cluster detail
"""
resp = requests.get(url=f"{HOST}/api/v1beta/projects/{project_id}/clusters/{cluster_id}",
auth=self.digest_auth)
return _response(resp)
def delete_cluster(self, project_id: str, cluster_id: str) -> dict:
"""
Delete the cluster
:param project_id: The project id
:param cluster_id: The cluster id
:return: Result for deletion
"""
url = f"{HOST}/api/v1beta/projects/{project_id}/clusters/{cluster_id}"
resp = requests.delete(url=url,
auth=self.digest_auth)
print(f"Method: {resp.request.method}, Request: {url}")
return _response(resp)
def _authorization():
"""
Generate authorization by public key and private key (https://tidbcloud.com/console/clusters)
:return: Digest auth
"""
public_key = os.environ.get("TIDBCLOUD_PUBLIC_KEY", None)
private_key = os.environ.get("TIDBCLOUD_PRIVATE_KEY", None)
if public_key is None or private_key is None:
print("TIDBCLOUD_PUBLIC_KEY or TIDBCLOUD_PRIVATE_KEY is None, you should set TIDBCLOUD_PUBLIC_KEY and "
"TIDBCLOUD_PRIVATE_KEY firstly.")
raise Exception("TIDBCLOUD_PUBLIC_KEY or TIDBCLOUD_PRIVATE_KEY not set.")
return HTTPDigestAuth(public_key, private_key)
def _response(resp: requests.models.Response) -> dict:
"""
Response from open api
:param resp: Result from the API
:return: Format response
"""
if resp.status_code != 200:
print(f"request invalid, code : {resp.status_code}, message : {resp.text}")
raise Exception(f"request invalid, code : {resp.status_code}, message : {resp.text}")
print(f"response : {resp.text}")
return resp.json()
def _get_developer_region(provider_region_specifications: dict) -> str:
"""
Get a available region
:param provider_region_specifications: Result of cloud providers, regions and available specifications
:return: A available region
"""
try:
items = provider_region_specifications["items"]
for item in items:
if item["cluster_type"] == "DEVELOPER":
region = item["region"]
return region
except KeyError:
print("developer region not found!")
raise
raise Exception("developer region not found!")
def usage_demo():
print("-" * 88)
print("Welcome to the TiDB Cloud API samples!")
print("-" * 88)
create_cluster = CreateDeveloperCluster()
print("1. Get all projects. ")
projects = create_cluster.get_all_projects()
try:
items = projects["items"]
if len(items) > 0:
sample_project_id = items[0]["id"]
except KeyError:
print("project id not found!")
raise
print()
print("2. Get the cloud providers, regions and available specifications")
provider_region_specifications = create_cluster.get_provider_regions_specifications()
region = _get_developer_region(provider_region_specifications)
print()
print(f"3. Create a developer cluster in your specified project ( project id : {sample_project_id} ). ")
cluster = create_cluster.create_developer_cluster(sample_project_id, region)
try:
sample_cluster_id = cluster["id"]
except KeyError:
print("cluster id not found!")
print()
print(f"4. Get the new cluster ( cluster id : {sample_cluster_id} ) detail. ")
create_cluster.get_cluster_by_id(sample_project_id, sample_cluster_id)
print()
# print("If necessary , delete the cluster.")
# create_cluster.delete_cluster(sample_project_id, sample_cluster_id)
print("Thanks for watching! ")
print("-" * 88)
if __name__ == "__main__":
usage_demo()