descriptor.py

# Copyright (c) 2016 Shotgun Software Inc.
#
# CONFIDENTIAL AND PROPRIETARY
#
# This work is provided "AS IS" and subject to the Shotgun Pipeline Toolkit
# Source Code License included in this distribution package. See LICENSE.
# By accessing, using, copying or modifying this work you indicate your
# agreement to the Shotgun Pipeline Toolkit Source Code License. All rights
# not expressly granted therein are reserved by Shotgun Software Inc.

import os
import copy

from ..log import LogManager
from ..util import filesystem
from .io_descriptor import create_io_descriptor
from .errors import TankDescriptorError
from ..util import LocalFileStorageManager
from . import constants


logger = LogManager.get_logger(__name__)


def create_descriptor(
        sg_connection,
        descriptor_type,
        dict_or_uri,
        bundle_cache_root_override=None,
        fallback_roots=None,
        resolve_latest=False,
        constraint_pattern=None,
        local_fallback_when_disconnected=True):
    """
    Factory method. Use this when creating descriptor objects.

    .. note:: Descriptors inherit their threading characteristics from
        the shotgun connection that they carry internally. They are reentrant
        and should not be passed between threads.

    :param sg_connection: Shotgun connection to associated site
    :param descriptor_type: Either ``Descriptor.APP``, ``CORE``, ``CONFIG``, ``INSTALLED_CONFIG``,
        ``ENGINE`` or ``FRAMEWORK``
    :param dict_or_uri: A std descriptor dictionary dictionary or string
    :param bundle_cache_root_override: Optional override for root path to where
                                       downloaded apps are cached. If not specified,
                                       the global bundle cache location will be used. This location is a per-user
                                       cache that is shared across all sites and projects.
    :param fallback_roots: Optional List of immutable fallback cache locations where
                           apps will be searched for. Note that when descriptors
                           download new content, it always ends up in the
                           bundle_cache_root.
    :param resolve_latest: If true, the latest version will be determined and returned.

                           If set to True, no version information needs to be supplied with
                           the descriptor dictionary/uri for descriptor types which support
                           a version number concept. Please note that setting this flag
                           to true will typically affect performance - an external connection
                           is often required in order to establish what the latest version is.

                           If a remote connection cannot be established when attempting to determine
                           the latest version, a local scan will be carried out and the highest
                           version number that is cached locally will be returned.
    :param constraint_pattern: If resolve_latest is True, this pattern can be used to constrain
                           the search for latest to only take part over a subset of versions.
                           This is a string that can be on the following form:

                                - ``v0.1.2``, ``v0.12.3.2``, ``v0.1.3beta`` - a specific version
                                - ``v0.12.x`` - get the highest v0.12 version
                                - ``v1.x.x`` - get the highest v1 version
    :param local_fallback_when_disconnected: If resolve_latest is set to True, specify the behaviour
                            in the case when no connection to a remote descriptor can be established,
                            for example because and internet connection isn't available. If True, the
                            descriptor factory will attempt to fall back on any existing locally cached
                            bundles and return the latest one available. If False, a
                            :class:`TankDescriptorError` is raised instead.

    :returns: :class:`Descriptor` object
    :raises: :class:`TankDescriptorError`
    """
    # use the environment variable if set - if not, fall back on the override or default locations
    if os.environ.get(constants.BUNDLE_CACHE_PATH_ENV_VAR):
        bundle_cache_root_override = os.path.expanduser(
            os.path.expandvars(os.environ.get(constants.BUNDLE_CACHE_PATH_ENV_VAR))
        )
    elif bundle_cache_root_override is None:
        bundle_cache_root_override = _get_default_bundle_cache_root()
        filesystem.ensure_folder_exists(bundle_cache_root_override)
    else:
        # expand environment variables
        bundle_cache_root_override = os.path.expanduser(os.path.expandvars(bundle_cache_root_override))

    fallback_roots = fallback_roots or []

    # expand environment variables
    fallback_roots = [os.path.expandvars(os.path.expanduser(x)) for x in fallback_roots]

    # first construct a low level IO descriptor
    io_descriptor = create_io_descriptor(
        sg_connection,
        descriptor_type,
        dict_or_uri,
        bundle_cache_root_override,
        fallback_roots,
        resolve_latest,
        constraint_pattern,
        local_fallback_when_disconnected
    )

    # now create a high level descriptor and bind that with the low level descriptor
    return Descriptor.create(
        sg_connection,
        descriptor_type,
        io_descriptor,
        bundle_cache_root_override,
        fallback_roots
    )


