File Manager

✏️ lect/config.py

# 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 json
import os
from abc import ABCMeta, abstractmethod, abstractproperty
from future.utils import iteritems
from secureio import write_file_via_tempfile

from clselect import utils
from .pkgmanager import BasePkgManager  # NOQA
from . import BaseSelectorError, ENABLED_STATUS, DISABLED_STATUS
from future.utils import with_metaclass

class BaseSelectorConfig(with_metaclass(ABCMeta, object)):
    """
    Base class that responsible for all interaction with CL selector config files
    """

def __init__(self, pkg):
        self.Cfg = self._get_config_object()
        self.pkg = pkg  # type: BasePkgManager

self.reload()

@abstractproperty
    def _config_file(self):
        """Should return path to the config file"""
        raise NotImplementedError()

@abstractmethod
    def _create_config_dirs(self):
        """Should create all needed directories for configs"""
        raise NotImplementedError()

@staticmethod
    def _get_config_object():
        """Override this method to change config parameters"""
        # Useful for IDE-level auto-completion and type checking
        class Cfg:
            # Defaults. None values means that it's not specified in config yet
            # and effective values depends on some logic in class properties
            default_version = None
            selector_enabled = None
            disabled_versions = None

return Cfg

@property
    def is_config_exists(self):
        """Check whether config file exists and is a regular file"""
        return os.path.isfile(self._config_file)

def _dump(self):
        """
        Returns underlying config as a plain dict. It will contain only
        explicitly configured options (e.g. no elements with None values)
        """
        tmp = {}
        for k, v in iteritems(self.Cfg.__dict__):
            if not k.startswith('__') and v is not None:
                tmp[k] = v
        return tmp

def _reset_cfg(self):
        """
        Reset self.Cfg object to all None values before it will be loaded
        from file as a part of self.reload()
        """
        for k, v in iteritems(self.Cfg.__dict__):
            if not k.startswith('__'):
                setattr(self.Cfg, k, None)

def reload(self):
        data = self._read_file_data()
        if not data:
            return  # No file or it's empty - nothing to load, use defaults

try:
            tmp = json.loads(data)
        except (ValueError, TypeError) as e:
            raise BaseSelectorError('Unable to parse json from {} ; Error: {}'
                                    .format(self._config_file, e))

self._reset_cfg()
        for k, v in iteritems(tmp):
            setattr(self.Cfg, k, v)

def _read_file_data(self):
        """
        Should return:
            - whole file data for normal case
            - None if file doesn't exists
            - '' for empty file
        """
        if not self.is_config_exists:
            return None

try:
            with open(self._config_file, 'rb') as fd:
                data = fd.read()
        except (IOError, OSError) as e:
            raise BaseSelectorError('Unable to read data from {} ; Error: {}'
                                    .format(self._config_file, e))
        return data

def save(self):
        if not self.is_config_exists:
            self._create_config_dirs()

data = utils.pretty_json(self._dump())
        return self._write_file_data(data)

def _write_file_data(self, data):
        try:
            write_file_via_tempfile(
                content=data,
                dest_path=self._config_file,
                perm=0o644,
                suffix='_tmp',
            )
        except (IOError, OSError) as e:
            raise BaseSelectorError('Could not write system config ({})'.format(e))

def _ensure_version_installed(self, version):
        if version not in self.pkg.installed_versions:
            raise BaseSelectorError('Version "{}" is not installed'
                                    .format(version))

@property
    def selector_enabled(self):
        """Returns effective selector_enabled value"""
        if self.Cfg.selector_enabled is None:
            # Selector is disabled by default until explicitly enabled by admin
            return False
        return self.Cfg.selector_enabled and bool(self.pkg.installed_versions)

@selector_enabled.setter
    def selector_enabled(self, value):
        if value and not self.pkg.installed_versions:
            raise BaseSelectorError(
                "It's not allowed to enable Selector when "
                "interpreter is not installed")
        self.Cfg.selector_enabled = value

def get_default_version(self):
        # If unspecified - we still return None so Frontend can show this
        # somehow user-friendly
        return self.Cfg.default_version

def set_default_version(self, version):
        if version is None:
            # We allow to reset to 'unspecified' state
            self.Cfg.default_version = None
            return

if version in (self.Cfg.disabled_versions or []):
            raise BaseSelectorError(
                "It's not allowed to set disabled version as the default one")
        self._ensure_version_installed(version)
        self.Cfg.default_version = version

def set_version_status(self, version, new_status):
        disabled_list = self.Cfg.disabled_versions
        if new_status == ENABLED_STATUS:

if disabled_list is not None and version in disabled_list:
                disabled_list.remove(version)
                if len(disabled_list) == 0:
                    self.Cfg.disabled_versions = None

elif new_status == DISABLED_STATUS:

if version == self.get_default_version():
                raise BaseSelectorError("It's not allowed to disable currently "
                                        "default version")

# We explicitly allow to disable even not installed versions too
            # for future usage
            if disabled_list is None:
                self.Cfg.disabled_versions = [version]
            else:
                if version not in disabled_list:
                    disabled_list.append(version)

else:
            raise BaseSelectorError('Unknown version status: "{}"'
                                    .format(new_status))

@abstractproperty
    def available_versions(self):
        raise NotImplementedError()

Batal

⬅ clselect 6 item

Nama	Tipe	Ukuran	Diubah	Aksi
📁__pycache__	dir	—	2026-03-11 19:03	📂 Buka
🐍__init__.py	py	3 KB	2026-02-17 11:01	✏️ 👁️
🐍apps_manager.py	py	25.2 KB	2026-02-17 11:01	✏️ 👁️
🐍config.py	py	6.7 KB	2026-02-17 11:01	✏️ 👁️
🐍pkgmanager.py	py	10.7 KB	2026-02-17 11:01	✏️ 👁️
🐍selector_manager.py	py	3.9 KB	2026-02-17 11:01	✏️ 👁️

✏️ Rename