getmystuph_in_phab.py
No OneTemporary
Actions

Subscribers

None

File Metadata

Created: Thu, May 16, 12:50

getmystuph_in_phab.py
View Options

	#!/usr/bin/env python3

	import argparse
	import yaml
	import keyring

	import collections

	import getmystuph
	from getmystuph.backends import _get_class
	from getmystuph import colored
	from getmystuph.utils import get_password
	import getmystuph.importers as getimp

	import logging
	import logging.config

	_logger = logging.getLogger('getmystuph').getChild('main')


	class PhabImporter(object):
	"""
	Parses a YAML configuration file:
	"""
	_config = None

	def __init__(self, args):
	self._config = yaml.load(args.config)
	self._dry_run = args.dry_run

	_logger.debug('ConfigurationParser: {0}'.format(self._config))

	def check_section(sect, config):
	if sect not in config:
	_msg = 'The configuration does not contain ' \
	'a \'{0}\' section'.format(sect)
	_logger.error(_msg)
	raise RuntimeError(_msg)

	check_section('global', self._config)
	check_section('in', self._config['global'])
	check_section('backend', self._config['global']['in'])
	check_section('out', self._config['global'])
	check_section('backend', self._config['global']['out'])

	self._in_backend = self._config['global']['in']['backend']
	self._out_backend = self._config['global']['out']['backend']

	self._keyring = None
	if 'use_keyring' in self._config['global'] and \
	self._config['global']['use_keyring']:
	self._keyring = keyring.get_keyring()

	self._imported_groups = {}

	self._in_directory = \
	getmystuph.Directory(dry_run=self._dry_run,
	keyring=self._keyring,
	**self._config['global']['in'])
	self._config['global']['in']['directory'] = self._in_directory

	self._out_directory = \
	getmystuph.Directory(dry_run=self._dry_run,
	keyring=self._keyring,
	**self._config['global']['out'])
	self._config['global']['out']['directory'] = self._out_directory

	self._users_db = getimp.ImportUserDB(self._in_directory,
	self._out_directory)

	def _import_repository(self, name, info):
	_logger.info('Getting repo {0}'.format(
	getmystuph.Repo.color_name(name)))
	_logger.debug(' --> repo info {0}'.format(colored(info,
	attrs=['bold'])))
	password = get_password(info['backend'],
	info['username'],
	keyring=self._keyring)

	repo = getmystuph.Repo(name, password=password, **info)

	_permissions = repo.permissions
	_logger.debug("Permissions for repo {0}: {1}".format(
	repo.color_name(name),
	colored(_permissions, attrs=['bold'])))
	_users = self._complete_users(repo.directory,
	_permissions.all_users)
	for u in _users:
	print(u)

	if 'import_scheme' in info and \
	info['import_scheme'] != 'all':
	try:
	query = repo.get_query()
	with query:
	print(query.list_branches())
	print(query.list_tags())
	except:
	_logger.warning(
	"No fine grain operation possible for repo {0}".format(
	repo.color_name(name)))

	def __get_full_info(self, info, _type):
	if info is None:
	info = {}

	def _merge_maps(_map, _default):
	for key, value in _default.items():
	if key not in _map:
	_map[key] = value
	elif type(_map[key]) is dict:
	_map[key] = _merge_maps(_map[key], value)
	return _map

	if '__all__' in self._config[_type]:
	_global_conf = self._config[_type]['__all__']
	_merge_maps(info, _global_conf)

	return info

	def import_all(self):
	methods = collections.OrderedDict(
	[('groups', 'GroupImporter'),
	('repositories', 'RepoImporter')])

	if 'groups' in self._config and \
	'__all__' in self._config['groups']:
	_group_importer = getimp.GroupImporter(
	'__all__', self._config['groups']['__all__'],
	backend_in=self._config['global']['in'],
	backend_out=self._config['global']['out'],
	user_db=self._users_db,
	keyring=self._keyring,
	dry_run=self._dry_run)
	else:
	_group_importer = None

	self._users_db.default_importer = _group_importer

	for _type in methods.keys():
	if _type in self._config:
	all_info = self._config[_type]
	if type(all_info) == list:
	all_info = {key: {} for key in all_info}

	for name, info in all_info.items():
	if name == '__all__':
	continue

	try:
	info = self.__get_full_info(info, _type)

	importer = getattr(getimp, methods[_type])(
	name, info,
	backend_in=self._config['global']['in'],
	backend_out=self._config['global']['out'],
	user_db=self._users_db,
	keyring=self._keyring,
	dry_run=self._dry_run)

	importer.transfer(name)
	except RuntimeError as _e:
	_logger.error('An error occurred while trying to'
	' import the {0} {1}: {2}'.format(
	_type, name, _e))

	def list_repositories(self, _type):
	_class = _get_class(_type, self._in_backend)
	_class.list_repositories(
	list_perm=True,
	user_db=self._users_db,
	keyring=self._keyring,
	dry_run=self._dry_run,
	**self._config['global']['in'])

	# Set up cli arguments
	parser = argparse.ArgumentParser(
	description='Import projects into c4science'
	)

	parser.add_argument(
	'--config',
	help='configuration file (YAML)',
	type=argparse.FileType(),
	required=True
	)

	parser.add_argument(
	'--dry-run',
	help='Do not do the creations',
	action='store_true'
	)

	parser.add_argument(
	'-v', '--verbose',
	help='Make the tool verbose',
	action='store_true'
	)

	args = parser.parse_args()

	if args.verbose:
	with open('.logging.conf', 'r') as fh:
	config = yaml.load(fh)
	logging.config.dictConfig(config)
	else:
	logging.basicConfig(level=logging.ERROR)
	imp = PhabImporter(args)

	#imp.list_repositories('git')
	imp.import_all()

	# LocalWords: keyring

getmystuph_in_phab.pyNo OneTemporaryActions

File Metadata

getmystuph_in_phab.pyView Options

Event Timeline

getmystuph_in_phab.py
No OneTemporary
Actions

getmystuph_in_phab.py
View Options