Page MenuHomec4science
Files F63372158

group_importer.py
No OneTemporary
Actions

File Metadata

Created
Sun, May 19, 16:02

group_importer.py
View Options

# -*- coding: utf-8 -*-
import logging
from . 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, user_db, **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 = Directory(**self._config, **kwargs)
self._out_directory = user_db.directory
self._db = user_db
def _create_group(self, name, **kwargs):
_logger.debug('Checking phid for group {0}'.format(name))
gid = self._out_directory.get_group_unique_id(name)
if gid != '':
_logger.debug('Group {0} -> {1}'.format(name, gid))
_logger.warning(
'{0} already exists in \'{1}\' try to update it'
.format(name, self._config['import-scheme']['backend']))
return (gid, False)
else:
gid = self._out_directory.create_group(name, **kwargs)
return (gid, True)
def _get_users_ids(self):
gid = self._in_directory.get_group_unique_id(self._name)
if gid == '':
_msg = '{0} is not a valid group in the directory {1}'.format(
self._colored_name, self._in_directory.backend_name)
_logger.error(_msg)
raise ValueError(_msg)
_logger.debug(' --> group id {0} -> {1}'
.format(self._colored_name,
colored(gid, attrs=['bold'])))
_users = self._in_directory.get_users_from_group(gid)
self._db.add_users(_users, self._in_directory)
return self._db.get_user_oids(_users)
def transfer(self):
_logger.info('Locking for group: {0}'.format(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
users_ids = self._get_users_ids()
_name = self._name
_newly_created = True
if 'name' in import_scheme:
_name = import_scheme['name'].format(original_name=self._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(self._colored_name)
_logger.error(msg)
raise ValueError(msg)
if import_scheme['type'] == 'project':
_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(self._colored_name)
_logger.error(_msg)
raise ValueError(_msg)
_project_name = import_scheme['project']
_pgid, _newly_created = self._create_group(_project_name)
gid = self._out_directory.get_group_unique_id(_name)
if gid != '':
_logger.warning(
'{0} already exists in \'{1}\' try to update it'
.format(_name, self._config['import-scheme']['backend']))
_newly_created = False
else:
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(_name,
gid))
_existing_users = self._out_directory.get_users_from_group(gid)
if users_ids:
users_ids.extend(_existing_users)
else:
users_ids = _existing_users
self._out_directory.set_group_users(gid, users_ids)

Event Timeline

c4science · Help