Secrets Manager_get.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455
  1. import boto3
  2. import base64
  3. import json
  4. from botocore.exceptions import ClientError
  5. def get_secret():
  6. secret_name = "api_secret"
  7. region_name = "cn-northwest-1"
  8. session = boto3.session.Session()
  9. client = session.client(
  10. service_name='secretsmanager',
  11. region_name=region_name
  12. )
  13. try:
  14. get_secret_value_response = client.get_secret_value(
  15. SecretId=secret_name
  16. )
  17. except ClientError as e:
  18. if e.response['Error']['Code'] == 'DecryptionFailureException':
  19. raise e
  20. elif e.response['Error']['Code'] == 'InternalServiceErrorException':
  21. raise e
  22. elif e.response['Error']['Code'] == 'InvalidParameterException':
  23. raise e
  24. elif e.response['Error']['Code'] == 'InvalidRequestException':
  25. raise e
  26. elif e.response['Error']['Code'] == 'ResourceNotFoundException':
  27. raise e
  28. else:
  29. if 'SecretString' in get_secret_value_response:
  30. # secret = json.loads(get_secret_value_response['SecretString'])
  31. return json.loads(get_secret_value_response['SecretString'])
  32. else:
  33. # decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])
  34. return base64.b64decode(get_secret_value_response['SecretBinary'])
  35. # {'username': 'admin', 'password': '12345678', 'engine': 'mysql', 'host': 'database-1.ce6iux1s3ccn.rds.cn-northwest-1.amazonaws.com.cn', 'port': 3306, 'dbInstanceIdentifier': 'database-1'}
  36. get_value = get_secret()['username']
  37. username = get_value['username']
  38. password = get_value['password']
  39. host = get_value['host']
  40. port = get_value['port']