import requests import os import config import json class LookerAPI(): def __init__(self, config): self.client_id = config.looker_client_id self.client_secret = config.looker_client_secret self.url = config.looker_url self.auth_token = self.get_access_token(self.client_id, self.client_secret, self.url) def get_access_token(self, client_id, client_secret, url): ''' This funtion logs into the Looker API with proper creds and returns the authorization header (with token) ''' credentials = { 'client_id': client_id, 'client_secret': client_secret } url ='{}/login'.format(url) auth = requests.post(url, params=credentials) if auth.status_code == 200: token = auth.json()['access_token'] return {'Authorization' : 'token {}'.format(token)} else: print('Authorization Failed') def run_look(self, look_id): ''' Runs a Look and returns output in JSON format Args: look_id (int): id of Look you want to run Returns: list: JSON represntation of output ''' url = '{url}/looks/{look_id}/run/json'.format(url=self.url, look_id=look_id) response = requests.get( url, headers=self.auth_token ) if response.status_code == 200: return response.json() return 'Did not receive status code: 200' def get_user(self, user_id): ''' Gets info about a given user Args: user_id (int): id of user Returns: dict: Json representation of output ''' url = '{url}/users/{user_id}'.format(url=self.url, user_id=user_id) response = requests.get( url, headers=self.auth_token ) if response.status_code == 200: return response.json() return 'Did not receive status code: 200' def get_all_users(self): ''' Returns JSON of all Looker Users ''' url = '{url}/users'.format(url=self.url) response = requests.get( url, headers=self.auth_token ) if response.status_code == 200: return response.json() else: return 'Did not receive status code: 200' def disable_user(self, user_id): ''' Disables a user Args: user_id(int): user id you want to disable Returns: N/A ''' url = '{url}/users/{user_id}'.format(url=self.url, user_id=user_id) body = json.dumps({'is_disabled': True}) response = requests.patch( url, data=body, headers=self.auth_token ) if response.status_code == 200: print('User ID {} successfully disabled'.format(user_id)) else: return 'Did not receive status code: 200' def get_all_groups(self): ''' Returns JSON of all Looker Groups ''' url = '{url}/groups'.format(url=self.url) response = requests.get( url, headers=self.auth_token ) if response.status_code == 200: return response.json() else: return 'Did not receive status code: 200' def add_user_to_group(self, user_id, group_id): ''' Adds a specified user to a specified group Args: user_id(int) group_id(int) Returns: JSON ''' url = '{url}/groups/{group_id}/users'.format(url=self.url, group_id=group_id) body = json.dumps({'user_id': user_id}) response = requests.post( url, data=body, headers=self.auth_token ) return response.json()