Gecko [ renovalkoronred.com.tr ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
__init__.py	219 B	-rw-r--r--
argparse_utils.py	10.98 KB	-rw-r--r--
dateutil.py	5.9 KB	-rw-r--r--
decorators.py	931 B	-rw-r--r--
func.py	15.82 KB	-rw-r--r--
htpasswd.py	2.15 KB	-rw-r--r--
litespeed.py	6.56 KB	-rw-r--r--
logsetup.py	4.45 KB	-rw-r--r--
proctitle.py	2.81 KB	-rw-r--r--
profiler.py	651 B	-rw-r--r--
progress.py	1.09 KB	-rw-r--r--
sentry.py	6.35 KB	-rw-r--r--
server_status.py	1.33 KB	-rw-r--r--
sizeutil.py	2.7 KB	-rw-r--r--
users_manager.py	3.01 KB	-rw-r--r--

Code Editor : users_manager.py

#!/usr/bin/python
# 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
import logging
from collections import namedtuple

from clcommon import cpapi, FormattedException

from lvestats.lib.commons import func
from typing import Optional, Dict, Generator, Tuple, List  # pylint: disable=unused-import

class UserNotFoundError(FormattedException):
    pass

class User(namedtuple('User', ['username', 'domain', 'reseller'])):
    pass

class UsersInfoManager(object):
    """
    Implements some different functions for user management;
    """
    def __init__(self):
        self.users_cache = dict()  # type: Dict[str, User]

def build_users_cache(self, reseller):
        # type: (Optional[str]) -> None
        """Cache data from cpapi for given reseller"""
        try:
            self.users_cache = dict(self._iter_panel_users(reseller))
        except cpapi.NotSupported:
            logging.info("Control panel API is not implemented, "
                         "some features may not work properly")

def _iter_panel_users(self, reseller):
        # type: (Optional[str]) -> Generator[Tuple[str, User]]
        for login, reseller_, domain in self._iter_panel_users_tuples(reseller):
            yield login, User(username=login, reseller=reseller_, domain=domain)

@staticmethod
    def _iter_panel_users_tuples(reseller):
        # type: (Optional[str]) -> Tuple[str, str, str]
        if reseller is None:
            for login, reseller_, domain in cpapi.cpinfo(keyls=('cplogin', 'reseller', 'dns')):
                yield login, reseller_, domain
        else:
            # TODO: do we really need get_reseller_domains?
            for login, domain in list(func.get_reseller_domains(reseller).items()):
                yield login, reseller, domain

def get_domain(self, username, raise_exc=True):
        # type: (str, bool) -> Optional[str]
        """Get domain for user"""
        try:
            return self.users_cache[username].domain
        except KeyError:
            if raise_exc:
                raise UserNotFoundError({
                    'message': "An error occurred while getting domain for user %(username)s",
                    'context': {'username': username}})
            return None

def get_reseller(self, username, raise_exc=True):
        # type: (str, bool) -> Optional[str]
        """Get reseller for user"""
        try:
            return self.users_cache[username].reseller
        except KeyError:
            if raise_exc:
                raise UserNotFoundError({
                    'message': "An error occurred while getting reseller for user %(username)s",
                    'context': {'username': username}})
            return None

def get_login_list(self):
        # type: () -> List[str]
        """Get list of cached users"""
        return list(self.users_cache.keys())

g_usersManager = UsersInfoManager()

${this.title}

Code Editor : users_manager.py