in reference-architectures/automated-password-rotation/terraform/code/main.py [0:0]
def reset_password(instance_name, db_name, location, db_user, db_pass, new_db_pass):
"""Function to reset 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
new_db_pass(string) : new password
Returns:
True/False(bool) : flag indicating success or failure in
resetting the password
"""
project_id = get_project_id()
instance_connection_name = project_id + ":" + location + ":" + instance_name
# Establish the connection pool
def conn():
return getconn(instance_connection_name, db_user, db_pass, db_name)
try:
pool = sqlalchemy.create_engine(
"postgresql+pg8000://",
creator=conn,
)
except Exception as e: # pylint: disable=broad-exception-caught
print(e)
try:
with pool.connect() as db_conn:
query = f"ALTER USER {db_user} WITH PASSWORD '{new_db_pass}'"
db_conn.execute(text(query))
db_conn.commit()
pool.dispose()
time.sleep(10)
return True
except Exception as e: # pylint: disable=broad-exception-caught
print("Failed to reset password due to the following exception")
print(e)
return False