_CondaAPI.dependencies() - Code Metrics - Inspection of "Merge pull request #87 from spyder-ide/fix/conda_a..." - spyder-ide/conda-manager - Measure and Improve Code Quality continuously with Scrutinizer

Completed
Push — master ( cfccd9...b77fc8 )

by Gonzalo
created 2016-07-04 19:04 UTC
_CondaAPI.dependencies() D

↳ Parent: Project
Complexity

Conditions
Size

Total Lines
Duplication

Lines	0
Ratio	0 %
Importance

Changes	1
Bugs	0	Features	0
Metric	Value
cc	8
c	1
b	0
f	0
dl	0
loc	33
rs	4
# -*- coding: utf-8 -*-
# -----------------------------------------------------------------------------
# Copyright © 2015- The Spyder Development Team
# Copyright © 2014-2015 Gonzalo Peña-Castellanos (@goanpeca)
#
# Licensed under the terms of the MIT License
# -----------------------------------------------------------------------------
"""
Updated `conda-api` running on a Qt QProcess to avoid UI blocking.

This also add some extra methods to the original conda-api.
"""

# Standard library imports
from collections import deque
from os.path import abspath, basename, expanduser, isdir, join
import json
import os
import platform
import re
import sys

# Third party imports
from qtpy.QtCore import QByteArray, QObject, QProcess, QTimer, Signal
import yaml

# Local imports
from conda_manager.utils.findpip import PIP_LIST_SCRIPT
from conda_manager.utils.logs import logger
from conda_manager.utils.py3compat import is_text_string

__version__ = '1.3.0'


# --- Errors
# -----------------------------------------------------------------------------
class PipError(Exception):
    """General pip error."""

    pass


class CondaError(Exception):
    """General Conda error."""

    pass


class CondaProcessWorker(CondaError):
    """General Conda error."""

    pass


class CondaEnvExistsError(CondaError):
    """Conda environment already exists."""

    pass


# --- Helpers
# -----------------------------------------------------------------------------
PY2 = sys.version[0] == '2'
PY3 = sys.version[0] == '3'
DEBUG = False


def to_text_string(obj, encoding=None):
    """Convert `obj` to (unicode) text string."""
    if PY2:
        # Python 2
        if encoding is None:
            return unicode(obj)
        else:
            return unicode(obj, encoding)
    else:
        # Python 3
        if encoding is None:
            return str(obj)
        elif isinstance(obj, str):
            # In case this function is not used properly, this could happen
            return obj
        else:
            return str(obj, encoding)


def handle_qbytearray(obj, encoding):
    """Qt/Python3 compatibility helper."""
    if isinstance(obj, QByteArray):
        obj = obj.data()

    return to_text_string(obj, encoding=encoding)


