Viewing file:  users_manager.py (3.01 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
#!/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()
|