Page MenuHomec4science
Files F77858586

phabricator.py
No OneTemporary
Actions

File Metadata

Created
Fri, Aug 16, 20:26

phabricator.py
View Options

import logging
import re
from ... import colored
from ... import export
from ... import dry_do
from ... import Directory
from ...utils import get_phabricator_instance
from .. import color_phid
__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 PhabDirectory(Directory):
def __init__(self, *args, host=None, username=None, token=None, **kwargs):
super().__init__(**kwargs)
self._host = '{0}/api/'.format(host.rstrip('/'))
self._phab = get_phabricator_instance(host=self._host,
username=username,
token=token)
def _set_default_policy(self, phid):
_default_policy = [{"type": "edit", "value": self.whoami},
{"type": "view", "value": "obj.project.members"},
{"type": "join", "value": "no-one"}]
_msg = "Setting default policy for project {0} to {1}".format(
color_phid(phid),
', '.join(["{0}: {1}".format(colored(m["type"], attrs=['bold']),
color_phid(m['value']))
for m in _default_policy]))
_logger.debug(_msg)
if not self._dry_run:
self._phab.project.edit(transactions=_default_policy,
objectIdentifier=phid)
else:
dry_do(_msg)
def color_name(self, name, **kwargs):
regex = re.compile(r'PHID-([A-Z]{4})-.+')
match = regex.match(name)
if match:
return color_phid(name)
else:
return super().color_name(name, **kwargs)
def is_valid_user(self, id):
return self.get_user_name(id) is not None
def is_valid_group(self, id):
return self.get_group_name(id) is not None
def get_users_from_group(self, id):
_res = self._phab.project.search(constraints={'phids': [id]},
attachments={'members': True})
if _res['data']:
return [member['phid'] for member in
_res['data'][0]['attachments']['members']['members']]
return []
def get_group_unique_id(self, name):
_res = self._phab.project.query(names=[name])
if _res['data']:
return list(_res['data'].keys())[0]
return None
def get_user_unique_id(self, email):
_res = self._phab.user.query(emails=[email])
if _res:
return _res[0]['phid']
return None
def get_user_unique_id_from_login(self, name):
_res = self._phab.user.query(emails=[name])
if _res:
return _res[0]['phid']
return None
def get_group_name(self, gid):
_res = self._phab.project.query(phids=[gid])
if _res['data']:
return _res[0]['data'][gid]['name']
return None
def get_user_name(self, uid):
_res = self._phab.user.query(phids=[uid])
if _res:
return _res[0]['realName']
return None
def get_user_email(self, uid):
raise RuntimeError("This information is not accessible")
def create_group(self, name, members=[]):
_unique_members = list(set(members))
_msg = 'Creating group {0} with members [{1}]'.format(
self.color_name(name, type='group'),
', '.join([color_phid(_id) for _id in _unique_members]))
_logger.debug(_msg)
if not self._dry_run:
_res = self._phab.project.create(name=name,
members=_unique_members)
_phid = _res['phid']
else:
_phid = "PHID-PROJ-notarealproject"
self._set_default_policy(_phid)
return _phid
def set_group_users(self, gid, uids):
_unique_uids = list(set(uids))
_msg = 'Setting users {0} as members of group {1}'.format(
', '.join([color_phid(_id) for _id in _unique_uids]),
color_phid(gid))
_logger.debug(_msg)
transactions = [{"type": "members.set", "value": _unique_uids}]
if not self._dry_run:
self._phab.project.edit(transactions=transactions,
objectIdentifier=gid)
else:
dry_do(_msg)
def create_subgroup(self, name, pgid, members=[]):
_unique_members = list(set(members))
_msg = 'Creating group {0} as a subgroup of {1} with members {2}'.format(
colored(name, 'red', attrs=['bold']),
colored(pgid, attrs=['bold']),
', '.join([color_phid(_id) for _id in _unique_members]))
_logger.debug(_msg)
transactions = [{"type": "parent", "value": pgid},
{"type": "name", "value": name}]
if members is not None:
transactions.append({"type": "members.set",
"value": _unique_members})
if not self._dry_run:
_res = self._phab.project.edit(transactions=transactions)
_phid = _res['object']['phid']
else:
_phid = 'PHID-PROJ-notarealproject'
self._set_default_policy(_phid)
return _phid
@property
def whoami(self):
me = self._phab.user.whoami()
return me['phid']

Event Timeline

c4science · Help