class ProcessWorker(QObject):
    """Conda worker based on a QProcess for non blocking UI."""

    sig_finished = Signal(object, object, object)
    sig_partial = Signal(object, object, object)

    def __init__(self, cmd_list, parse=False, pip=False, callback=None,
                 extra_kwargs=None):
        """Conda worker based on a QProcess for non blocking UI.

        Parameters
        ----------
        cmd_list : list of str
            Command line arguments to execute.
        parse : bool (optional)
            Parse json from output.
        pip : bool (optional)
            Define as a pip command.
        callback : func (optional)
            If the process has a callback to process output from comd_list.
        extra_kwargs : dict
            Arguments for the callback.
        """
        super(ProcessWorker, self).__init__()
        self._result = None
        self._cmd_list = cmd_list
        self._parse = parse
        self._pip = pip
        self._conda = not pip
        self._callback = callback
        self._fired = False
        self._communicate_first = False
        self._partial_stdout = None
        self._extra_kwargs = extra_kwargs if extra_kwargs else {}

        self._timer = QTimer()
        self._process = QProcess()

        self._timer.setInterval(150)

        self._timer.timeout.connect(self._communicate)
        # self._process.finished.connect(self._communicate)
        self._process.readyReadStandardOutput.connect(self._partial)

    def _partial(self):
        """Callback for partial output."""
        raw_stdout = self._process.readAllStandardOutput()
        stdout = handle_qbytearray(raw_stdout, _CondaAPI.UTF8)

        json_stdout = stdout.replace('\n\x00', '')
        try:
            json_stdout = json.loads(json_stdout)
        except Exception:
            json_stdout = stdout

        if self._partial_stdout is None:
            self._partial_stdout = stdout
        else:
            self._partial_stdout += stdout

        self.sig_partial.emit(self, json_stdout, None)

    def _communicate(self):
        """Callback for communicate."""
        if (not self._communicate_first and
                self._process.state() == QProcess.NotRunning):
            self.communicate()
        elif self._fired:
            self._timer.stop()

    def communicate(self):
        """Retrieve information."""
        self._communicate_first = True
        self._process.waitForFinished()

        if self._partial_stdout is None:
            raw_stdout = self._process.readAllStandardOutput()
            stdout = handle_qbytearray(raw_stdout, _CondaAPI.UTF8)
        else:
            stdout = self._partial_stdout

        raw_stderr = self._process.readAllStandardError()
        stderr = handle_qbytearray(raw_stderr, _CondaAPI.UTF8)
        result = [stdout.encode(_CondaAPI.UTF8), stderr.encode(_CondaAPI.UTF8)]

        # FIXME: Why does anaconda client print to stderr???
        if PY2:
            stderr = stderr.decode()
        if 'using anaconda' not in stderr.lower():
            if stderr.strip() and self._conda:
                logger.error('{0}:\nSTDERR:\n{1}\nEND'.format(
                        ' '.join(self._cmd_list), stderr))
            elif stderr.strip() and self._pip:
                logger.error("pip error: {}".format(self._cmd_list))
        result[-1] = ''

        if self._parse and stdout:
            try:
                result = json.loads(stdout), result[-1]
            except Exception as error:
                result = stdout, str(error)

            if 'error' in result[0]:
                if not isinstance(result[0], dict):
                    result = {'error': str(result[0])}, None
                error = '{0}: {1}'.format(" ".join(self._cmd_list),
                                          result[0]['error'])
                result = result[0], error

        if self._callback:
            result = self._callback(result[0], result[-1],
                                    **self._extra_kwargs), result[-1]

        self._result = result
        self.sig_finished.emit(self, result[0], result[-1])

        if result[-1]:
            logger.error(str(('error', result[-1])))

        self._fired = True

        return result

    def close(self):
        """Close the running process."""
        self._process.close()

    def is_finished(self):
        """Return True if worker has finished processing."""
        return self._process.state() == QProcess.NotRunning and self._fired

    def start(self):
        """Start process."""
        logger.debug(str(' '.join(self._cmd_list)))

        if not self._fired:
            self._partial_ouput = None
            self._process.start(self._cmd_list[0], self._cmd_list[1:])
            self._timer.start()
        else:
            raise CondaProcessWorker('A Conda ProcessWorker can only run once '
                                     'per method call.')


