Admin Control Panel v2.0

✎ Editing: selector_manager.py ← Back

# coding: utf-8

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
import os
from abc import ABCMeta
from future.utils import iteritems

from clselect.baseclselect import (
    ENABLED_STATUS,
    NOT_INSTALLED_STATUS
)
from .config import BaseSelectorConfig  # NOQA
from .pkgmanager import BasePkgManager  # NOQA
from future.utils import with_metaclass

class BaseSelectorManager(with_metaclass(ABCMeta, object)):
    """Responsible for actual interpreter selector high-level API"""

def __init__(self, cfg, pkg):
        """
        :type cfg: config.BaseSelectorConfig
        :type pkg: pkgmanager.BasePkgManager
        """
        self.pkg = pkg  # type: BasePkgManager
        self.cfg = cfg  # type: BaseSelectorConfig
        self.is_root_user = os.geteuid() == 0

@property
    def selector_enabled(self):
        return self.cfg.selector_enabled

@selector_enabled.setter
    def selector_enabled(self, val):
        self.cfg.selector_enabled = val
        self.cfg.save()

@property
    def default_version(self):
        return self.cfg.get_default_version()

def switch_default_version(self, version):
        self.cfg.set_default_version(version)
        self.cfg.save()

def set_version_status(self, version, status):
        """Disable/Enable particular interpreter version globally"""
        self.cfg.set_version_status(version, status)
        self.cfg.save()

def is_version_enabled(self, version):
        """Check whether particular (installed) version is enabled"""
        enabled = [k for k, v in iteritems(self.cfg.available_versions)
                   if v['status'] == ENABLED_STATUS]
        return version in enabled

def install_version(self, version):
        return self.pkg.install_version(version)

def uninstall_version(self, version):
        if version == self.cfg.get_default_version():
            return "It's not allowed to uninstall default version"
        return self.pkg.remove_version(version)

def get_summary(self, installed_interpreters_only=False):
        all_versions = self.cfg.available_versions
        available_to_install = []

if not installed_interpreters_only and self.is_root_user:
            # Only root can see "in progress" and "not installed yet" versions
            # We may change this in future if we have such requirement
            processed_now = self.pkg.in_progress()
            if processed_now:
                all_versions.update(processed_now)

available_to_install = self.pkg.checkout_available()
            if available_to_install:
                available_to_install = set(available_to_install)
                installed_and_in_progress = set(all_versions.keys())
                diff = {
                    ver: {'base_dir': '', 'status': NOT_INSTALLED_STATUS}
                    for ver in (available_to_install - installed_and_in_progress)
                }
                all_versions.update(diff)

# Prettify output by converting Major version to Full versions
        # if possible
        all_versions = {self.pkg.get_full_version(k): v
                        for k, v in iteritems(all_versions)}
        if "" in all_versions:
            all_versions.pop("")
        return {
            'default_version': self.cfg.get_default_version(),
            'available_versions': all_versions,

# `Updating` means pkg info is not available right now and we
            # already started background process to rebuild that cache.
            # Once it finished status becomes `ready` again.
            # If we run as user it's always `ready`
            'cache_status': ('updating' if available_to_install is None
                             else 'ready'),
            'selector_enabled': self.selector_enabled,
        }

Cancel