dashboard_data = dashboard_data.replace('OSM_ID', uid).replace('OSM_NAME', name)
dashboard_json_data = json.loads(dashboard_data)
- # get folder id
+ # Get folder id
if project_name:
folder_name = project_name
else:
response = requests.request(
"POST", self.url + "/api/dashboards/db/", data=json.dumps(dashboard_json_data), headers=self.headers)
- # get team id
+ # Don't create the dashboard if already exists.
+ if json.loads(response.text).get("status") == "name-exists":
+ return
+
+ # Get team id
if project_name is not None:
name = project_name
response_team = requests.request(
"GET", self.url + "/api/teams/search?name={}".format(name), headers=self.headers)
- if len(json.loads(response_team.text)["teams"]) > 0:
- team_id = json.loads(response_team.text)["teams"][0]["id"]
- permission_data = {"items": [{"teamId": team_id, "permission": 2}, ]}
- # provide permission to dashboard
- response = requests.request(
- "POST", self.url + "/api/folders/{}/permissions".format(project_name), json=permission_data,
+
+ # Remove default permissions of admin user's dashboard so that it is not visible to non-admin users
+ if len(json.loads(response_team.text)["teams"]) == 0:
+ # As team information is not available so it is admin user
+ dahboard_id = json.loads(response.text)["id"]
+ requests.request(
+ "POST", self.url + "/api/dashboards/id/{}/permissions".format(dahboard_id),
headers=self.headers)
+
log.info("Dashboard %s is created in Grafana", name)
return response
except Exception:
log.info("New user %s created in Grafana", user)
return response_users
- # create grafana team with member
+ # Create Grafana team with member
def create_grafana_teams_members(self, project_name, user_name, is_admin, proj_list):
- # check if user exist in grafana or not
+ # Check if user exist in Grafana
user_response = requests.request("GET", self.url + "/api/users/lookup?loginOrEmail={}".format(user_name),
headers=self.headers)
user_obj = json.loads(user_response.text)
user_id = user_obj["id"]
- # Get Teams for user
+ # Get teams for user
team_objs = requests.request("GET", self.url + "/api/users/{}/teams".format(user_id), headers=self.headers)
team_obj = json.loads(team_objs.text)
team_list = []
requests.request("DELETE", self.url + "/api/teams/{}/members/{}".format(team_id, user_id),
headers=self.headers)
if project_name != "admin":
- # add member to team
+ # Add member to team
response_team = requests.request("GET", self.url + "/api/teams/search?name={}".format(project_name),
headers=self.headers)
+ # Search if team in Grafana corresponding to the project already exists
if not json.loads(response_team.text)["teams"]:
- # team doesn't exist in grafana. Creating it first
self.create_grafana_teams(project_name)
response_team = requests.request("GET", self.url + "/api/teams/search?name={}".format(project_name),
headers=self.headers)
team_id = json.loads(response_team.text)["teams"][0]["id"]
if project_name not in team_list:
+ # Create a team in Grafana corresponding to the project as it doesn't exist
member_payload = {
"userId": user_id
}
requests.request("POST", self.url + "/api/teams/{}/members".format(team_id), json=member_payload,
headers=self.headers)
- # if role is admin change permission to admin
- if is_admin:
+ # Check if user role or project name is admin
+ if is_admin or project_name == 'admin':
+ # Give admin righsts to user
url = "/api/org/users/{}/".format(user_id)
permission_payload = {"role": "Admin", }
requests.request("PATCH", self.url + url, json=permission_payload, headers=self.headers)
log.info("User %s is assigned Admin permission", user_name)
else:
+ # Give editor rights to user
url = "/api/org/users/{}/".format(user_id)
permission_payload = {"role": "Editor", }
requests.request("PATCH", self.url + url, json=permission_payload, headers=self.headers)
log.info("User %s is assigned Editor permission", user_name)
- # create grafana team
+ # Create team in Grafana
def create_grafana_teams(self, team_name):
team_payload = {"name": team_name, }
requests.request("POST", self.url + "/api/teams", json=team_payload, headers=self.headers)
log.info("New team %s created in Grafana", team_name)
- # create grafana folder
+ # Create folder in Grafana
def create_grafana_folders(self, folder_name):
folder_payload = {"uid": folder_name, "title": folder_name}
requests.request("POST", self.url + "/api/folders", json=folder_payload, headers=self.headers)
log.info("Dashboard folder %s created", folder_name)
+ response_team = requests.request("GET", self.url + "/api/teams/search?name={}".format(folder_name),
+ headers=self.headers)
+ # Create team if it doesn't already exists
+ if len(json.loads(response_team.text)["teams"]) == 0:
+ self.create_grafana_teams(folder_name)
+ response_team = requests.request("GET", self.url + "/api/teams/search?name={}".format(folder_name),
+ headers=self.headers)
+ # Assign required permission to the team's folder
+ team_id = json.loads(response_team.text)["teams"][0]["id"]
+ permission_data = {"items": [{"teamId": team_id, "permission": 2}, ]}
+ requests.request("POST", self.url + "/api/folders/{}/permissions".format(folder_name),
+ json=permission_data, headers=self.headers)
+
+ # delete user from grafana
def delete_grafana_users(self, user_name):
- # find user id
+ # Get user id
response_id = requests.request("GET", self.url + "/api/users/lookup?loginOrEmail={}".format(user_name),
headers=self.headers)
try:
user_id = json.loads(response_id.text)["id"]
except Exception:
log.exception("Exception processing message: ")
- # delete user
+ # Delete user
response = requests.request("DELETE", self.url + "/api/admin/users/{}".format(user_id), headers=self.headers)
log.info("User %s deleted in Grafana", user_name)
return response
+ # delete team from grafana
def delete_grafana_team(self, project_name):
- # delete grafana folder
+ # Delete Grafana folder
requests.request("DELETE", self.url + "/api/folders/{}".format(project_name),
headers=self.headers)
- # delete grafana team
+ # Delete Grafana team
team_obj = requests.request("GET", self.url + "/api/teams/search?name={}".format(project_name),
headers=self.headers)
team_id = json.loads(team_obj.text)["teams"][0]["id"]
log.info("Team %s deleted in Grafana", project_name)
return response
+ # update grafana team
def update_grafana_teams(self, project_new_name, project_old_name):
team_obj = requests.request("GET", self.url + "/api/teams/search?name={}".format(project_old_name),
headers=self.headers)
response = requests.request("PUT", self.url + "/api/teams/{}".format(team_id), json=data, headers=self.headers)
log.info("Grafana team updated %s", response.text)
return response
+
+ # remove member from grafana team
+ def remove_grafana_team_member(self, user_name, project_data):
+ # Get user id
+ response_id = requests.request("GET", self.url + "/api/users/lookup?loginOrEmail={}".format(user_name),
+ headers=self.headers)
+ user_id = json.loads(response_id.text)["id"]
+ for project in project_data:
+ # Get team id
+ team_obj = requests.request("GET", self.url + "/api/teams/search?name={}".format(project['project']),
+ headers=self.headers)
+ team_id = json.loads(team_obj.text)["teams"][0]["id"]
+ response = requests.request("DELETE", self.url + "/api/teams/{}/members/{}".format(team_id, user_id),
+ headers=self.headers)
+ return response
projects / osm / MON.git / blobdiff