# --- API
# -----------------------------------------------------------------------------
class _CondaAPI(QObject):
    """Conda API to connect to conda in a non blocking way via QProcess."""

    ENCODING = 'ascii'
    UTF8 = 'utf-8'
    DEFAULT_CHANNELS = ['https://repo.continuum.io/pkgs/pro',
                        'https://repo.continuum.io/pkgs/free']

    def __init__(self, parent=None):
        """Conda API to connect to conda in a non blocking way via QProcess."""
        super(_CondaAPI, self).__init__()
        self._parent = parent
        self._queue = deque()
        self._timer = QTimer()
        self._current_worker = None
        self._workers = []

        self._timer.setInterval(1000)
        self._timer.timeout.connect(self._clean)

        self.ROOT_PREFIX = None
        self.set_root_prefix()

        # Set config files path
        self.user_rc_path = abspath(expanduser('~/.condarc'))
        self.sys_rc_path = join(self.ROOT_PREFIX, '.condarc')

    def _clean(self):
        """Remove references of inactive workers periodically."""
        if self._workers:
            for w in self._workers:
                if w.is_finished():
                    self._workers.remove(w)
        else:
            self._current_worker = None
            self._timer.stop()

    def _start(self):
        if len(self._queue) == 1:
            self._current_worker = self._queue.popleft()
            self._workers.append(self._current_worker)
            self._current_worker.start()
            self._timer.start()

    def is_active(self):
        """Check if a worker is still active."""
        return len(self._workers) == 0

    def terminate_all_processes(self):
        """Kill all working processes."""
        for worker in self._workers:
            worker.close()

    # --- Conda api
    # -------------------------------------------------------------------------
    def _call_conda(self, extra_args, abspath=True, parse=False,
                    callback=None):
        """
        Call conda with the list of extra arguments, and return the worker.

        The result can be force by calling worker.communicate(), which returns
        the tuple (stdout, stderr).
        """
        if abspath:
            if sys.platform == 'win32':
                python = join(self.ROOT_PREFIX, 'python.exe')
                conda = join(self.ROOT_PREFIX, 'Scripts',
                             'conda-script.py')
            else:
                python = join(self.ROOT_PREFIX, 'bin/python')
                conda = join(self.ROOT_PREFIX, 'bin/conda')
            cmd_list = [python, conda]
        else:
            # Just use whatever conda is on the path
            cmd_list = ['conda']

        cmd_list.extend(extra_args)

        process_worker = ProcessWorker(cmd_list, parse=parse,
                                       callback=callback)
        process_worker.sig_finished.connect(self._start)
        self._queue.append(process_worker)
        self._start()

        return process_worker

    def _call_and_parse(self, extra_args, abspath=True, callback=None):
        return self._call_conda(extra_args, abspath=abspath, parse=True,
                                callback=callback)

    @staticmethod
    def _setup_install_commands_from_kwargs(kwargs, keys=tuple()):
        """Setup install commands for conda."""
        cmd_list = []
        if kwargs.get('override_channels', False) and 'channel' not in kwargs:
            raise TypeError('conda search: override_channels requires channel')

        if 'env' in kwargs:
            cmd_list.extend(['--name', kwargs.pop('env')])
        if 'prefix' in kwargs:
            cmd_list.extend(['--prefix', kwargs.pop('prefix')])
        if 'channel' in kwargs:
            channel = kwargs.pop('channel')
            if isinstance(channel, str):
                cmd_list.extend(['--channel', channel])
            else:
                cmd_list.append('--channel')
                cmd_list.extend(channel)

        for key in keys:
            if key in kwargs and kwargs[key]:
                cmd_list.append('--' + key.replace('_', '-'))

        return cmd_list

    def set_root_prefix(self, prefix=None):
        """
        Set the prefix to the root environment (default is /opt/anaconda).

        This function should only be called once (right after importing
        conda_api).
        """
        if prefix:
            self.ROOT_PREFIX = prefix
        else:
            # Find some conda instance, and then use info to get 'root_prefix'
            worker = self._call_and_parse(['info', '--json'], abspath=False)
            info = worker.communicate()[0]
            self.ROOT_PREFIX = info['root_prefix']

    def get_conda_version(self):
        """Return the version of conda being used (invoked) as a string."""
        return self._call_conda(['--version'],
                                callback=self._get_conda_version)

    @staticmethod
    def _get_conda_version(stdout, stderr):
        """Callback for get_conda_version."""
        # argparse outputs version to stderr in Python < 3.4.
        # http://bugs.python.org/issue18920
        pat = re.compile(r'conda:?\s+(\d+\.\d\S+|unknown)')
        m = pat.match(stderr.decode().strip())
        if m is None:
            m = pat.match(stdout.decode().strip())

        if m is None:
            raise Exception('output did not match: {0}'.format(stderr))

        return m.group(1)

    def get_envs(self, log=True):
        """Return environment list of absolute path to their prefixes."""
        if log:
            logger.debug('')
#        return self._call_and_parse(['info', '--json'],
#                                    callback=lambda o, e: o['envs'])
        envs = os.listdir(os.sep.join([self.ROOT_PREFIX, 'envs']))
        envs = [os.sep.join([self.ROOT_PREFIX, 'envs', i]) for i in envs]

        valid_envs = [e for e in envs if os.path.isdir(e) and
                      self.environment_exists(prefix=e)]

        return valid_envs

    def get_prefix_envname(self, name):
        """Return full prefix path of environment defined by `name`."""
        prefix = None
        if name == 'root':
            prefix = self.ROOT_PREFIX