def _get_default_bundle_cache_root():
    """
    Returns the cache location for the default bundle cache.

    :returns: path on disk
    """
    return os.path.join(
        LocalFileStorageManager.get_global_root(LocalFileStorageManager.CACHE),
        "bundle_cache"
    )


class Descriptor(object):
    """
    A descriptor describes a particular version of an app, engine or core component.
    It also knows how to access metadata such as documentation, descriptions etc.

    Descriptor is subclassed to distinguish different types of payload;
    apps, engines, configs, cores etc. Each payload may have different accessors
    and helper methods.
    """

    APP = constants.DESCRIPTOR_APP
    FRAMEWORK = constants.DESCRIPTOR_FRAMEWORK
    ENGINE = constants.DESCRIPTOR_ENGINE
    CONFIG = constants.DESCRIPTOR_CONFIG
    CORE = constants.DESCRIPTOR_CORE
    INSTALLED_CONFIG = constants.DESCRIPTOR_INSTALLED_CONFIG

    _factory = {}

    @classmethod
    def register_descriptor_factory(cls, descriptor_type, subclass):
        """
        Registers a descriptor subclass with the :meth:`create` factory.
        This is an internal method that should not be called by external
        code.

        :param descriptor_type: Either ``Descriptor.APP``, ``CORE``,
            ``CONFIG``, ``INSTALLED_CONFIG``, ``ENGINE`` or ``FRAMEWORK``
        :param subclass: Class deriving from Descriptor to associate.
        """
        cls._factory[descriptor_type] = subclass

    @classmethod
    def create(cls, sg_connection, descriptor_type, io_descriptor, bundle_cache_root_override, fallback_roots):
        """
        Factory method used by :meth:`create_descriptor`. This is an internal
        method that should not be called by external code.

        :param descriptor_type: Either ``Descriptor.APP``, ``CORE``,
            ``CONFIG``, ``INSTALLED_CONFIG``, ``ENGINE`` or ``FRAMEWORK``
        :param sg_connection: Shotgun connection to associated site
        :param io_descriptor: Associated low level descriptor transport object.
        :param bundle_cache_root_override: Override for root path to where
            downloaded apps are cached.
        :param fallback_roots: List of immutable fallback cache locations where
            apps will be searched for.
        :returns: Instance of class deriving from :class:`Descriptor`
        :raises: TankDescriptorError
        """
        if descriptor_type not in cls._factory:
            raise TankDescriptorError("Unsupported descriptor type %s" % descriptor_type)
        class_obj = cls._factory[descriptor_type]
        return class_obj(sg_connection, io_descriptor, bundle_cache_root_override, fallback_roots)

    def __init__(self, io_descriptor):
        """
        .. note:: Use the factory method :meth:`create_descriptor` when
                  creating new descriptor objects.

        :param io_descriptor: Associated IO descriptor.
        """
        # construct a suitable IO descriptor for this descriptor
        self._io_descriptor = io_descriptor

    def __eq__(self, other):
        # By default, we can assume equality if the path to the data
        # on disk is equivalent.
        if isinstance(other, self.__class__):
            return self.get_path() == other.get_path()
        else:
            return False

    def __ne__(self, other):
        return not (self == other)

    def __repr__(self):
        """
        Used for debug logging
        """
        class_name = self.__class__.__name__
        return "<%s %r>" % (class_name, self._io_descriptor)

    def __str__(self):
        """
        Used for pretty printing
        """
        return "%s %s" % (self.system_name, self.version)

    def _get_manifest(self):
        """
        Returns the info.yml metadata associated with this descriptor.

        :returns: dictionary with the contents of info.yml
        """
        return self._io_descriptor.get_manifest(constants.BUNDLE_METADATA_FILE)

    ###############################################################################################
    # data accessors

    def get_dict(self):
        """
        Returns the dictionary associated with this descriptor

        :returns: Dictionary that can be used to construct the descriptor
        """
        return self._io_descriptor.get_dict()

    # legacy support for previous method name
    get_location = get_dict

    def get_uri(self):
        """
        Returns the uri associated with this descriptor
        The uri is a string based representation that is equivalent to the
        descriptor dictionary returned by the get_dict() method.

        :returns: Uri string that can be used to construct the descriptor
        """
        return self._io_descriptor.get_uri()

    def copy(self, target_folder):
        """
        Copy the config descriptor into the specified target location.

        :param target_folder: Folder to copy the descriptor to
        """
        self._io_descriptor.copy(target_folder)

    def clone_cache(self, cache_root):
        """
        The descriptor system maintains an internal cache where it downloads
        the payload that is associated with the descriptor. Toolkit supports
        complex cache setups, where you can specify a series of path where toolkit
        should go and look for cached items.

        This is an advanced method that helps in cases where a user wishes to
        administer such a setup, allowing a cached payload to be copied from
        its current location into a new cache structure.

        If the descriptor's payload doesn't exist on disk, it will be downloaded.

        :param cache_root: Root point of the cache location to copy to.
        """
        return self._io_descriptor.clone_cache(cache_root)

    @property
    def display_name(self):
        """
        The display name for this item.
        If no display name has been defined, the system name will be returned.
        """
        meta = self._get_manifest()
        display_name = meta.get("display_name")
        if display_name is None:
            display_name = self.system_name
        return display_name

    def is_dev(self):
        """
        Returns true if this item is intended for development purposes

        :returns: True if this is a developer item
        """
        return self._io_descriptor.is_dev()

    def is_immutable(self):
        """
        Returns true if this descriptor never changes its content.
        This is true for most descriptors as they represent a particular
        version, tag or commit of an item. Examples of non-immutable
        descriptors include path and dev descriptors, where the
        descriptor points at a "live" location on disk where a user
        can make changes at any time.

        :returns: True if this is a developer item
        """
        return self._io_descriptor.is_immutable()

    @property
    def description(self):
        """
        A short description of the item.
        """
        meta = self._get_manifest()
        desc = meta.get("description")
        if desc is None:
            desc = "No description available."
        return desc

    @property
    def icon_256(self):
        """
        The path to a 256px square png icon file representing this item
        """
        app_icon = os.path.join(self._io_descriptor.get_path(), "icon_256.png")
        if os.path.exists(app_icon):
            return app_icon
        else:
            # return default
            default_icon = os.path.abspath(os.path.join(
                os.path.dirname(__file__),
                "resources",
                "default_bundle_256px.png"
            ))
            return default_icon

    @property
    def support_url(self):
        """
        A url that points at a support web page associated with this item.
        If not url has been defined, ``None`` is returned.
        """
        meta = self._get_manifest()
        support_url = meta.get("support_url")
        if support_url is None:
            support_url = "https://support.shotgunsoftware.com"
        return support_url

    @property
    def documentation_url(self):
        """
        The documentation url for this item. If no documentation url has been defined,
        a url to the toolkit user guide is returned.
        """
        meta = self._get_manifest()
        doc_url = meta.get("documentation_url")
        if doc_url is None:
            doc_url = "https://support.shotgunsoftware.com/hc/en-us/articles/115000068574-User-Guide"
        return doc_url

    @property
    def deprecation_status(self):
        """
        Information about deprecation status.

        :returns: Returns a tuple (is_deprecated, message) to indicate
                  if this item is deprecated.
        """
        return self._io_descriptor.get_deprecation_status()

    @property
    def system_name(self):
        """
        A short name, suitable for use in configuration files and for folders on disk.
        """
        return self._io_descriptor.get_system_name()

    @property
    def version(self):
        """
        The version number string for this item.
        """
        return self._io_descriptor.get_version()

    def get_path(self):
        """
        Returns the path to a location where this item is cached.

        When locating the item, any bundle cache fallback paths
        will first be searched in the order they have been defined,
        and lastly the main bundle cached will be checked.

        If the item is not locally cached, ``None`` is returned.

        :returns: Path string or ``None`` if not cached.
        """
        return self._io_descriptor.get_path()

    @property
    def changelog(self):
        """
        Information about the changelog for this item.

        :returns: A tuple (changelog_summary, changelog_url). Values may be ``None``
                  to indicate that no changelog exists.
        """
        return self._io_descriptor.get_changelog()

    def ensure_local(self):
        """
        Helper method. Ensures that the item is locally available.
        """
        return self._io_descriptor.ensure_local()

    def exists_local(self):
        """
        Returns true if this item exists in a local repo.
        """
        return self._io_descriptor.exists_local()

    def download_local(self):
        """
        Retrieves this version to local repo.
        """
        return self._io_descriptor.download_local()

    def find_latest_version(self, constraint_pattern=None):
        """
        Returns a descriptor object that represents the latest version.

        .. note:: Different descriptor types implements this logic differently,
                  but general good practice is to follow the semantic version numbering
                  standard for any versions used in conjunction with toolkit. This ensures
                  that toolkit can track and correctly determine not just the latest version
                  but also apply constraint pattern matching such as looking for the latest
                  version matching the pattern ``v1.x.x``. You can read more about semantic
                  versioning here: http://semver.org/

        :param constraint_pattern: If this is specified, the query will be constrained
               by the given pattern. Version patterns are on the following forms:

                - v0.1.2, v0.12.3.2, v0.1.3beta - a specific version
                - v0.12.x - get the highest v0.12 version
                - v1.x.x - get the highest v1 version

        :returns: instance derived from :class:`Descriptor`
        """
        # make a copy of the descriptor
        latest = copy.copy(self)
        # find latest I/O descriptor
        latest._io_descriptor = self._io_descriptor.get_latest_version(constraint_pattern)
        return latest

    def find_latest_cached_version(self, constraint_pattern=None):
        """
        Returns a descriptor object that represents the latest version
        that can be found in the local bundle caches.

        .. note:: Different descriptor types implements this logic differently,
                  but general good practice is to follow the semantic version numbering
                  standard for any versions used in conjunction with toolkit. This ensures
                  that toolkit can track and correctly determine not just the latest version
                  but also apply constraint pattern matching such as looking for the latest
                  version matching the pattern ``v1.x.x``. You can read more about semantic
                  versioning here: http://semver.org/

        :param constraint_pattern: If this is specified, the query will be constrained
               by the given pattern. Version patterns are on the following forms:

                - v0.1.2, v0.12.3.2, v0.1.3beta - a specific version
                - v0.12.x - get the highest v0.12 version
                - v1.x.x - get the highest v1 version

        :returns: Instance derived from :class:`Descriptor` or ``None`` if no cached version
                  is available.
        """
        io_desc = self._io_descriptor.get_latest_cached_version(constraint_pattern)
        if io_desc is None:
            return None

        # make a copy of the descriptor
        latest = copy.copy(self)
        # find latest I/O descriptor
        latest._io_descriptor = io_desc
        return latest

    def has_remote_access(self):
        """
        Probes if the current descriptor is able to handle
        remote requests. If this method returns, true, operations
        such as :meth:`download_local` and :meth:`find_latest_version`
        can be expected to succeed.

        :return: True if a remote is accessible, false if not.
        """
        return self._io_descriptor.has_remote_access()

    # compatibility accessors to ensure that all systems
    # calling this (previously internal!) parts of toolkit
    # will still work.
    def get_display_name(self):
        return self.display_name

    def get_description(self):
        return self.description

    def get_icon_256(self):
        return self.icon_256

    def get_support_url(self):
        return self.support_url

    def get_doc_url(self):
        return self.documentation_url

    def get_deprecation_status(self):
        return self.deprecation_status

    def get_system_name(self):
        return self.system_name

    def get_version(self):
        return self.version

    def get_changelog(self):
        return self.changelog