[EDIT] FILE: models.py

# 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 # from typing import NamedTuple, Dict, List, Optional # NOQA class _Base: __slots__ = ('_received_fields', '_api') def __init__(self, opts): """ Initializes class by given dictionary and api object. :type opts: dict """ # required for subclasses to know # which fields we really received from # API and which are just None # e.g. case when we asked vendor's script to # return username & package and tried to access # 'id' somewhere (see __getattribute__ implementation) self._received_fields = set(opts.keys()) def __getattribute__(self, item): """ When parsing data in __init__, we save list of received fields. When accesing any of those fields, we must check that is was received first. This is needed for dynamic instances of "user" which can be different depending on 'fields' argument passed to the integration script. :type item: str :return: object or raise exception """ if item == '__slots__' or item not in self.__slots__ or item in self._received_fields: return object.__getattribute__(self, item) raise AttributeError(f'{item} is not set, but used in code') def __repr__(self): class_name = self.__class__.__name__ fields_dict = {k: getattr(self, k) for k in self._received_fields} return f'{class_name} ({fields_dict})' def __eq__(self, other): if self.__slots__ != other.__slots__: return False # models with different scopes cannot be equal if self._received_fields != other._received_fields: return False # loop and check all values for slot in set(self.__slots__) & self._received_fields: if getattr(self, slot) != getattr(other, slot): return False return True def __ne__(self, other): return not self == other class PanelInfo(_Base): __slots__ = ( 'name', 'version', 'user_login_url', 'supported_cl_features' ) _DEPRECATED_FEATURE_UPGRADES = { 'wpos': 'accelerate_wp' } def __init__(self, opts): # optional field default value opts.setdefault('supported_cl_features', None) self.name: str = opts['name'] self.version: str = opts['version'] self.user_login_url: str = opts['user_login_url'] self.supported_cl_features: Optional[Dict[str, bool]] = \ self._upgrade_feature_names(opts['supported_cl_features']) super().__init__(opts) def _upgrade_feature_names(self, features: Dict[str, bool]): """ Automatically convert old feature names into new ones that we defined in this class. e.g. feature 'wpos' is now called 'accelerate_wp' """ if features is None: return features new_features = features.copy() for old_name, new_name in self._DEPRECATED_FEATURE_UPGRADES.items(): if old_name not in new_features: continue new_features[new_name] = new_features[old_name] del new_features[old_name] return new_features DbAccess = NamedTuple('DbAccess', [ ('login', str), ('password', str), ('host', str), ('port', str), ]) DbInfo = NamedTuple('DBInfo', [ ('access', DbAccess), ('mapping', Dict[str, List[str]]) ]) class Databases(_Base): __slots__ = ( 'mysql', ) def __init__(self, opts): self.mysql = None # type: Optional[DbInfo] mysql_raw = opts.get('mysql') if mysql_raw is not None: access = mysql_raw['access'] self.mysql = DbInfo( access=DbAccess(**access), mapping=mysql_raw['mapping'] ) super().__init__(opts) class Package(_Base): __slots__ = ( 'name', 'owner', ) def __init__(self, opts): self.owner = opts['owner'] # type: str self.name = opts['name'] # type: str super().__init__(opts) class User(_Base): __slots__ = ( 'id', 'username', 'owner', 'domain', 'package', 'email', 'locale_code', ) def __init__(self, opts): self.id = opts.get('id') # type: int self.username = opts.get('username') # type: str self.owner = opts.get('owner') # type: str if opts.get('package'): self.package = Package(opts.get('package')) # type: Package else: self.package = None self.email = opts.get('email') # type: str self.domain = opts.get('domain') # type: str self.locale_code = opts.get('locale_code') # type: str super().__init__(opts) class InstalledPHP(_Base): __slots__ = ( 'identifier', 'version', 'modules_dir', 'dir', 'bin', 'ini' ) def __init__(self, opts): self.identifier = opts.get('identifier') # type: str self.version = opts.get('version') # type: str self.modules_dir = opts.get('modules_dir') # type: str self.dir = opts.get('dir') # type: str self.bin = opts.get('bin') # type: str self.ini = opts.get('ini') # type: str super().__init__(opts) class PHPConf(NamedTuple): """ An object representing structure of input PHP configuration for a domain """ version: str # PHP version XY ini_path: str # path to directory with additional ini files is_native: bool = False # is PHP version set in native.conf fpm: Optional[str] = None # FPM service name handler: Optional[str] = None # current handler name php_version_id: Optional[str] = None # php version identifier class DomainData(_Base): __slots__ = [ 'owner', 'document_root', 'is_main', 'php', ] def __init__(self, opts): # optional field default value opts.setdefault('php', None) self.owner = opts['owner'] # type: str self.document_root = opts['document_root'] # type: str self.is_main = opts['is_main'] # type: bool self.php = None # type: Optional[PHPConf] php_conf = opts['php'] if php_conf is not None: self.php = PHPConf(**php_conf) super().__init__(opts) class Reseller(_Base): __slots__ = [ 'id', 'name', 'locale_code', 'email', ] def __init__(self, opts): self.id = opts['id'] # type: str self.name = opts['name'] # type: str self.locale_code = opts['locale_code'] # type: str self.email = opts['email'] # type: Optional[str] super().__init__(opts) class Admin(_Base): __slots__ = [ 'name', 'unix_user', 'locale_code', 'email', 'is_main', ] def __init__(self, opts): self.name = opts['name'] # type: str self.unix_user = opts['unix_user'] # type: Optional[str] self.locale_code = opts['locale_code'] # type: str self.email = opts['email'] # type: Optional[str] self.is_main = opts['is_main'] # type: bool super().__init__(opts)
SAVE CANCEL