def verify_change_password()

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