Homec4science
Diffusion Phabricator API scripts (static-import) Browse

phab-api/getmystuph_in_phab.py
bdd2d73844a8static-import

Last Change
Raw File

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,
cache_file=self._config['global'].pop('user_db_cache', None))
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))
self._users_db.save()
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
c4science · Help