golden hour

/opt/imunify360/venv/lib/python3.11/site-packages/vendors_api

File/Folder	Size	Actions
Readme.md	4.13 KB
__init__.py	413 B
config.py	2.52 KB
exceptions.py	3.01 KB
models.py	7.38 KB
parser.py	13.48 KB
schemas	-

Edit: 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)