diff --git a/atlassian/assets.py b/atlassian/assets.py new file mode 100644 index 000000000..ce4af44d7 --- /dev/null +++ b/atlassian/assets.py @@ -0,0 +1,637 @@ +# coding=utf-8 +import logging + +from .rest_client import AtlassianRestAPI +# from deprecated import deprecated + +log = logging.getLogger(__name__) + + +class Assets(AtlassianRestAPI): + """Assets for Jira API wrapper.""" + + # https://developer.atlassian.com/cloud/assets/rest + + def __init__(self, *args, **kwargs): + """ + Initialize Assets() + + :param args: + :param kwargs: + :return: Assets() + """ + kwargs["api_root"] = "rest/assets/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(Assets, self).__init__(*args, **kwargs) + + def __cloud_init(self, *args, **kwargs): + """ + Creates a Cloud specific version of Assets() + + Returns: + Assets(AtlassianRestAPI) + """ + # trigger a normal init and avoid looping + del kwargs["cloud"] + temp = Assets(*args, **kwargs) + # retrieve cloud workspace id and generate the api_root + kwargs["api_root"] = f"/jsm/assets/workspace/{temp.__get_workspace_id()}/v1/" + # Assets cloud uses the atlassian base url, not custom instance urls + kwargs["url"] = "https://api.atlassian.com" + # set cloud back to true and return + kwargs["cloud"] = True + # Assets 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/assets/workspace", + headers=self.default_headers, + ) + return result["values"][0]["workspaceId"] + + def _get_assets_workspace_ids(self): + """ + Returns all Assets workspace Ids. + :return: List + """ + result = self.get( + "rest/servicedeskapi/assets/workspace", + headers=self.experimental_headers, + ) + return [i["workspaceId"] for i in result["values"]] + + def _get_assets_workspace_id(self): + """ + Returns the first Assets workspace ID. + :return: str + """ + return next(iter(self._get_assets_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/assets/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/assets/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/assets/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 Assets Users + 1 Assets Managers + 2 Assets Administrators + 3 Assets 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/assets/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/assets/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/assets/1.0/icon/1/icon.png?size=16", + "url48": "http://jira/rest/assets/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 Assets + def reindex_assets(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_assets(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) + + # AQL + # Fetch Objects by AQL + # Cloud-only. Server use navlist_aql() + # https://developer.atlassian.com/cloud/assets/rest/api-group-object/#api-object-aql-post + def aql( + self, + query, + start=0, + max_results=25, + include_attributes=True + ): + """ + Resource dedicated to finding objects based on the Assets Query Language (AQL) + + :param query: + :param start: + :param max_results: + :param include_attributes: + :return: + """ + if not self.cloud: + raise NotImplementedError + params = { + "startAt": start, + "maxResults": max_results, + "includeAttributes": include_attributes + } + data = { + "qlQuery": query + } + url = self.url_joiner(self.api_root, "object/aql") + return self.post(url, params=params, data=data) + + # Navlist AQL + # Retrieve a list of objects based on an AQL query + # Server-only. Cloud use aql() + # https://developer.atlassian.com/server/jira-servicedesk/rest/v1006/api-group-assets---object/#api-assets-1-0-object-navlist-aql-post + def navlist_aql( + self, + query, + 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 query: + :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: + """ + if self.cloud: + raise NotImplementedError + data = { + "qlQuery": query, + "objectSchemaId": object_schema_id, + "page": page, + "orderAsc": order_asc, + "resultPerPage": result_per_page, + "includeAttributes": include_attributes, + "includeAttributesDeep": include_attributes_deep, + "includeTypeAttributes": include_type_attributes, + "includeExtendedInfo": include_extended_info + } + if order_by_attribute_id: + data["orderByAttributeId"] = order_by_attribute_id + if extended: + data["extended"] = extended + url = self.url_joiner(self.api_root, "object/navlist/aql") + return self.get(url, data=data) + + # 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 Assets + + :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) + + 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 Assets + + :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) + + # Objectconnectedtickets + def get_object_connected_tickets(self, object_id): + """ + Relation between Jira issues and Assets objects + + :param object_id: + :return: + """ + url = self.url_joiner( + self.api_root, + f"objectconnectedtickets/{object_id}/tickets", + ) + return self.get(url) + + def list_object_schema(self): + """ + Resource to find object schemas in Assets + :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 + + 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/assets/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/assets/rest/api-group-objecttype/#api-objecttype-id-position-post + # TODO: Post objecttype create: + # https://developer.atlassian.com/cloud/assets/rest/api-group-objecttype/#api-objecttype-create-post + + # Assets ObjectTypeAttribute API + # TODO: Post objecttypeattribute {objectTypeId}: + # https://developer.atlassian.com/cloud/assets/rest/api-group-objecttypeattribute/#api-objecttypeattribute-objecttypeid-post + # TODO: Put objecttypeattribute {objectTypeId} {id}: + # https://developer.atlassian.com/cloud/assets/rest/api-group-objecttypeattribute/#api-objecttypeattribute-objecttypeid-id-put + # TODO: Delete objecttypeattribute {id}: + # https://developer.atlassian.com/cloud/assets/rest/api-group-objecttypeattribute/#api-objecttypeattribute-id-delete + + # Assets Progress API + # TODO: Get progress category imports {id}: + # https://developer.atlassian.com/cloud/assets/rest/api-group-progress/#api-progress-category-imports-id-get + def get_progress_of_reindex(self): + """ + Show ongoing assets processes + :return: + """ + if self.cloud: + raise NotImplementedError + # Yes, the Assets Server API still uses "insight-reindex" as a category (as of April 2025) + # https://developer.atlassian.com/server/jira-servicedesk/rest/v1006/api-group-assets---progress/#api-assets-1-0-progress-category-category-resourceid-get + 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 assets 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) + + # Assets Config API + # TODO: Get config statustype: + # https://developer.atlassian.com/cloud/assets/rest/api-group-config/#api-config-statustype-get + # TODO: Post config statustype: + # https://developer.atlassian.com/cloud/assets/rest/api-group-config/#api-config-statustype-post + # TODO: Get config statustype {id}: + # https://developer.atlassian.com/cloud/assets/rest/api-group-config/#api-config-statustype-id-get + # TODO: Put config statustype {id}: + # https://developer.atlassian.com/cloud/assets/rest/api-group-config/#api-config-statustype-id-put + # TODO: Delete config statustype {id}: + # https://developer.atlassian.com/cloud/assets/rest/api-group-config/#api-config-statustype-id-delete + + # Update Issue with Assets Field + def update_issue_assets_field(self, key, field_id, assets_keys, add=False): + """ + Set the value of an Assets field. + Args: + key (str): Jira issue key, eg. SFT-446 + field_id (str): The internal Jira name of the Assets field, e.g. customfield_10200 + assets_keys (list): List of Assets 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 assets object updated. + """ + base_url = self.resource_url("issue") + action = "add" if add else "set" + data = { + "update": { + field_id: [{action: [{"key": i} for i in assets_keys]}], + } + } + data = {"fields": {field_id: [{"key": i} for i in assets_keys]}} + return self.put(f"{base_url}/{key}", data=data) \ No newline at end of file