# coding=utf-8 import logging from .rest_client import AtlassianRestAPI from deprecated import deprecated log = logging.getLogger(__name__) class Insight(AtlassianRestAPI): """Insight for Jira API wrapper.""" # https://insight-javadoc.riada.io/insight-javadoc-8.6/insight-rest/ def __init__(self, *args, **kwargs): """ Initialize Insight() :param args: :param kwargs: :return: Insight() """ kwargs["api_root"] = "rest/insight/1.0" # If cloud is set to true, trigger private __cloud__init method if kwargs.get("cloud"): args, kwargs = self.__cloud_init(*args, **kwargs) super(Insight, self).__init__(*args, **kwargs) def __cloud_init(self, *args, **kwargs): """ Creates an InsightCloud specific version of Insight() Returns: Insight(AtlassianRestAPI) """ # trigger a normal init and avoid looping del kwargs["cloud"] temp = Insight(*args, **kwargs) # retrieve cloud workspace id and generate the api_root kwargs["api_root"] = f"/jsm/insight/workspace/{temp.__get_workspace_id()}/v1/" # insight cloud uses the atlassian base url, not custom instnace urls kwargs["url"] = "https://api.atlassian.com" # set cloud back to true and return kwargs["cloud"] = True # Insight cloud is particular about its headers. self.default_headers = {"Accept": "application/json", "Content-Type": "application/json"} return args, kwargs def __get_workspace_id(self): result = self.get( "rest/servicedeskapi/insight/workspace", headers=self.default_headers, ) return result["values"][0]["workspaceId"] def _get_insight_workspace_ids(self): """ Returns all Insight workspace Ids. :return: List """ result = self.get( "rest/servicedeskapi/insight/workspace", headers=self.experimental_headers, ) return [i["workspaceId"] for i in result["values"]] def _get_insight_workspace_id(self): """ Returns the first Insight workspace ID. :return: str """ return next(iter(self._get_insight_workspace_ids())) # Attachments def get_attachments_of_objects(self, object_id): """ Get Attachment info Example output: [ { "id": 1, "author": "admin", "mimeType": "image/png", "filename": "astronaut.png", "filesize": "490 B", "created": "2019-11-27T11:42:22.283Z", "comment": "", "commentOutput": "", "url": "http://jira/rest/insight/1.0/attachments/1" } ] :param object_id Object ID :return list of object id: required(string) author: (string) mimeType: (string) filename: required(string) filesize: (string) created: required(datetime) comment: (string) commentOutput: (string) url: required(string) """ if self.cloud: raise NotImplementedError url = self.url_joiner( self.api_root, f"attachments/object/{object_id}", ) return self.get(url) def upload_attachment_to_object(self, object_id, filename): """ Add attachment to Object :param object_id: int :param filename: str, name, if file in current directory or full path to file """ if self.cloud: raise NotImplementedError log.info("Adding attachment...") url = f"rest/insight/1.0/attachments/object/{object_id}" with open(filename, "rb") as attachment: files = {"file": attachment} return self.post(url, headers=self.no_check_headers, files=files) def delete_attachment(self, attachment_id): """ Add attachment to Object :param attachment_id: int """ if self.cloud: raise NotImplementedError log.info("Deleting attachment...") url = f"rest/insight/1.0/attachments/{attachment_id}" return self.delete(url) # Comments # Handle comments on objets def add_comment_to_object(self, comment, object_id, role): """ Add comment to Object :param comment: str :param object_id: int - Object ID :param role: int - Role ID 0 Insight Users 1 Insight Managers 2 Insight Administrators 3 Insight Developers :return: { "created": "2019-11-27T12:37:41.492Z", "updated": "2019-11-27T12:37:41.492Z", "id": 1, "actor": { "avatarUrl": "https://www.gravatar.com/avatar/64e1b8d34f425d19e1ee2ea7236d3028?d=mm&s=48", "displayName": "admin", "name": "admin", "key": "admin", "renderedLink": "admin", "isDeleted": false }, "role": 0, "comment": "A comment to be added", "commentOutput": "A comment to be added", "objectId": 1, "canEdit": true, "canDelete": true } """ if self.cloud: raise NotImplementedError params = {"comment": comment, "objectId": object_id, "role": role} url = "rest/insight/1.0/comment/create" return self.post(url, params=params) def get_comment_of_object(self, object_id): """ The object id to fetch comments from :param object_id: :return: """ if self.cloud: raise NotImplementedError url = f"rest/insight/1.0/comment/object/{object_id}" return self.get(url) # Icon # Resources dedicated to load and find icons def get_icon_by_id(self, icon_id): """ Load a single icon by id :param icon_id: :return: { "id": 1, "name": "Delete", "url16": "http://jira/rest/insight/1.0/icon/1/icon.png?size=16", "url48": "http://jira/rest/insight/1.0/icon/1/icon.png?size=48" } """ url = self.url_joiner(self.api_root, f"icon/{icon_id}") return self.get(url) def get_all_global_icons(self): """ All existing global icons :return: """ url = self.url_joiner(self.api_root, "icon/global") return self.get(url) # Import # Start configured imports. To see an ongoing import see the Progress resource def start_import_configuration(self, import_id): """ The id of the import configuration that should be started :param import_id: :return: """ url = self.url_joiner( self.api_root, f"import/start/{import_id}", ) return self.post(url) # Index # Handle the indexing of Insight def reindex_insight(self): """ Should the reindex clean the index before doing the reindex :return: """ if self.cloud: raise NotImplementedError url = self.url_joiner(self.api_root, "index/reindex/start") return self.post(url) def reindex_current_node_insight(self): """ Should the reindex clean the index before doing the reindex :return: """ if self.cloud: raise NotImplementedError url = self.url_joiner(self.api_root, "index/reindex/currentnode") return self.post(url) # IQL # Resource dedicated to finding objects based on the Insight Query Language (IQL) def iql( self, iql, object_schema_id, page=1, order_by_attribute_id=None, order_asc=True, result_per_page=25, include_attributes=True, include_attributes_deep=1, include_type_attributes=False, include_extended_info=False, extended=None, ): """ Resource dedicated to finding objects based on the Insight Query Language (IQL) :param iql: :param object_schema_id: :param page: :param order_by_attribute_id: :param order_asc: :param result_per_page: :param include_attributes: :param include_attributes_deep: :param include_type_attributes: :param include_extended_info: :param extended: :return: """ params = {"iql": iql, "objectSchemaId": object_schema_id, "page": page} if order_by_attribute_id: params["orderByAttributeId"] = order_by_attribute_id params["orderAsc"] = order_asc params["resultPerPage"] = result_per_page params["includeAttributes"] = include_attributes params["includeAttributesDeep"] = include_attributes_deep params["includeTypeAttributes"] = include_type_attributes params["includeExtendedInfo"] = include_extended_info if extended: params["extended"] = extended url = self.url_joiner(self.api_root, "iql/objects") return self.get(url, params=params) # Object def get_object(self, object_id): """ Load one object :param object_id: :return: """ url = self.url_joiner(self.api_root, f"object/{object_id}") return self.get(url) def update_object( self, object_id, object_type_id, attributes, has_avatar=False, avatar_uuid="", ): """ Update an existing object in Insight :param object_id: :param object_type_id: :param attributes: :param has_avatar: :param avatar_uuid: :return: """ body = { "attributes": attributes, "objectTypeId": object_type_id, "avatarUUID": avatar_uuid, "hasAvatar": has_avatar, } url = self.url_joiner(self.api_root, f"object/{object_id}") return self.put(url, data=body) def delete_object(self, object_id): """ Delete the referenced object :param object_id: :return: """ url = self.url_joiner(self.api_root, f"object/{object_id}") return self.delete(url) def get_object_attributes(self, object_id): """ List all attributes for the given object :param object_id: :return: """ url = self.url_joiner(self.api_root, f"object/{object_id}/attributes") return self.get(url) def get_object_history(self, object_id, asc=False, abbreviate=True): """ Retrieve the history entries for this object :param object_id: :param asc: :param abbreviate: :return: """ params = {"asc": asc, "abbreviate": abbreviate} url = self.url_joiner(self.api_root, f"object/{object_id}/history") return self.get(url, params=params) @deprecated(version="3.29.0", reason="Use get_object_reference_info()") def get_object_referenceinfo(self, object_id): """Let's use the get_object_reference_info()""" self.get_object_reference_info(object_id) def get_object_reference_info(self, object_id): """ Find all references for an object :param object_id: :return: """ url = self.url_joiner(self.api_root, f"object/{object_id}/referenceinfo") return self.get(url) def create_object(self, object_type_id, attributes, has_avatar=False, avatar_uuid=""): """ Create a new object in Insight :param object_type_id: :param attributes: :param has_avatar: :param avatar_uuid: :return: :return: """ data = { "attributes": attributes, "objectTypeId": object_type_id, "avatarUUID": avatar_uuid, "hasAvatar": has_avatar, } url = self.url_joiner(self.api_root, "object/create") return self.post(url, data=data) def create_object_navlist_iql( self, iql, object_type_id, results_per_page, order_by_type_attr_id=None, object_id=None, object_schema_id=None, include_attributes=None, attributes_to_display=None, page=1, asc=0, ): """ A filter object that is used to find a paginated result set based on an object type and an IQL query :param iql: :param object_type_id: :param page: :param results_per_page: :param order_by_type_attr_id: :param asc: :param object_id: :param object_schema_id: :param include_attributes: :param attributes_to_display: :return: """ data = { "objectTypeId": object_type_id, "iql": iql, "resultsPerPage": results_per_page, "page": page, "asc": asc, } if attributes_to_display is not None: data["attributesToDisplay"] = attributes_to_display if include_attributes is not None: data["includeAttributes"] = include_attributes if object_schema_id is not None: data["objectSchemaId"] = object_schema_id if order_by_type_attr_id is not None: data["orderByTypeAttrId"] = order_by_type_attr_id if object_id is not None: data["objectId"] = object_id url = self.url_joiner(self.api_root, "iql/objects") return self.post(url, data=data) # Objectconnectedtickets def get_object_connected_tickets(self, object_id): """ Relation between Jira issues and Insight objects :param object_id: :return: """ url = self.url_joiner( self.api_root, f"objectconnectedtickets/{object_id}/tickets", ) return self.get(url) # Object schema @deprecated(version="3.29.1", reason="Use list_object_schema()") def list_objectschema(self): return self.list_object_schema() def list_object_schema(self): """ Resource to find object schemas in Insight :return: { "objectschemas": [ { "id": 1, "name": "Test", "objectSchemaKey": "TEST", "status": "Ok", "created": "2019-11-26T08:05:46.894Z", "updated": "2019-11-26T08:05:46.894Z", "objectCount": 2, "objectTypeCount": 3 } ] } """ url = self.url_joiner(self.api_root, "objectschema/list") return self.get(url) def create_object_schema(self, object_schema_key, description): raise NotImplementedError @deprecated(version="3.29.1", reason="Use get_objectschema()") def get_objectschema(self, schema_id): return self.get_objectschema(schema_id=schema_id) def get_object_schema(self, schema_id): """ Find a schema by id :param schema_id: """ url = self.url_joiner(self.api_root, f"objectschema/{schema_id}") return self.get(url) def update_object_schema(self, schema_id): """ Update an object schema """ raise NotImplementedError def delete_object_schema(self, schema_id): """ Delete a schema """ raise NotImplementedError def get_object_schema_attributes(self, schema_id): """ Find all object type attributes for this object schema """ raise NotImplementedError def get_object_schema_object_types_flat(self, schema_id, query=None, exclude=None, includeObjectCounts=None): """ Find all object types for this object schema https://developer.atlassian.com/cloud/assets/rest/api-group-objectschema/#api-objectschema-id-objecttypes-flat-get Args: schema_id (str): id of the object schema query (bool, optional): Object Type Names to search for, defaults to None (Use API default) exclude (str, optional): Exclude objects with this name, defaults to None (Use API default) includeObjectCounts (bool, optional): Populate objectCount attribute for each object type, defaults to None (Use API default) """ kwargs = list(locals().items()) params = dict() params.update({k: v for k, v in kwargs if v is not None and k not in ["self", "schema_id"]}) return self.get( f"{self.api_root}/objectschema/{schema_id}/objecttypes/flat", params=params, ) def get_object_type_attributes( self, type_id, only_value_editable=None, order_by_name=None, query=None, include_values_exist=None, exclude_parent_attributes=None, include_children=None, order_by_required=None, ): """ Find all attributes for this object type https://developer.atlassian.com/cloud/insight/rest/api-group-objecttype/#api-objecttype-id-attributes-get Args: type_id (str): id of the object type only_value_editable (bool, optional): only return editable values, defaults to None (Use API default) order_by_name (bool, optional): values, defaults to None (Use API default) query (str, optional): Not documented in API, defaults to None (Use API default) include_values_exist (bool, optional): Include only where values exist, defaults to None (Use API default) exclude_parent_attributes (bool, optional): Exclude parent attributes, defaults to None (Use API default) include_children (bool, optional): include attributes from children, defaults to None (Use API default) order_by_required (bool, optional): Order by required fields, defaults to None (Use API default) """ kwargs = list(locals().items()) params = dict() params.update({k: v for k, v in kwargs if v is not None and k not in ["self", "type_id"]}) return self.get( f"{self.api_root}/objecttype/{type_id}/attributes", headers=self.experimental_headers, params=params, ) # Objecttype # TODO: Post objecttype {id} position: # https://developer.atlassian.com/cloud/insight/rest/api-group-objecttype/#api-objecttype-id-position-post # TODO: Post objecttype create: # https://developer.atlassian.com/cloud/insight/rest/api-group-objecttype/#api-objecttype-create-post # Insight ObjectTypeAttribute API # TODO: Post objecttypeattribute {objectTypeId}: # https://developer.atlassian.com/cloud/insight/rest/api-group-objecttypeattribute/#api-objecttypeattribute-objecttypeid-post # TODO: Put objecttypeattribute {objectTypeId} {id}: # https://developer.atlassian.com/cloud/insight/rest/api-group-objecttypeattribute/#api-objecttypeattribute-objecttypeid-id-put # TODO: Delete objecttypeattribute {id}: # https://developer.atlassian.com/cloud/insight/rest/api-group-objecttypeattribute/#api-objecttypeattribute-id-delete # Insight Progress API # TODO: Get progress category imports {id}: # https://developer.atlassian.com/cloud/insight/rest/api-group-progress/#api-progress-category-imports-id-get def get_progress_of_reindex(self): """ Show ongoing insight processes :return: """ if self.cloud: raise NotImplementedError url = self.url_joiner(self.api_root, "progress/category/insight-reindex/reindex") return self.get(url) def get_progress_of_import(self, import_id): """ Show ongoing insight processes :type import_id: int: The id of the import source configuration that the progress should be fetched for :return: """ if self.cloud: raise NotImplementedError url = self.url_joiner(self.api_root, f"progress/category/imports/{import_id}") return self.get(url) # Insight Config API # TODO: Get config statustype: # https://developer.atlassian.com/cloud/insight/rest/api-group-config/#api-config-statustype-get # TODO: Post config statustype: # https://developer.atlassian.com/cloud/insight/rest/api-group-config/#api-config-statustype-post # TODO: Get config statustype {id}: # https://developer.atlassian.com/cloud/insight/rest/api-group-config/#api-config-statustype-id-get # TODO: Put config statustype {id}: # https://developer.atlassian.com/cloud/insight/rest/api-group-config/#api-config-statustype-id-put # TODO: Delete config statustype {id}: # https://developer.atlassian.com/cloud/insight/rest/api-group-config/#api-config-statustype-id-delete # Update Issue with Insight Field def update_issue_insight_field(self, key, field_id, insight_keys, add=False): """ Set the value of an Insight field. Args: key (str): Jira issue key, eg. SFT-446 field_id (str): The internal Jira name of the Insight field, e.g. customfield_10200 insight_keys (list): List of Insight objects to associate with the field. Limited to 20 objects. If the field only takes a single object pass a single value list. add (bool, optional): Add to the existing field rather than setting the field value. Defaults to False. Returns: [type]: The insight object updated. """ base_url = self.resource_url("issue") action = "add" if add else "set" data = { "update": { field_id: [{action: [{"key": i} for i in insight_keys]}], } } data = {"fields": {field_id: [{"key": i} for i in insight_keys]}} return self.put(f"{base_url}/{key}", data=data) def check_duplicate_attribute_values(self): """ Check for duplicate attribute values in Insight objects with cardinality maximum 1 used for Data Center link: https://confluence.atlassian.com/jirakb/duplicated-attribute-values-in-insight-objects-with-cardinality-maximum-1-1114816155.html :return: """ url = "rest/insight/1.0/health/consistency/duplicates/attributevalues" return self.get(url) def delete_duplicate_attribute_values(self): """ Delete duplicate attribute values in Insight objects with cardinality maximum 1 used for Data Center link: https://confluence.atlassian.com/jirakb/duplicated-attribute-values-in-insight-objects-with-cardinality-maximum-1-1114816155.html :return: """ url = "rest/insight/1.0/health/consistency/duplicates/attributevalues" return self.delete(url)