_CondaAPI

Complexity

Total Complexity

176

Size/Duplication

Total Lines	895
Duplicated Lines	5.92 %

Importance

Changes	7
Bugs	0	Features	0

# -*- coding: utf-8 -*-
        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(''))

This code seems to be duplicated in your project.

        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:

This code seems to be duplicated in your project.

            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()