Homec4science
Diffusion Phabricator API scripts (static-import) Browse

phab-api/getmystuph/importers/group_importer.py
bdd2d73844a8static-import

Last Change
Raw File

group_importer.py
View Options

# -*- coding: utf-8 -*-
import logging
from .importer import Importer
from .. import export
from .. import colored
from .. import Directory
__author__ = "Nicolas Richart"
__copyright__ = "Copyright (C) 2016, EPFL (Ecole Polytechnique Fédérale " \
"de Lausanne) - SCITAS (Scientific IT and Application " \
"Support)"
__credits__ = ["Nicolas Richart"]
__license__ = "BSD"
__version__ = "0.1"
__maintainer__ = "Nicolas Richart"
__email__ = "nicolas.richart@epfl.ch"
_logger = logging.getLogger(__name__)
@export
class GroupImporter(Importer):
__default_import_scheme = {'type': 'project', 'name': '{original_name}',
'user-import-strategy': 'additive'}
def __init__(self, name, config, **kwargs):
super().__init__(name, config,
self.__default_import_scheme, **kwargs)
if 'user-import-strategy' not in self._config['import-scheme']:
self._config['import-scheme']['user-import-strategy'] = 'additive'
self._in_directory = self._backend_in['directory']
self._out_directory = self._backend_out['directory']
def _create_group(self, name, **kwargs):
_logger.debug('Checking phid for group {0}'.format(
self._out_directory.color_name(name, type='group')))
gid = self._out_directory.get_group_unique_id(name)
if gid is not None:
_logger.debug('Group {0} -> {1}'.format(
self._out_directory.color_name(name, type='group'),
self._out_directory.color_name(gid)))
_logger.warning(
'{0} already exists in \'{1}\' try to update it'
.format(
self._out_directory.color_name(name, type='group'),
self._out_directory.backend_name))
return (gid, False)
else:
gid = self._out_directory.create_group(name, **kwargs)
return (gid, True)
def _get_users_ids(self, name):
gid = self._in_directory.get_group_unique_id(name)
if gid is None:
_msg = '{0} is not a valid group in the directory {1}'.format(
self._out_directory.color_name(name, type='group'),
self._in_directory.backend_name)
_logger.error(_msg)
raise ValueError(_msg)
_logger.debug(' --> group id {0} -> {1}'
.format(self._in_directory.color_name(name,
type='group'),
self._in_directory.color_name(gid)))
_users = self._in_directory.get_users_from_group(gid)
self._user_db.add_users(_users)
_ids = self._user_db.get_user_oids(_users)
return (gid, _ids)
def transfered_name(self, name):
_name = name
import_scheme = self._config['import-scheme']
if 'name' in import_scheme:
_name = import_scheme['name'].format(original_name=name)
return _name
def transfer(self, name):
_colored_name = self._in_directory.color_name(name, type='group')
_logger.info('Locking for group: {0} ({1})'.format(_colored_name,
self._colored_name))
_logger.debug(' --> group info {0}'.format(colored(self._config,
attrs=['bold'])))
import_scheme = self._config['import-scheme']
if import_scheme['type'] == 'ignore':
return
_in_gid, users_ids = self._get_users_ids(name)
_newly_created = True
_name = self.transfered_name(name)
if 'type' not in import_scheme or \
import_scheme['type'] not in ['project', 'sub-project']:
msg = "You should specify a type of " + \
"import-scheme for group {0}".format(_colored_name)
_logger.error(msg)
raise ValueError(msg)
if import_scheme['type'] == 'project':
_out_gid, _newly_created = self._create_group(_name,
members=users_ids)
elif import_scheme['type'] == 'sub-project':
if 'project' not in import_scheme:
_msg = 'To create {0} as a sub-project you ' + \
'have to specify a parent project' \
.format(_colored_name)
_logger.error(_msg)
raise ValueError(_msg)
_project_name = import_scheme['project']
_pgid, _newly_created = self._create_group(_project_name)
_out_gid = self._out_directory.get_group_unique_id(_name)
if _out_gid is not None:
_logger.warning(
'{0} already exists in \'{1}\' try to update it'.format(
self._out_directory.color_name(_name, type='group'),
self._out_directory.backend_name))
_newly_created = False
else:
_out_gid = self._out_directory.create_subgroup(
_name, _pgid, members=users_ids)
_newly_created = True
if not _newly_created:
if import_scheme['user-import-strategy'] == 'additive':
_logger.debug(
'Looking for existing users in {0} ({1})'.format(
self._out_directory.color_name(_name, type='group'),
self._out_directory.color_name(_out_gid)))
_existing_users = \
self._out_directory.get_users_from_group(_out_gid)
if users_ids:
users_ids.extend(_existing_users)
else:
users_ids = _existing_users
self._out_directory.set_group_users(_out_gid, users_ids)
self._user_db.import_group(_in_gid, _out_gid, name, _name)
return _out_gid
c4science · Help