add_grafana_users.py

import csv
import json
import logging
import os

import requests

logger = logging.getLogger(__name__)

"""CSV format for users file:
name,email
*David Ohana,NotMyRealEmail@ibm.com
John Doe,jdoe112233@ibm.com
"""

def main():
    logging.basicConfig(level=logging.DEBUG)
    grafana_server_url = "https://localhost:3001"
    grafana_admin_password = os.environ["grafana_admin_password"]
    grafana_default_user_password = os.environ["grafana_default_user_password"]
    users_csv_file = "grafana_users.csv"
    ignore_errors = False
    add_user_endpoint = grafana_server_url + "/api/admin/users/"
    patch_user_endpoint = grafana_server_url + "/api/org/users/"
    logger.info("add_user_endpoint: {}".format(add_user_endpoint))

    logger.info("reading users file from {}".format(users_csv_file))
    with open(users_csv_file) as f:
        next(f, None)  # skip the header row
        reader = csv.reader(f)
        user_list = list(reader)

    logger.info("read {} users".format(len(user_list)))

    for user in user_list:
        name = str(user[0])
        email = str(user[1])
        is_admin = False
        if name.startswith("*"):
            name = name[1:]
            is_admin = True

        user_dict = {
            "name": name,
            "email": email,
        }
        logger.info("adding user: {}".format(str(user_dict)))
        user_dict["login"] = name
        user_dict["password"] = grafana_default_user_password

        r = requests.post(
            add_user_endpoint,
            json=user_dict,
            verify=False,
            auth=('admin', grafana_admin_password)
        )
        logger.info("user add response: {} - {}".format(r.status_code, r.text))
        if r.status_code != 200:
            logger.error("Failed to add user")
            if ignore_errors:
                continue
            break

        if is_admin:
            logger.info("setting user {} as admin".format(email))
            response_dict = json.loads(r.text)
            if "id" not in response_dict:
                logger.warning("cannot get user id")
                continue

            user_id = response_dict["id"]
            permission_doc = {"isGrafanaAdmin": True}
            permission_end_point = "{}{}/permissions".format(add_user_endpoint, user_id)
            r = requests.put(permission_end_point,
                             json=permission_doc,
                             verify=False,
                             auth=('admin', grafana_admin_password))
            logger.info("permission response: {} - {}".format(r.status_code, r.text))

            logger.info("setting user org role {} as admin".format(email))
            role_doc = {"role": "Admin"}
            role_end_point = patch_user_endpoint + str(user_id)
            r = requests.patch(role_end_point,
                               json=role_doc,
                               verify=False,
                               auth=('admin', grafana_admin_password))
            logger.info("role response: {} - {}".format(r.status_code, r.text))


if __name__ == "__main__":
    main()