import boto3
from botocore.exceptions import ClientError
def get_secrets():
secret_name = "your_secret_name"
region_name = "your_region"
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'] == 'ResourceNotFoundException':
print("The requested secret " + secret_name + " was not found")
elif e.response['Error']['Code'] == 'InvalidRequestException':
print("The request was invalid due to:", e)
elif e.response['Error']['Code'] == 'InvalidParameterException':
print("The request had invalid params:", e)
else:
secret = get_secret_value_response['SecretString']
return secret
import psycopg2
secret = json.loads(get_secrets())
try:
conn = psycopg2.connect(
dbname="your_database_name",
user=secret['username'],
password=secret['password'],
host=secret['host'],
port=secret['port']
)
except psycopg2.Error as e:
print('Unable to connect to the database:', e)
现在您可以使用连接对象(conn)执行查询或任何其他数据库操作。