#        envs, error = self.get_envs().communicate()
        envs = self.get_envs()
        for p in envs:
            if basename(p) == name:
                prefix = p

        return prefix

    @staticmethod
    def linked(prefix):
        """Return set of canonical names of linked packages in `prefix`."""
        logger.debug(str(prefix))

        if not isdir(prefix):
            return set()

        meta_dir = join(prefix, 'conda-meta')
        if not isdir(meta_dir):
            # We might have nothing in linked (and no conda-meta directory)
            return set()

        return set(fn[:-5] for fn in os.listdir(meta_dir)
                   if fn.endswith('.json'))

    @staticmethod
    def split_canonical_name(cname):
        """Split a canonical package name into name, version, build."""
        return tuple(cname.rsplit('-', 2))

    def info(self, abspath=True):
        """
        Return a dictionary with configuration information.

        No guarantee is made about which keys exist.  Therefore this function
        should only be used for testing and debugging.
        """
        logger.debug(str(''))

        return self._call_and_parse(['info', '--json'], abspath=abspath)

    def package_info(self, package, abspath=True):
        """Return a dictionary with package information."""
        return self._call_and_parse(['info', package, '--json'],
                                    abspath=abspath)

    def search(self, regex=None, spec=None, **kwargs):
        """Search for packages."""
        cmd_list = ['search', '--json']

        if regex and spec:
            raise TypeError('conda search: only one of regex or spec allowed')

        if regex:
            cmd_list.append(regex)

        if spec:
            cmd_list.extend(['--spec', spec])

        if 'platform' in kwargs:
            cmd_list.extend(['--platform', kwargs.pop('platform')])

        cmd_list.extend(
            self._setup_install_commands_from_kwargs(
                kwargs,
                ('canonical', 'unknown', 'use_index_cache', 'outdated',
                 'override_channels')))

        return self._call_and_parse(cmd_list,
                                    abspath=kwargs.get('abspath', True))

    def create_from_yaml(self, name, yamlfile):
        """
        Create new environment using conda-env via a yaml specification file.

        Unlike other methods, this calls conda-env, and requires a named
        environment and uses channels as defined in rcfiles.

        Parameters
        ----------
        name : string
            Environment name
        yamlfile : string
            Path to yaml file with package spec (as created by conda env export
        """
        logger.debug(str((name, yamlfile)))
        cmd_list = ['env', 'create', '-n', name, '-f', yamlfile, '--json']
        return self._call_and_parse(cmd_list)

    def create(self, name=None, prefix=None, pkgs=None, channels=None):
        """Create an environment with a specified set of packages."""
        logger.debug(str((prefix, pkgs, channels)))

        # TODO: Fix temporal hack
        if (not pkgs or (not isinstance(pkgs, (list, tuple)) and
                         not is_text_string(pkgs))):
            raise TypeError('must specify a list of one or more packages to '
                            'install into new environment')

        cmd_list = ['create', '--yes', '--json', '--mkdir']
        if name:
            ref = name
            search = [os.path.join(d, name) for d in
                      self.info().communicate()[0]['envs_dirs']]
            cmd_list.extend(['--name', name])
        elif prefix:
            ref = prefix
            search = [prefix]
            cmd_list.extend(['--prefix', prefix])
        else:
            raise TypeError('must specify either an environment name or a '
                            'path for new environment')

        if any(os.path.exists(prefix) for prefix in search):
            raise CondaEnvExistsError('Conda environment {0} already '
                                      'exists'.format(ref))

        # TODO: Fix temporal hack
        if isinstance(pkgs, (list, tuple)):
            cmd_list.extend(pkgs)
        elif is_text_string(pkgs):
            cmd_list.extend(['--file', pkgs])

        # TODO: Check if correct
        if channels:
            cmd_list.extend(['--override-channels'])

            for channel in channels:
                cmd_list.extend(['--channel'])
                cmd_list.extend([channel])

        return self._call_and_parse(cmd_list)

    def parse_token_channel(self, channel, token):
        """
        Adapt a channel to include token of the logged user.

        Ignore default channels.
        """
        if (token and channel not in self.DEFAULT_CHANNELS and
                channel != 'defaults'):
            url_parts = channel.split('/')
            start = url_parts[:-1]
            middle = 't/{0}'.format(token)
            end = url_parts[-1]
            token_channel = '{0}/{1}/{2}'.format('/'.join(start), middle, end)
            return token_channel
        else:
            return channel

    def install(self, name=None, prefix=None, pkgs=None, dep=True,
                channels=None, token=None):
        """
        Install a set of packages into an environment by name or path.

        If token is specified, the channels different from the defaults will
        get the token appended.
        """
        logger.debug(str((prefix, pkgs, channels)))

        # TODO: Fix temporal hack
        if not pkgs or not isinstance(pkgs, (list, tuple, str)):
            raise TypeError('must specify a list of one or more packages to '
                            'install into existing environment')

        cmd_list = ['install', '--yes', '--json', '--force-pscheck']
        if name:
            cmd_list.extend(['--name', name])
        elif prefix:
            cmd_list.extend(['--prefix', prefix])
        else:
            # Just install into the current environment, whatever that is
            pass

        # TODO: Check if correct
        if channels:
            cmd_list.extend(['--override-channels'])

            for channel in channels:
                cmd_list.extend(['--channel'])
                channel = self.parse_token_channel(channel, token)
                cmd_list.extend([channel])

        # TODO: Fix temporal hack
        if isinstance(pkgs, (list, tuple)):
            cmd_list.extend(pkgs)
        elif isinstance(pkgs, str):
            cmd_list.extend(['--file', pkgs])

        if not dep:
            cmd_list.extend(['--no-deps'])

        return self._call_and_parse(cmd_list)

    def update(self, *pkgs, **kwargs):
        """Update package(s) (in an environment) by name."""
        cmd_list = ['update', '--json', '--yes']

        if not pkgs and not kwargs.get('all'):
            raise TypeError("Must specify at least one package to update, or "
                            "all=True.")

        cmd_list.extend(
            self._setup_install_commands_from_kwargs(
                kwargs,
                ('dry_run', 'no_deps', 'override_channels',
                 'no_pin', 'force', 'all', 'use_index_cache', 'use_local',
                 'alt_hint')))

        cmd_list.extend(pkgs)

        return self._call_and_parse(cmd_list, abspath=kwargs.get('abspath',
                                                                 True))

    def remove(self, name=None, prefix=None, pkgs=None, all_=False):
        """
        Remove a package (from an environment) by name.

        Returns {
            success: bool, (this is always true),
            (other information)
        }
        """
        logger.debug(str((prefix, pkgs)))

        cmd_list = ['remove', '--json', '--yes']

        if not pkgs and not all_:
            raise TypeError("Must specify at least one package to remove, or "
                            "all=True.")

        if name:
            cmd_list.extend(['--name', name])
        elif prefix:
            cmd_list.extend(['--prefix', prefix])
        else:

            raise TypeError('must specify either an environment name or a '
                            'path for package removal')

        if all_:
            cmd_list.extend(['--all'])
        else:
            cmd_list.extend(pkgs)

        return self._call_and_parse(cmd_list)

    def remove_environment(self, name=None, path=None, **kwargs):
        """
        Remove an environment entirely.

        See ``remove``.
        """
        return self.remove(name=name, path=path, all=True, **kwargs)

    def clone_environment(self, clone, name=None, prefix=None, **kwargs):
        """Clone the environment `clone` into `name` or `prefix`."""
        cmd_list = ['create', '--json']

        if (name and prefix) or not (name or prefix):
            raise TypeError("conda clone_environment: exactly one of `name` "
                            "or `path` required")

        if name:
            cmd_list.extend(['--name', name])

        if prefix:
            cmd_list.extend(['--prefix', prefix])

        cmd_list.extend(['--clone', clone])

        cmd_list.extend(
            self._setup_install_commands_from_kwargs(
                kwargs,
                ('dry_run', 'unknown', 'use_index_cache', 'use_local',
                 'no_pin', 'force', 'all', 'channel', 'override_channels',
                 'no_default_packages')))

        return self._call_and_parse(cmd_list, abspath=kwargs.get('abspath',
                                                                 True))

    # FIXME:
