in reference-architectures/automated-password-rotation/terraform/code/main.py [0:0]
def verify_change_password(instance_name, db_name, location, db_user, db_pass):
"""Function to test Postgresql user password.
Args:
instance_name(string) : Cloudsql instance name
db_name(string) : postgres db user
location(string) : Cloudsql instance location
db_user(string) : posygres db user
db_pass(string) : postgres db password
Returns:
True/False(bool) : flag indicating success or failure in
verifying the password
"""
project_id = get_project_id()
instance_connection_name = project_id + ":" + location + ":" + instance_name
# Establish the connection pool
def conn_verify():
return getconn(instance_connection_name, db_user, db_pass, db_name)
pool = sqlalchemy.create_engine(
"postgresql+pg8000://",
creator=conn_verify,
)
print("Verifying the new password works by executing a sample query")
try:
with pool.connect() as db_conn:
results = db_conn.execute(
sqlalchemy.text("select * from information_schema.tables limit 1;")
).fetchall()
for row in results:
print(row)
return True
except Exception as e: # pylint: disable=broad-exception-caught
print("Cannot verify that the new password due to the following exception")
print(e)
return False