| 174 | |
| 175 | |
| 176 | class RDS(): |
| 177 | def __init__(self, access_key, secret_key, session_token=None, |
| 178 | default_region='ap-south-1'): |
| 179 | self._clients = {} |
| 180 | |
| 181 | self._access_key = access_key |
| 182 | self._secret_key = secret_key |
| 183 | self._session_token = session_token |
| 184 | |
| 185 | self._default_region = default_region |
| 186 | |
| 187 | ########################################################################## |
| 188 | # AWS Helper functions |
| 189 | ########################################################################## |
| 190 | def _get_aws_client(self, type): |
| 191 | """ Create/cache/return an AWS client object """ |
| 192 | if type in self._clients: |
| 193 | return self._clients[type] |
| 194 | |
| 195 | session = boto3.Session( |
| 196 | aws_access_key_id=self._access_key, |
| 197 | aws_secret_access_key=self._secret_key, |
| 198 | aws_session_token=self._session_token |
| 199 | ) |
| 200 | |
| 201 | self._clients[type] = session.client( |
| 202 | type, region_name=self._default_region) |
| 203 | |
| 204 | return self._clients[type] |
| 205 | |
| 206 | def get_available_db_version(self, engine='postgres'): |
| 207 | rds = self._get_aws_client('rds') |
| 208 | return rds.describe_db_engine_versions(Engine=engine) |
| 209 | |
| 210 | def get_available_db_instance_class(self, engine='postgres', |
| 211 | engine_version='10'): |
| 212 | rds = self._get_aws_client('rds') |
| 213 | _instances = rds.describe_orderable_db_instance_options( |
| 214 | Engine=engine, |
| 215 | EngineVersion=engine_version) |
| 216 | _instances_list = _instances['OrderableDBInstanceOptions'] |
| 217 | _marker = _instances['Marker'] if 'Marker' in _instances else None |
| 218 | while _marker: |
| 219 | _tmp_instances = rds.describe_orderable_db_instance_options( |
| 220 | Engine=engine, |
| 221 | EngineVersion=engine_version, |
| 222 | Marker=_marker) |
| 223 | _instances_list = [*_instances_list, |
| 224 | *_tmp_instances['OrderableDBInstanceOptions']] |
| 225 | _marker = _tmp_instances['Marker'] if 'Marker'\ |
| 226 | in _tmp_instances else None |
| 227 | |
| 228 | return _instances_list |
| 229 | |
| 230 | def get_db_instance(self, instance_name): |
| 231 | rds = self._get_aws_client('rds') |
| 232 | return rds.describe_db_instances( |
| 233 | DBInstanceIdentifier=instance_name) |
no outgoing calls
no test coverage detected