assets.py

# coding=utf-8
import logging

from .rest_client import AtlassianRestAPI

# from deprecated import deprecated

log = logging.getLogger(__name__)


class AssetsCloud(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(AssetsCloud, 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 = AssetsCloud(*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": "<a href=\"/jira/secure/ViewProfile.jspa?name=admin\">admin</a>",
                "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, data=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)