"""Dashboard tools. Contains tools for syncing over dashboards. """ import requests import config import utils def get_dashboard_json(dashboard_id, instance=config.LOOKER_MAIN_INSTANCE_URL): """Get json of a dashboard. Args: dashboard_id (int): id of dashboard. Found in dashboard url after .com/dashboards instance (str): url of an instance. Should only use consts from config.py Returns: dict: json representation of a dashboard. bool: False if dashboard with given id wasn't found on a given instance. """ validation = config.get_validation(instance) response = requests.get( f'{instance}/dashboards/{dashboard_id}', headers=validation ) return utils.process_response(response) def post_dashboard(client_ready_dashboard_json): """Create dashboard on client looker instance. Args: client_ready_dashboard_json (dict): json representation of a dashboard. Returns: dict: json of a newly created dashboard on client instance. bool: False if failed to post. """ response = requests.post( f'{config.LOOKER_CLIENT_INSTANCE_URL}/dashboards', headers=config.LOOKER_CLIENT_VALIDATION, json=client_ready_dashboard_json ) return utils.process_response(response) def update_client_dashboard(client_ready_json, dashboard_id): """Update a dashboard on a client looker instance. Args: client_ready_json (dict): json representation of a dashboard, ready to be posted to client instance. dashboard_id (int): id of a dashboard on a client looker instance. Returns: dict: json of an updated dashboard on client instance. bool: False if failed to post. """ response = requests.patch( f'{config.LOOKER_CLIENT_INSTANCE_URL}/dashboards/{dashboard_id}', headers=config.LOOKER_CLIENT_VALIDATION, json=client_ready_json ) return utils.process_response(response) def change_dashboard_space_id(dashboard_json): """Change space_id of a dashboard. Args: dashboard_json (dict): json representation of a dashboard. Returns: dict: json representation of a dashboard. """ dashboard_json['space_id'] = config.CLIENT_LOOKER_SPACE_ID return dashboard_json def change_dashboard_folder_id(dashboard_json): """Change folder_id of a dashboard. Args: dashboard_json (dict): json representation of a dashboard. Returns: dict: json representation of a dashboard. """ dashboard_json['folder_id'] = config.CLIENT_LOOKER_FOLDER_ID return dashboard_json def change_dashboard_user_id(dashboard_json): """Change user_id of a dashboard. Original dashboard user_id contains value for some user on main instance. That could have unpredictable consequences if posted to client instance unchanged, in case there are two different users (between main and client instances) who share the "same" user_id. Args: dashboard_json (dict): json representation of a dashboard. Returns: dict: json representation of a dashboard. """ dashboard_json['user_id'] = config.CLIENT_LOOKER_USER_ID return dashboard_json def adjust_dashboard_json(dashboard_json): """Adjust data for client looker instance. Args: dashboard_json (dict): json representation of a dashboard. Returns: dict: json representation of a dashboard. """ changed_space = change_dashboard_space_id(dashboard_json) changed_folder = change_dashboard_folder_id(changed_space) changed_user = change_dashboard_user_id(changed_folder) return changed_user def get_all_dashboards(instance=config.LOOKER_CLIENT_INSTANCE_URL): """Get all dashboards (default - from client instance). Args: instance (str): URL of an instance. Returns: list: contains dicts which are json representations of dashboards on a given instance. """ validation = config.get_validation(instance) response = requests.get( f'{instance}/dashboards', headers=validation ) return utils.process_response(response) def check_dashboard_on_client_instance(dashboard_title): """Check if dashboard already exists on client instance. Args: dashboard_title (str): title of a dashboard on main instance. Returns: int: id of a dashboard it it exists on an instance. bool: False is not found. """ list_of_dashboards = get_all_dashboards() for dashboard in list_of_dashboards: if dashboard['title'] == dashboard_title: return dashboard['id'] return False def get_list_of_dashboard_filters( instance_dashboard_id, instance=config.LOOKER_CLIENT_INSTANCE_URL): """Get the list of all dashboard filters. Args: instance_dashboard_id (int): id of a dashboard at a given instance. instance (str): url of an instance. Returns: list: contains dicts, each one is a json representation of a dashboard filter. """ validation = config.get_validation(instance) response = requests.get( f'{instance}/dashboards/{instance_dashboard_id}/dashboard_filters', headers=validation ) return utils.process_response(response) def post_filter_to_client_instance(filter_json): """Post dashboard filter to client looker instance. Args: filter_json (dict): json representation of dashboard filter. Returns: bool: True is posted successfully, False otherwise. """ response = requests.post( f'{config.LOOKER_CLIENT_INSTANCE_URL}/dashboard_filters', headers=config.LOOKER_CLIENT_VALIDATION, json=filter_json ) return utils.process_response(response) def update_dashboard_filter(filter_json, client_filter_id): """Update dashboard filter on client looker instance. Args: filter_json (dict): json representation of a filter. client_filter_id (int): id of a filter in client looker instance. Returns: bool: True if updated successfully, False otherwise. """ response = requests.patch( f'{config.LOOKER_CLIENT_INSTANCE_URL}' f'/dashboard_filters/{client_filter_id}', headers=config.LOOKER_CLIENT_VALIDATION, json=filter_json ) return utils.process_response(response) def sync_dashboard_filters( main_list_of_dashboard_filters, client_dashboard_id, client_list_of_filters=False): """Sync over all the necessary dashboard filters. Args: main_list_of_dashboard_filters (list): contains dicts, which are json representations of dashboard filters. Comes as a return value of get_list_of_dashboard_filters function. client_dashboard_id (int): id of a dashboard on client instance. client_list_of_filters (list or False): list contains dicts which are json representations of a dashboard filters. False if there is no such list on the first place. """ if not client_list_of_filters: for filter in main_list_of_dashboard_filters: filter['dashboard_id'] = client_dashboard_id post_filter_to_client_instance(filter) else: client_filter_titles =\ (c_filter['title'] for c_filter in client_list_of_filters) for main_filter in main_list_of_dashboard_filters: main_filter['dashboard_id'] = client_dashboard_id if not main_filter['title'] in client_filter_titles: post_filter_to_client_instance(main_filter) else: for client_filter in client_list_of_filters: if main_filter['title'] == client_filter['title']: client_filter_id = client_filter['id'] update_dashboard_filter( main_filter, client_filter_id) def get_dashboard_element_json(element_id): """Get json of a dashboard element. Args: element_id (int): id of dashboard element on main looker instance. Returns: dict: json representation of a dashboard element. """ response = requests.get( f'{config.LOOKER_MAIN_INSTANCE_URL}/dashboard_elements/{element_id}', headers=config.LOOKER_MAIN_VALIDATION ) return utils.process_response(response) def adjust_dashboard_element_json( main_element_json, client_dashboard_id, client_query_id, client_look_id): """Set some values in element json for appropriate ones. Args: main_element_json (dict): json representation of a dashboard element. client_dashboard_id (int): id of a dashboard on client looker instance. client_query_id (int or None): id a query on a client looker instance. client_look_id (int or None): id a look on a client instance. Returns: dict: json representation of a dashboard element, ready to be posted or updated to client instance. """ main_element_json['dashboard_id'] = client_dashboard_id main_element_json['query_id'] = client_query_id main_element_json['look_id'] = client_look_id return main_element_json def check_element_in_dashboard(element_title, list_of_dashboard_elements): """Check if element is already in a client dashboard. Args: element_title (str): title of a dashboard element we're checking for. list_of_dashboard_elements (list): contains strings, which are titles of elements present. Most likely it's used to see whether you need to update or post an element, so it has to be a list from client instance dashboard. Returns: bool: True is an element with given name was found, False otherwise. """ for element in list_of_dashboard_elements: if element['title'] == element_title: return True return False def post_dashboard_element(element_json): """Create a dashboard element on client looker instance. Args: element_json (dict): json representation of an element. Should have query_id for correct query on client instance. Returns: int: id of a newly created element. bool: False if response code was other than 200. """ response = requests.post( f'{config.LOOKER_CLIENT_INSTANCE_URL}/dashboard_elements', headers=config.LOOKER_CLIENT_VALIDATION, json=element_json ) return utils.process_response(response)['id'] def update_dashboard_element(element_json): """Update dashboard element on client looker instance. Args: element_json (dict): json representation of an element. Returns: bool: True is updated successfully, False otherwise. """ response = requests.patch( f'{config.LOOKER_CLIENT_INSTANCE_URL}' f"/dashboard_elements/{element_json['id']}", json=element_json, headers=config.LOOKER_CLIENT_VALIDATION ) return utils.process_response(response)['id'] def get_main_dashboard_layout(layout_id): """Get json of a dashboard layout on main looker instance. Args: layout_id (int): id of a dashboard layout on main looker instance. Returns: dict: json representation of a dashboard layout. bool: False if failed to post. """ response = requests.get( f'{config.LOOKER_MAIN_INSTANCE_URL}/dashboard_layouts/{layout_id}', headers=config.LOOKER_MAIN_VALIDATION ) return utils.process_response(response) def post_dashboard_layout(layout_json): """Create a dashboard layout on client looker instance. Args: layout_json (dict): json representation of a dashboard layout. Should have dashboard_id key set to correct value for client instance. Returns: int: id of a newly created layout. bool: False if request wasn't successful. """ response = requests.post( f'{config.LOOKER_CLIENT_INSTANCE_URL}/dashboard_layouts', headers=config.LOOKER_CLIENT_VALIDATION, json=layout_json ) return utils.process_response(response)['id'] def update_dashboard_layout(client_layout_id, layout_json): """Update a dashboard layout on client looker instance. Args: client_layout_id (int)L id of a dashboard layout on client looker instance. layout_json (dict): json representation of a dashboard layout. Returns: bool: True if updated successfully, False otherwise. """ response = requests.patch( f'{config.LOOKER_CLIENT_INSTANCE_URL}' f'/dashboard_layouts/{client_layout_id}', headers=config.LOOKER_CLIENT_VALIDATION, json=layout_json ) return utils.process_response(response) def get_all_dashboard_layout_components( layout_id, instance=config.LOOKER_MAIN_INSTANCE_URL): """Get list of all dashboard layout component jsons. Args: Returns: list: contains dicts which are json representations of dashboard layout components. """ validation = config.get_validation(instance) response = requests.get( f'{instance}/dashboard_layouts/' f'{layout_id}/dashboard_layout_components', headers=validation ) return utils.process_response(response) def update_layout_component(component_json, client_component_id): """Update a dashboard layout component on client looker instance. Args: component_json (dict): json representation of a layout component. client_component_id (int): id of a component on client instance to be updated. Returns: bool: True if updated successfully, False otherwise. """ response = requests.patch( f'{config.LOOKER_CLIENT_INSTANCE_URL}' f'/dashboard_layout_components/{client_component_id}', headers=config.LOOKER_CLIENT_VALIDATION, json=component_json ) return utils.process_response(response) def delete_dashboard_element(element_id): """Delete dashboard element on client instance. Args: element_id (int): id of a dashboard element. """ requests.delete( f'{config.LOOKER_CLIENT_INSTANCE_URL}/dashboard_elements/{element_id}', headers=config.LOOKER_CLIENT_VALIDATION)