#    def process(self, name=None, prefix=None, cmd=None):
#        """Create a Popen process for cmd inside specified conda env.
#
#        The returned object will need to be invoked with p.communicate() or
#        similar.
#        """
#        if bool(name) == bool(prefix):
#            raise TypeError('exactly one of name or prefix must be specified')
#
#        if not cmd:
#            raise TypeError('cmd to execute must be specified')
#
#        if not args:
#            args = []
#
#        if name:
#            prefix = self.get_prefix_envname(name)
#
#        conda_env = dict(os.environ)
#        sep = os.pathsep
#
#        if sys.platform == 'win32':
#            conda_env['PATH'] = join(prefix,
#                                     'Scripts') + sep + conda_env['PATH']
#        else:
#            # Unix
#            conda_env['PATH'] = join(prefix, 'bin') + sep + conda_env['PATH']
#
#        conda_env['PATH'] = prefix + os.pathsep + conda_env['PATH']
#
#        cmd_list = [cmd]
#        cmd_list.extend(args)

#         = self.subprocess.process(cmd_list, env=conda_env, stdin=stdin,
#                                        stdout=stdout, stderr=stderr)

    @staticmethod
    def _setup_config_from_kwargs(kwargs):
        """Setup config commands for conda."""
        cmd_list = ['--json', '--force']

        if 'file' in kwargs:
            cmd_list.extend(['--file', kwargs['file']])

        if 'system' in kwargs:
            cmd_list.append('--system')

        return cmd_list

