config.py

#!/opt/cloudlinux/venv/bin/python3 -bb
# coding=utf-8
#
# Copyright  Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#

import json
import os
import time
from typing import Dict, Any, Optional  # NOQA

from clwizard.constants import ModuleStatus, MODULES_STATUS_FILE, MAIN_LOG_PATH
from clwizard.utils import atomic_write, setup_logger
from .exceptions import (
    NoSuchModule,
    MalformedConfigError,
)

class Config:
    """
    Low-level logic of interaction with 'states' file
    """

def __init__(self):
        # useful for IDE-level auto-completion and type checking
        class Cfg:
            worker_pid = None  # type: int
            modules = {}  # type: Dict[str, Dict]
        self.Cfg = Cfg
        self.log = setup_logger('wizard.config', MAIN_LOG_PATH)

self.reload()

def set_modules(self, modules):
        # type: (Dict[str, Dict]) -> None
        """
        Forget about the previous config and create a new one with specified
        modules and instructions.

{'cagefs': {'options': {'enable_for_new_users': True, 'enable_for_existing_users': True}},
         'mod_lsapi': {'options': {'use_beta_for_da': True}},
         'mysql_governor': {'options': {'mode': 'single'}},
         'php': {'options': {'install_versions': ['5.2', '5.3'],
                                                 'install_modules': False, 'default_version': '5.3'}},
         'nodejs': {'options': {'versions': ['7.4', '8.9'], 'default_version': '8.9'}},
         'python': {'options': {'versions': ['2.7', '3.6']}},
         'ruby': {'options': {'versions': ['2.7']}}}
        """
        self.Cfg.modules = {}
        for module_name, instructions in modules.items():
            self.Cfg.modules[module_name] = {
                'status': ModuleStatus.PENDING,
                'options': instructions.get('options', {})
            }

@property
    def statuses(self):
        # type: () -> Dict[str, str]
        """Get dictionary with modules statuses"""
        return {module: options['status'] for module, options in self.Cfg.modules.items()}

def get_module_options(self, module_name):
        # type: (str) -> Dict[str, Dict[str, Any]]
        """Get modules options (dictionary)"""
        try:
            return self.Cfg.modules[module_name].get('options', {})
        except KeyError as e:
            raise NoSuchModule(module_name) from e

def get_module_status(self, module_name):
        # type: (str) -> str
        """Get modules states in format (see ModuleStatus)"""
        try:
            return self.Cfg.modules[module_name]['status']
        except KeyError as e:
            raise NoSuchModule(module_name) from e

def get_module_status_time(self, module_name):
        # type: (str) -> str
        """Get modules states in format (see ModuleStatus)"""
        try:
            return self.Cfg.modules[module_name].get('status_time')
        except KeyError as e:
            raise NoSuchModule(module_name) from e

def set_module_status(self, module_name, new_state):
        # type: (str, str) -> None
        """Set new module state"""
        if module_name not in self.Cfg.modules:
            raise NoSuchModule(module_name)
        self.Cfg.modules[module_name]['status'] = new_state
        self.Cfg.modules[module_name]['status_time'] = time.time()

@property
    def worker_pid(self):
        # type: () -> Optional[int]
        """Get background worker process id"""
        return self.Cfg.worker_pid

@worker_pid.setter
    def worker_pid(self, new_pid):
        # type: (int) -> None
        """Set new background worker process id"""
        self.Cfg.worker_pid = new_pid

def _reset_cfg(self):
        """
        Reset self.Cfg object to default values before it will be loaded
        from file as a part of self.reload()
        """
        # reset public class attributes to defaults
        for k, v in self.Cfg.__dict__.items():
            if not k.startswith('__'):
                setattr(self.Cfg, k, None)
        self.Cfg.modules = {}

def reload(self):
        """
        Reset config object and load data from json config.

:raises MalformedConfigError: cannot parse json config
        """
        self._reset_cfg()

json_data = self._read_json_config()
        if json_data is None:
            return  # No file - nothing to load, use defaults

self.Cfg.worker_pid = json_data['pid']
        for module_name, info in json_data['modules'].items():
            self.Cfg.modules[module_name] = {
                'status': info['status'],
                'status_time': info.get('status_time'),
                'options': info.get('options', {})
            }

def save(self):
        """Dump python object state to file"""
        state = {
            'pid': self.Cfg.worker_pid,
            'modules': self.Cfg.modules
        }
        self._write_json_config(state)

def _read_json_config(self):
        # type: () -> Optional[Dict]
        """
        Load state config and parse it using json

:raises MalformedConfigError: cannot parse json config
        """
        if not os.path.exists(MODULES_STATUS_FILE):
            return None

try:
            with open(MODULES_STATUS_FILE, encoding='utf-8') as f:
                return json.load(f)
        except (IOError, OSError) as e:
            self.log.error("Unable to load config file due to error %s", str(e))
            return None
        except (TypeError, ValueError) as e:
            self.log.error('Unable to load json config file, %s', str(e))
            raise MalformedConfigError(config_path=MODULES_STATUS_FILE) from e

@staticmethod
    def _write_json_config(schema):
        # type: (Dict) -> None
        """Write data to file using atomic write"""
        with atomic_write(MODULES_STATUS_FILE) as f:
            json.dump(schema, f, indent=2)

docroot: /home/induvedam/public_html

curdir: /opt/cloudlinux/venv/lib/python3.11/site-packages/clwizard/config

	.	82
	..	175
X	__init__.py	3145
	__pycache__	4096
X	config.py	6022
X	exceptions.py	1354