This repository has been archived by the owner on Jun 14, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 17
/
rds.py
182 lines (155 loc) · 6.53 KB
/
rds.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
"""Library for aws Relational Database Service (sts) methods"""
import logging
from typing import List, Optional
from functools import lru_cache
import boto3
import botocore
from mypy_boto3_ec2.client import EC2Client
from mypy_boto3_ec2.type_defs import InstanceTypeInfoTypeDef
from mypy_boto3_iam.client import IAMClient
from mypy_boto3_iam.type_defs import (
EvaluationResultTypeDef,
SimulatePolicyResponseTypeDef,
)
from mypy_boto3_rds.client import RDSClient
from mypy_boto3_rds.type_defs import DBInstanceTypeDef
from mypy_boto3_sts.client import STSClient
from mypy_boto3_sts.type_defs import CredentialsTypeDef
from mypy_boto3_cloudwatch.client import CloudWatchClient
from driver.aws.exceptions import InvalidCustomerSettingsError
@lru_cache
def get_db_instance_info(
db_instance_identifier: str, client: RDSClient
) -> DBInstanceTypeDef:
"""
Ensures that we can connect to the target AWS RDS instance by calling describe_db_instances.
Returns:
The DBInstance object returned by AWS
Raises:
botocore.exceptions.ClientError: If we receive an error from AWS
InvalidCustomerSettingsError: If customer supplied an invalid db_instance_identifier
InvalidPermissionError: If it's not allowed to describe db instances
DBInstanceNotFound: If the database instance is not found
"""
resp = client.describe_db_instances(DBInstanceIdentifier=db_instance_identifier)
if len(resp["DBInstances"]) == 0:
raise InvalidCustomerSettingsError(
"No instance was found for provided db identifier"
)
if len(resp["DBInstances"]) > 1:
raise InvalidCustomerSettingsError(
"Multiple instances found for " "provided db identifier. Expected only one"
)
return resp["DBInstances"][0]
def get_db_hostname(db_instance_identifier: str, client: RDSClient) -> str:
"""
Ensures that we can connect to the target AWS RDS instance by calling describe_db_instances.
Returns:
The DBInstance object returned by AWS
Raises:
botocore.exceptions.ClientError: If we receive an error from AWS
InvalidCustomerSettingsError: If customer supplied an invalid db_instance_identifier
InvalidPermissionError: If it's not allowed to describe db instances
DBInstanceNotFound: If the database instance is not found
"""
instance_info = get_db_instance_info(db_instance_identifier, client)
return instance_info["Endpoint"]["Address"]
def get_db_cluster_identifier(db_instance_identifier: str, client: RDSClient) -> str:
"""
Get the cluster identifier for the given db instance if it's a member of a cluster
Returns:
The DBInstance object returned by AWS
Raises:
botocore.exceptions.ClientError: If we receive an error from AWS
InvalidCustomerSettingsError: If customer supplied an invalid db_instance_identifier
InvalidPermissionError: If it's not allowed to describe db instances
DBInstanceNotFound: If the database instance is not found
"""
instance_info = get_db_instance_info(db_instance_identifier, client)
return (
instance_info["DBClusterIdentifier"]
if "DBClusterIdentifier" in instance_info
else ""
)
def get_db_port(db_instance_identifier: str, client: RDSClient) -> str:
"""
Ensures that we can connect to the target AWS RDS instance by calling describe_db_instances.
Returns:
The DBInstance object returned by AWS
Raises:
botocore.exceptions.ClientError: If we receive an error from AWS
InvalidCustomerSettingsError: If customer supplied an invalid db_instance_identifier
InvalidPermissionError: If it's not allowed to describe db instances
DBInstanceNotFound: If the database instance is not found
"""
instance_info = get_db_instance_info(db_instance_identifier, client)
return instance_info["Endpoint"]["Port"]
def get_db_version(db_instance_identifier: str, client: RDSClient) -> str:
"""
Get's database version information
"""
instance_info = get_db_instance_info(db_instance_identifier, client)
return instance_info["EngineVersion"].replace(".", "_").replace("-", "_")
def get_db_type(db_instance_identifier: str, client: RDSClient) -> str:
"""
Get's database type information
"""
instance_info = get_db_instance_info(db_instance_identifier, client)
db_type = instance_info["Engine"].replace(".", "_").replace("-", "_")
if db_type == "aurora": # for aurora mysql 5.6
db_type = "aurora_mysql"
return db_type
def get_db_parameter_group_name(db_instance_identifier: str, client: RDSClient) -> str:
"""
Get database parameter group name
"""
db_parameter_group_name = ""
instance_info = get_db_instance_info(db_instance_identifier, client)
parameter_groups = instance_info.get("DBParameterGroups")
if parameter_groups:
db_parameter_group_name = parameter_groups[0].get("DBParameterGroupName")
if not db_parameter_group_name:
logging.warning(
"RDS client: Unable to get parameter group name for instance %s",
db_instance_identifier,
)
return db_parameter_group_name
def get_db_non_default_parameters(
db_instance_identifier: str, client: RDSClient
) -> List[str]:
"""
Get list of database parameters that are set by the user
"""
db_non_default_parameters = []
db_parameter_group_name = get_db_parameter_group_name(
db_instance_identifier, client
)
if db_parameter_group_name:
db_parameters = []
try:
response = client.describe_db_parameters(
DBParameterGroupName=db_parameter_group_name
)
db_parameters = response["Parameters"]
except KeyError as ex:
logging.warning(
"RDS client: Unable to collect parameters for Parameter Group Name %s",
db_parameter_group_name,
)
db_non_default_parameters = [
parameter.get("ParameterName")
for parameter in db_parameters
if parameter.get("Source") == "user" and "ParameterName" in parameter
]
else:
logging.warning(
"RDS client: Cannot fetch parameters without parameter group name for instance %s",
db_instance_identifier,
)
return db_non_default_parameters
def get_db_auth_token(
db_username: str, hostname: str, port: int, client: RDSClient
) -> str:
return client.generate_db_auth_token(
DBHostname=hostname, Port=port, DBUsername=db_username
)