#    def config_path(self, **kwargs):
#        """Get the path to the config file."""
#        cmd_list = ['config', '--get']
#        cmd_list.extend(self._setup_config_from_kwargs(kwargs))
#
#        return self._call_and_parse(cmd_list,
#                                    abspath=kwargs.get('abspath', True),
#                                    callback=lambda o, e: o['rc_path'])

#    def config_get(self, *keys, **kwargs):
#        """
#        Get the values of configuration keys.
#
#        Returns a dictionary of values. Note, the key may not be in the
#        dictionary if the key wasn't set in the configuration file.
#        """
#        cmd_list = ['config', '--get']
#        cmd_list.extend(keys)
#        cmd_list.extend(self._setup_config_from_kwargs(kwargs))
#
#        return self._call_and_parse(cmd_list,
#                                    abspath=kwargs.get('abspath', True),
#                                    callback=lambda o, e: o['get'])

#    def config_set(self, key, value, **kwargs):
#        """
#        Set a key to a (bool) value.
#
#        Returns a list of warnings Conda may have emitted.
#        """
#        cmd_list = ['config', '--set', key, str(value)]
#        cmd_list.extend(self._setup_config_from_kwargs(kwargs))
#
#        return self._call_and_parse(
#            cmd_list,
#            abspath=kwargs.get('abspath', True),
#            callback=lambda o, e: o.get('warnings', []))

    def config_add(self, key, value, **kwargs):
        """
        Add a value to a key.

        Returns a list of warnings Conda may have emitted.
        """
        cmd_list = ['config', '--add', key, value]
        cmd_list.extend(self._setup_config_from_kwargs(kwargs))

        return self._call_and_parse(
            cmd_list,
            abspath=kwargs.get('abspath', True),
            callback=lambda o, e: o.get('warnings', []))

    def config_remove(self, key, value, **kwargs):
        """
        Remove a value from a key.

        Returns a list of warnings Conda may have emitted.
        """
        cmd_list = ['config', '--remove', key, value]
        cmd_list.extend(self._setup_config_from_kwargs(kwargs))

        return self._call_and_parse(
            cmd_list,
            abspath=kwargs.get('abspath', True),
            callback=lambda o, e: o.get('warnings', []))

#    def config_delete(self, key, **kwargs):
#        """
#        Remove a key entirely.
#
#        Returns a list of warnings Conda may have emitted.
#        """
#        cmd_list = ['config', '--remove-key', key]
#        cmd_list.extend(self._setup_config_from_kwargs(kwargs))
#
#        return self._call_and_parse(
#            cmd_list,
#            abspath=kwargs.get('abspath', True),
#            callback=lambda o, e: o.get('warnings', []))

