Source code for tank.descriptor.io_descriptor.factory

# Copyright (c) 2016 Shotgun Software Inc.
# 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 copy

from ..errors import TankDescriptorError
from .base import IODescriptorBase

from ... import LogManager
from tank_vendor import six

log = LogManager.get_logger(__name__)

def create_io_descriptor(
    Factory method. Use this method to construct all DescriptorIO instances.

    A descriptor is immutable in the sense that it always points at the same code -
    this may be a particular frozen version out of that toolkit app store that
    will not change or it may be a dev area where the code can change. Given this,
    descriptors are cached and only constructed once for a given descriptor URL.

    :param sg: Shotgun connection to associated site
    :param descriptor_type: Either AppDescriptor.APP, CORE, ENGINE or FRAMEWORK
    :param dict_or_uri: A std descriptor dictionary dictionary or string
    :param bundle_cache_root: Root path to where downloaded apps are cached
    :param fallback_roots: 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
    :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: Descriptor object
    :raises: :class:`TankDescriptorError`
    # resolve into both dict and uri form
    if isinstance(dict_or_uri, six.string_types):
        descriptor_dict = IODescriptorBase.dict_from_uri(dict_or_uri)
        # make a copy to make sure the original object is never altered
        descriptor_dict = copy.deepcopy(dict_or_uri)

    # at this point we didn't have a cache hit,
    # so construct the object manually
    if resolve_latest and is_descriptor_version_missing(descriptor_dict):
        # if someone is requesting a latest descriptor and not providing a version token
        # make sure to add an artificial one so that we can resolve it.
        descriptor_dict["version"] = "latest"

    # Ensure that the descrptor version is a str.  For some descriptors a
    # verson is not expected, so if the key doesn't exist we'll do nothing.
    if isinstance(descriptor_dict.get("version"), six.binary_type):
        # On Python 2 this will have no effect, but on Python 3, we will decode
        # bytes to a str.
        descriptor_dict["version"] = six.ensure_str(descriptor_dict["version"])

    # instantiate the Descriptor
    descriptor = IODescriptorBase.create(descriptor_type, descriptor_dict, sg)

    # specify where to go look for caches
    descriptor.set_cache_roots(bundle_cache_root, fallback_roots)

    if resolve_latest:
        # attempt to get "remote" latest first
        # and if that fails, fall back on the latest item
        # available in the local cache.
        log.debug("Trying to resolve latest version...")
        if descriptor.has_remote_access():
                "Remote connection is available - attempting to get latest version from remote..."
            descriptor = descriptor.get_latest_version(constraint_pattern)
            log.debug("Resolved latest to be %r" % descriptor)

            if local_fallback_when_disconnected:
                # get latest from bundle cache
                    "Remote connection is not available - will try to get "
                    "the latest locally cached version of %s..." % descriptor
                latest_cached_descriptor = descriptor.get_latest_cached_version(
                if latest_cached_descriptor is None:
                        "No locally cached versions of %r available." % descriptor
                    raise TankDescriptorError(
                        "Could not get latest version of %s. "
                        "For more details, see the log." % descriptor
                    "Latest locally cached descriptor is %r" % latest_cached_descriptor
                descriptor = latest_cached_descriptor

                # do not attempt to get the latest locally cached version
                    "Remote connection not available to determine latest version."
                raise TankDescriptorError(
                    "Could not get latest version of %s. "
                    "For more details, see the log." % descriptor

    return descriptor

[docs]def is_descriptor_version_missing(dict_or_uri): """ Helper method which checks if a descriptor needs a version. If the given descriptor dictionary or uri is one of the types which requires a version token, and this token is not defined, ``True`` will be returned, otherwise ``False``. This is useful for a standard pattern which can be used used where you want to allow users to configure toolkit descriptors which track either the latest version or a specific one. In this pattern, the user hints that they want to track latest version by omitting the version token altogether. The following standard pattern can then be implemented:: # determine if we should request the latest version # of the given descriptor if is_descriptor_version_missing(descriptor_uri): # require the descriptor system to return # the latest descriptor it can detect resolve_latest = True else: # normal direct lookup of a particular # descriptor version resolve_latest = False descriptor_obj = create_descriptor( sg_connection, Descriptor.CONFIG, descriptor_uri, resolve_latest=resolve_latest ) :param dict_or_uri: A std descriptor dictionary dictionary or string :return: Boolean to indicate if a required version token is missing """ # resolve into both dict and uri form if isinstance(dict_or_uri, six.string_types): descriptor_dict = descriptor_uri_to_dict(dict_or_uri) else: # make a copy to make sure the original object is never altered descriptor_dict = dict_or_uri # We only do this for descriptor types that supports a version number concept descriptors_using_version = ["app_store", "shotgun", "manual", "git", "git_branch"] if ( "version" not in descriptor_dict and descriptor_dict.get("type") in descriptors_using_version ): return True else: return False
[docs]def descriptor_uri_to_dict(uri): """ Translates a descriptor uri into a dictionary. :param uri: descriptor string uri :returns: descriptor dictionary """ return IODescriptorBase.dict_from_uri(uri)
[docs]def descriptor_dict_to_uri(ddict): """ Translates a descriptor dictionary into a uri. :param ddict: descriptor dictionary :returns: descriptor uri """ return IODescriptorBase.uri_from_dict(ddict)