| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455 |
- import boto3
- import base64
- import json
- from botocore.exceptions import ClientError
- def get_secret():
- secret_name = "api_secret"
- region_name = "cn-northwest-1"
- session = boto3.session.Session()
- client = session.client(
- service_name='secretsmanager',
- region_name=region_name
- )
- try:
- get_secret_value_response = client.get_secret_value(
- SecretId=secret_name
- )
- except ClientError as e:
- if e.response['Error']['Code'] == 'DecryptionFailureException':
- raise e
- elif e.response['Error']['Code'] == 'InternalServiceErrorException':
- raise e
- elif e.response['Error']['Code'] == 'InvalidParameterException':
- raise e
- elif e.response['Error']['Code'] == 'InvalidRequestException':
- raise e
- elif e.response['Error']['Code'] == 'ResourceNotFoundException':
- raise e
- else:
- if 'SecretString' in get_secret_value_response:
- # secret = json.loads(get_secret_value_response['SecretString'])
- return json.loads(get_secret_value_response['SecretString'])
- else:
- # decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])
- return base64.b64decode(get_secret_value_response['SecretBinary'])
- # {'username': 'admin', 'password': '12345678', 'engine': 'mysql', 'host': 'database-1.ce6iux1s3ccn.rds.cn-northwest-1.amazonaws.com.cn', 'port': 3306, 'dbInstanceIdentifier': 'database-1'}
- get_value = get_secret()['username']
- username = get_value['username']
- password = get_value['password']
- host = get_value['host']
- port = get_value['port']
|