#    def run(self, command, abspath=True):
#        """
#        Launch the specified app by name or full package name.
#
#        Returns a dictionary containing the key "fn", whose value is the full
#        package (ending in ``.tar.bz2``) of the app.
#        """
#        cmd_list = ['run', '--json', command]
#
#        return self._call_and_parse(cmd_list, abspath=abspath)

    # --- Additional methods
    # -----------------------------------------------------------------------------
    def dependencies(self, name=None, prefix=None, pkgs=None, channels=None,
                     dep=True):
        """Get dependenciy list for packages to be installed in an env."""
        if not pkgs or not isinstance(pkgs, (list, tuple)):
            raise TypeError('must specify a list of one or more packages to '
                            'install into existing environment')

        cmd_list = ['install', '--dry-run', '--json', '--force-pscheck']

        if not dep:
            cmd_list.extend(['--no-deps'])

        if name:
            cmd_list.extend(['--name', name])
        elif prefix:
            cmd_list.extend(['--prefix', prefix])
        else:
            pass

        cmd_list.extend(pkgs)

        # TODO: Check if correct
        if channels:
            cmd_list.extend(['--override-channels'])

            for channel in channels:
                cmd_list.extend(['--channel'])
                cmd_list.extend([channel])

        return self._call_and_parse(cmd_list)

    def environment_exists(self, name=None, prefix=None, abspath=True,
                           log=True):
        """Check if an environment exists by 'name' or by 'prefix'.

        If query is by 'name' only the default conda environments directory is
        searched.
        """
        if log:
            logger.debug(str((name, prefix)))

        if name and prefix:
            raise TypeError("Exactly one of 'name' or 'prefix' is required.")

        if name:
            prefix = self.get_prefix_envname(name, log=log)

        if prefix is None:
            prefix = self.ROOT_PREFIX

        return os.path.isdir(os.path.join(prefix, 'conda-meta'))

    def clear_lock(self, abspath=True):
        """Clean any conda lock in the system."""
        cmd_list = ['clean', '--lock', '--json']
        return self._call_and_parse(cmd_list, abspath=abspath)

    def package_version(self, prefix=None, name=None, pkg=None, build=False):
        """Get installed package version in a given env."""
        package_versions = {}

        if name and prefix:
            raise TypeError("Exactly one of 'name' or 'prefix' is required.")

        if name:
            prefix = self.get_prefix_envname(name)

        if self.environment_exists(prefix=prefix):

            for package in self.linked(prefix):
                if pkg in package:
                    n, v, b = self.split_canonical_name(package)
                    if build:
                        package_versions[n] = '{0}={1}'.format(v, b)
                    else:
                        package_versions[n] = v

        return package_versions.get(pkg)

    @staticmethod
    def get_platform():
        """Get platform of current system (system and bitness)."""
        _sys_map = {'linux2': 'linux', 'linux': 'linux',
                    'darwin': 'osx', 'win32': 'win', 'openbsd5': 'openbsd'}

        non_x86_linux_machines = {'armv6l', 'armv7l', 'ppc64le'}
        sys_platform = _sys_map.get(sys.platform, 'unknown')
        bits = 8 * tuple.__itemsize__

        if (sys_platform == 'linux' and
                platform.machine() in non_x86_linux_machines):
            arch_name = platform.machine()
            subdir = 'linux-{0}'.format(arch_name)
        else:
            arch_name = {64: 'x86_64', 32: 'x86'}[bits]
            subdir = '{0}-{1}'.format(sys_platform, bits)

        return subdir

    def load_rc(self, path=None, system=False):
        """
        Load the conda configuration file.

        If both user and system configuration exists, user will be used.
        """
        if os.path.isfile(self.user_rc_path) and not system:
            path = self.user_rc_path
        elif os.path.isfile(self.sys_rc_path):
            path = self.sys_rc_path

        if not path or not os.path.isfile(path):
            return {}

        with open(path) as f:
            return yaml.load(f) or {}

    def get_condarc_channels(self,
                             normalize=False,
                             conda_url='https://conda.anaconda.org',
                             channels=None):
        """Return all the channel urls defined in .condarc.

        If no condarc file is found, use the default channels.
        the `default_channel_alias` key is ignored and only the anaconda client
        `url` key is used.
        """
        # https://docs.continuum.io/anaconda-repository/configuration
        # They can only exist on a system condarc
        default_channels = self.load_rc(system=True).get('default_channels',
                                                         self.DEFAULT_CHANNELS)

        normalized_channels = []
        if channels is None:
            condarc = self.load_rc()
            channels = condarc.get('channels')

            if channels is None:
                channels = ['defaults']

        if normalize:
            template = '{0}/{1}' if conda_url[-1] != '/' else '{0}{1}'
            for channel in channels:
                if channel == 'defaults':
                    normalized_channels += default_channels
                elif channel.startswith('http'):
                    normalized_channels.append(channel)
                else:
                    # Append to the conda_url that comes from anaconda client
                    # default_channel_alias key is deliberately ignored
                    normalized_channels.append(template.format(conda_url,
                                                               channel))
            channels = normalized_channels

        return channels

    # --- Pip commands
    # -------------------------------------------------------------------------
    def _call_pip(self, name=None, prefix=None, extra_args=None,
                  callback=None):
        """Call pip in QProcess worker."""
        cmd_list = self._pip_cmd(name=name, prefix=prefix)
        cmd_list.extend(extra_args)

        process_worker = ProcessWorker(cmd_list, pip=True, callback=callback)
        process_worker.sig_finished.connect(self._start)
        self._queue.append(process_worker)
        self._start()

        return process_worker

    def _pip_cmd(self, name=None, prefix=None):
        """Get pip location based on environment `name` or `prefix`."""
        if (name and prefix) or not (name or prefix):
            raise TypeError("conda pip: exactly one of 'name' ""or 'prefix' "
                            "required.")

        if name and self.environment_exists(name=name):
            prefix = self.get_prefix_envname(name)

        if sys.platform == 'win32':
            python = join(prefix, 'python.exe')  # FIXME:
            pip = join(prefix, 'pip.exe')        # FIXME:
        else:
            python = join(prefix, 'bin/python')
            pip = join(prefix, 'bin/pip')

        cmd_list = [python, pip]

        return cmd_list

    def pip_list(self, name=None, prefix=None, abspath=True):
        """Get list of pip installed packages."""
        if (name and prefix) or not (name or prefix):
            raise TypeError("conda pip: exactly one of 'name' ""or 'prefix' "
                            "required.")

        if name:
            prefix = self.get_prefix_envname(name)

        pip_command = os.sep.join([prefix, 'bin', 'python'])
        cmd_list = [pip_command, PIP_LIST_SCRIPT]
        process_worker = ProcessWorker(cmd_list, pip=True, parse=True,
                                       callback=self._pip_list,
                                       extra_kwargs={'prefix': prefix})
        process_worker.sig_finished.connect(self._start)
        self._queue.append(process_worker)
        self._start()

        return process_worker

    def _pip_list(self, stdout, stderr, prefix=None):
        """Callback for `pip_list`."""
        result = stdout  # A dict
        linked = self.linked(prefix)

        pip_only = []

        linked_names = [self.split_canonical_name(l)[0] for l in linked]

        for pkg in result:
            name = self.split_canonical_name(pkg)[0]
            if name not in linked_names:
                pip_only.append(pkg)
            # FIXME: NEED A MORE ROBUST WAY!
