def reset_password()

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