#            if '<pip>' in line and '#' not in line:
#                temp = line.split()[:-1] + ['pip']
#                temp = '-'.join(temp)
#                if '-(' in temp:
#                    start = temp.find('-(')
#                    end = temp.find(')')
#                    substring = temp[start:end+1]
#                    temp = temp.replace(substring, '')
#                result.append(temp)

        return pip_only

    def pip_remove(self, name=None, prefix=None, pkgs=None):
        """Remove a pip package in given environment by `name` or `prefix`."""
        logger.debug(str((prefix, pkgs)))

        if isinstance(pkgs, (list, tuple)):
            pkg = ' '.join(pkgs)
        else:
            pkg = pkgs

        extra_args = ['uninstall', '--yes', pkg]

        return self._call_pip(name=name, prefix=prefix, extra_args=extra_args)

    def pip_search(self, search_string=None):
        """Search for pip packages in PyPI matching `search_string`."""
        extra_args = ['search', search_string]
        return self._call_pip(name='root', extra_args=extra_args,
                              callback=self._pip_search)

        # if stderr:
        #     raise PipError(stderr)
        # You are using pip version 7.1.2, however version 8.0.2 is available.
        # You should consider upgrading via the 'pip install --upgrade pip'
        # command.

    @staticmethod
    def _pip_search(stdout, stderr):
        """Callback for pip search."""
        result = {}
        lines = to_text_string(stdout).split('\n')
        while '' in lines:
            lines.remove('')

        for line in lines:
            if ' - ' in line:
                parts = line.split(' - ')
                name = parts[0].strip()
                description = parts[1].strip()
                result[name] = description

        return result


CONDA_API = None


def CondaAPI():
    """Conda non blocking api."""
    global CONDA_API

    if CONDA_API is None:
        CONDA_API = _CondaAPI()

    return CONDA_API

COUNTER = 0


# --- Local testing
# -----------------------------------------------------------------------------
def ready_print(worker, output, error):  # pragma : no cover
    """Local test helper."""
    global COUNTER
    COUNTER += 1
    print(COUNTER, output, error)


def test():  # pragma : no cover
    """Run local test."""
    from conda_manager.utils.qthelpers import qapplication

    app = qapplication()
    conda_api = CondaAPI()
#    print(conda_api.get_condarc_channels())
#    print(conda_api.get_condarc_channels(normalize=True))
#    print(conda_api.user_rc_path)
#    print(conda_api.sys_rc_path)
#    print(conda_api.load_rc())
    worker = conda_api.config_add('channels', 'goanpeca')
    worker.sig_finished.connect(ready_print)

#    print(conda_api.proxy_servers())
    app.exec_()


if __name__ == '__main__':  # pragma : no cover
    test()

spyder-ide / conda-manager

GitHub Access Token became invalid

Push — master ( cfccd9...b77fc8 )

_CondaAPI.dependencies() D

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like
