api.py
727 lines
| 23.5 KiB
| text/x-python
|
PythonLexer
r1824 | # -*- coding: utf-8 -*- | |||
""" | ||||
rhodecode.controllers.api | ||||
~~~~~~~~~~~~~~~~~~~~~~~~~ | ||||
API controller for RhodeCode | ||||
:created_on: Aug 20, 2011 | ||||
:author: marcink | ||||
:copyright: (C) 2011-2012 Marcin Kuzminski <marcin@python-works.com> | ||||
:license: GPLv3, see COPYING for more details. | ||||
""" | ||||
# This program is free software; you can redistribute it and/or | ||||
# modify it under the terms of the GNU General Public License | ||||
# as published by the Free Software Foundation; version 2 | ||||
# of the License or (at your opinion) any later version of the license. | ||||
# | ||||
# This program is distributed in the hope that it will be useful, | ||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of | ||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | ||||
# GNU General Public License for more details. | ||||
# | ||||
# You should have received a copy of the GNU General Public License | ||||
# along with this program; if not, write to the Free Software | ||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, | ||||
# MA 02110-1301, USA. | ||||
r1500 | import traceback | |||
import logging | ||||
r1445 | from rhodecode.controllers.api import JSONRPCController, JSONRPCError | |||
r1594 | from rhodecode.lib.auth import HasPermissionAllDecorator, \ | |||
r2151 | HasPermissionAnyDecorator, PasswordGenerator, AuthUser | |||
r1734 | ||||
from rhodecode.model.meta import Session | ||||
r1445 | from rhodecode.model.scm import ScmModel | |||
r2151 | from rhodecode.model.db import User, UsersGroup, Repository | |||
|
r1584 | from rhodecode.model.repo import RepoModel | ||
|
r1586 | from rhodecode.model.user import UserModel | ||
from rhodecode.model.users_group import UsersGroupModel | ||||
r2120 | from rhodecode.lib.utils import map_groups | |||
r1734 | ||||
|
r1593 | log = logging.getLogger(__name__) | ||
r1500 | ||||
r1445 | ||||
|
r1593 | class ApiController(JSONRPCController): | ||
r1445 | """ | |||
API Controller | ||||
|
r1584 | |||
r1445 | Each method needs to have USER as argument this is then based on given | |||
API_KEY propagated as instance of user object | ||||
|
r1584 | |||
r1445 | Preferably this should be first argument also | |||
|
r1584 | |||
Each function should also **raise** JSONRPCError for any | ||||
r1445 | errors that happens | |||
|
r1584 | |||
r1445 | """ | |||
|
r1593 | @HasPermissionAllDecorator('hg.admin') | ||
r1843 | def pull(self, apiuser, repo_name): | |||
r1445 | """ | |||
Dispatch pull action on given repo | ||||
|
r1584 | |||
r1500 | :param user: | |||
r1843 | :param repo_name: | |||
r1445 | """ | |||
r1843 | if Repository.is_valid(repo_name) is False: | |||
raise JSONRPCError('Unknown repo "%s"' % repo_name) | ||||
|
r1587 | |||
r1445 | try: | |||
r1843 | ScmModel().pull_changes(repo_name, self.rhodecode_user.username) | |||
return 'Pulled from %s' % repo_name | ||||
r1445 | except Exception: | |||
r1843 | raise JSONRPCError('Unable to pull changes from "%s"' % repo_name) | |||
r1445 | ||||
|
r1593 | @HasPermissionAllDecorator('hg.admin') | ||
r2010 | def get_user(self, apiuser, userid): | |||
|
r1587 | """" | ||
Get a user by username | ||||
r1793 | :param apiuser: | |||
:param username: | ||||
|
r1587 | """ | ||
r1445 | ||||
r2010 | user = UserModel().get_user(userid) | |||
r1989 | if user is None: | |||
return user | ||||
|
r1587 | |||
r1793 | return dict( | |||
id=user.user_id, | ||||
username=user.username, | ||||
firstname=user.name, | ||||
lastname=user.lastname, | ||||
email=user.email, | ||||
r2506 | emails=user.emails, | |||
r1793 | active=user.active, | |||
admin=user.admin, | ||||
r2151 | ldap_dn=user.ldap_dn, | |||
last_login=user.last_login, | ||||
permissions=AuthUser(user_id=user.user_id).permissions | ||||
r1793 | ) | |||
@HasPermissionAllDecorator('hg.admin') | ||||
def get_users(self, apiuser): | ||||
"""" | ||||
Get all users | ||||
:param apiuser: | ||||
""" | ||||
result = [] | ||||
for user in User.getAll(): | ||||
result.append( | ||||
dict( | ||||
id=user.user_id, | ||||
r1716 | username=user.username, | |||
firstname=user.name, | ||||
lastname=user.lastname, | ||||
email=user.email, | ||||
active=user.active, | ||||
admin=user.admin, | ||||
r2151 | ldap_dn=user.ldap_dn, | |||
last_login=user.last_login, | ||||
r1793 | ) | |||
) | ||||
|
r1586 | return result | ||
|
r1593 | @HasPermissionAllDecorator('hg.admin') | ||
r2008 | def create_user(self, apiuser, username, email, password, firstname=None, | |||
r1950 | lastname=None, active=True, admin=False, ldap_dn=None): | |||
r1500 | """ | |||
r2002 | Create new user | |||
|
r1584 | |||
r1500 | :param apiuser: | |||
:param username: | ||||
:param password: | ||||
r1950 | :param email: | |||
r1500 | :param name: | |||
:param lastname: | ||||
|
r1584 | :param active: | ||
:param admin: | ||||
:param ldap_dn: | ||||
r1500 | """ | |||
r1716 | if User.get_by_username(username): | |||
|
r1593 | raise JSONRPCError("user %s already exist" % username) | ||
|
r1589 | |||
r2008 | if User.get_by_email(email, case_insensitive=True): | |||
raise JSONRPCError("email %s already exist" % email) | ||||
if ldap_dn: | ||||
# generate temporary password if ldap_dn | ||||
password = PasswordGenerator().gen_password(length=8) | ||||
r1500 | try: | |||
r2365 | user = UserModel().create_or_update( | |||
r1843 | username, password, email, firstname, | |||
lastname, active, admin, ldap_dn | ||||
) | ||||
r1749 | Session.commit() | |||
r1843 | return dict( | |||
r2365 | id=user.user_id, | |||
msg='created new user %s' % username, | ||||
user=dict( | ||||
id=user.user_id, | ||||
username=user.username, | ||||
firstname=user.name, | ||||
lastname=user.lastname, | ||||
email=user.email, | ||||
active=user.active, | ||||
admin=user.admin, | ||||
ldap_dn=user.ldap_dn, | ||||
last_login=user.last_login, | ||||
) | ||||
r1843 | ) | |||
r1500 | except Exception: | |||
|
r1593 | log.error(traceback.format_exc()) | ||
raise JSONRPCError('failed to create user %s' % username) | ||||
|
r1587 | |||
|
r1593 | @HasPermissionAllDecorator('hg.admin') | ||
r2009 | def update_user(self, apiuser, userid, username, password, email, | |||
firstname, lastname, active, admin, ldap_dn): | ||||
r2002 | """ | |||
Updates given user | ||||
:param apiuser: | ||||
:param username: | ||||
:param password: | ||||
:param email: | ||||
:param name: | ||||
:param lastname: | ||||
:param active: | ||||
:param admin: | ||||
:param ldap_dn: | ||||
""" | ||||
r2365 | usr = UserModel().get_user(userid) | |||
if not usr: | ||||
raise JSONRPCError("user ID:%s does not exist" % userid) | ||||
r2002 | ||||
try: | ||||
r2507 | user = UserModel().create_or_update( | |||
r2002 | username, password, email, firstname, | |||
lastname, active, admin, ldap_dn | ||||
) | ||||
Session.commit() | ||||
return dict( | ||||
id=usr.user_id, | ||||
r2507 | msg='updated user ID:%s %s' % (user.user_id, user.username), | |||
user=dict( | ||||
id=user.user_id, | ||||
username=user.username, | ||||
firstname=user.name, | ||||
lastname=user.lastname, | ||||
email=user.email, | ||||
active=user.active, | ||||
admin=user.admin, | ||||
ldap_dn=user.ldap_dn, | ||||
last_login=user.last_login, | ||||
) | ||||
r2002 | ) | |||
except Exception: | ||||
log.error(traceback.format_exc()) | ||||
r2365 | raise JSONRPCError('failed to update user %s' % userid) | |||
@HasPermissionAllDecorator('hg.admin') | ||||
def delete_user(self, apiuser, userid): | ||||
"""" | ||||
Deletes an user | ||||
:param apiuser: | ||||
""" | ||||
usr = UserModel().get_user(userid) | ||||
if not usr: | ||||
raise JSONRPCError("user ID:%s does not exist" % userid) | ||||
try: | ||||
UserModel().delete(userid) | ||||
Session.commit() | ||||
return dict( | ||||
id=usr.user_id, | ||||
msg='deleted user ID:%s %s' % (usr.user_id, usr.username) | ||||
) | ||||
except Exception: | ||||
log.error(traceback.format_exc()) | ||||
raise JSONRPCError('failed to delete ID:%s %s' % (usr.user_id, | ||||
usr.username)) | ||||
r2002 | ||||
@HasPermissionAllDecorator('hg.admin') | ||||
|
r1593 | def get_users_group(self, apiuser, group_name): | ||
|
r1587 | """" | ||
Get users group by name | ||||
r1793 | :param apiuser: | |||
:param group_name: | ||||
|
r1587 | """ | ||
|
r1593 | users_group = UsersGroup.get_by_group_name(group_name) | ||
|
r1587 | if not users_group: | ||
return None | ||||
r1445 | ||||
|
r1587 | members = [] | ||
for user in users_group.members: | ||||
user = user.user | ||||
r1594 | members.append(dict(id=user.user_id, | |||
username=user.username, | ||||
firstname=user.name, | ||||
lastname=user.lastname, | ||||
email=user.email, | ||||
active=user.active, | ||||
admin=user.admin, | ||||
ldap=user.ldap_dn)) | ||||
|
r1587 | |||
r1594 | return dict(id=users_group.users_group_id, | |||
r1843 | group_name=users_group.users_group_name, | |||
r1594 | active=users_group.users_group_active, | |||
members=members) | ||||
|
r1587 | |||
|
r1593 | @HasPermissionAllDecorator('hg.admin') | ||
def get_users_groups(self, apiuser): | ||||
|
r1584 | """" | ||
|
r1587 | Get all users groups | ||
|
r1584 | |||
r1793 | :param apiuser: | |||
|
r1584 | """ | ||
|
r1586 | |||
|
r1584 | result = [] | ||
|
r1586 | for users_group in UsersGroup.getAll(): | ||
|
r1587 | members = [] | ||
for user in users_group.members: | ||||
user = user.user | ||||
r1594 | members.append(dict(id=user.user_id, | |||
username=user.username, | ||||
firstname=user.name, | ||||
lastname=user.lastname, | ||||
email=user.email, | ||||
active=user.active, | ||||
admin=user.admin, | ||||
ldap=user.ldap_dn)) | ||||
|
r1587 | |||
r1594 | result.append(dict(id=users_group.users_group_id, | |||
r1843 | group_name=users_group.users_group_name, | |||
r1594 | active=users_group.users_group_active, | |||
members=members)) | ||||
|
r1584 | return result | ||
|
r1593 | @HasPermissionAllDecorator('hg.admin') | ||
r1843 | def create_users_group(self, apiuser, group_name, active=True): | |||
r1500 | """ | |||
Creates an new usergroup | ||||
|
r1584 | |||
r1843 | :param group_name: | |||
r1500 | :param active: | |||
""" | ||||
|
r1586 | |||
r1843 | if self.get_users_group(apiuser, group_name): | |||
raise JSONRPCError("users group %s already exist" % group_name) | ||||
|
r1589 | |||
r1500 | try: | |||
r1843 | ug = UsersGroupModel().create(name=group_name, active=active) | |||
r1749 | Session.commit() | |||
r1594 | return dict(id=ug.users_group_id, | |||
r1843 | msg='created new users group %s' % group_name) | |||
r1500 | except Exception: | |||
|
r1593 | log.error(traceback.format_exc()) | ||
r1843 | raise JSONRPCError('failed to create group %s' % group_name) | |||
|
r1584 | |||
|
r1593 | @HasPermissionAllDecorator('hg.admin') | ||
r1810 | def add_user_to_users_group(self, apiuser, group_name, username): | |||
|
r1584 | """" | ||
r2146 | Add a user to a users group | |||
|
r1584 | |||
r1793 | :param apiuser: | |||
:param group_name: | ||||
r1810 | :param username: | |||
|
r1584 | """ | ||
|
r1586 | try: | ||
|
r1593 | users_group = UsersGroup.get_by_group_name(group_name) | ||
|
r1586 | if not users_group: | ||
|
r1593 | raise JSONRPCError('unknown users group %s' % group_name) | ||
|
r1587 | |||
r1989 | user = User.get_by_username(username) | |||
if user is None: | ||||
r1810 | raise JSONRPCError('unknown user %s' % username) | |||
|
r1587 | |||
|
r1593 | ugm = UsersGroupModel().add_user_to_group(users_group, user) | ||
r1989 | success = True if ugm != True else False | |||
msg = 'added member %s to users group %s' % (username, group_name) | ||||
msg = msg if success else 'User is already in that group' | ||||
r1749 | Session.commit() | |||
r1989 | ||||
return dict( | ||||
id=ugm.users_group_member_id if ugm != True else None, | ||||
success=success, | ||||
msg=msg | ||||
) | ||||
|
r1587 | except Exception: | ||
|
r1593 | log.error(traceback.format_exc()) | ||
r1989 | raise JSONRPCError('failed to add users group member') | |||
@HasPermissionAllDecorator('hg.admin') | ||||
def remove_user_from_users_group(self, apiuser, group_name, username): | ||||
""" | ||||
Remove user from a group | ||||
:param apiuser | ||||
:param group_name | ||||
:param username | ||||
""" | ||||
try: | ||||
users_group = UsersGroup.get_by_group_name(group_name) | ||||
if not users_group: | ||||
raise JSONRPCError('unknown users group %s' % group_name) | ||||
user = User.get_by_username(username) | ||||
if user is None: | ||||
raise JSONRPCError('unknown user %s' % username) | ||||
success = UsersGroupModel().remove_user_from_group(users_group, user) | ||||
msg = 'removed member %s from users group %s' % (username, group_name) | ||||
msg = msg if success else "User wasn't in group" | ||||
Session.commit() | ||||
return dict(success=success, msg=msg) | ||||
except Exception: | ||||
log.error(traceback.format_exc()) | ||||
raise JSONRPCError('failed to remove user from group') | ||||
|
r1587 | |||
|
r1593 | @HasPermissionAnyDecorator('hg.admin') | ||
r2010 | def get_repo(self, apiuser, repoid): | |||
|
r1587 | """" | ||
Get repository by name | ||||
r1793 | :param apiuser: | |||
:param repo_name: | ||||
|
r1587 | """ | ||
r2010 | repo = RepoModel().get_repo(repoid) | |||
r1749 | if repo is None: | |||
r2379 | raise JSONRPCError('unknown repository "%s"' % (repo or repoid)) | |||
|
r1584 | |||
|
r1587 | members = [] | ||
for user in repo.repo_to_perm: | ||||
perm = user.permission.permission_name | ||||
user = user.user | ||||
r1793 | members.append( | |||
dict( | ||||
r2146 | type="user", | |||
r1793 | id=user.user_id, | |||
username=user.username, | ||||
firstname=user.name, | ||||
lastname=user.lastname, | ||||
email=user.email, | ||||
active=user.active, | ||||
admin=user.admin, | ||||
ldap=user.ldap_dn, | ||||
permission=perm | ||||
) | ||||
) | ||||
|
r1587 | for users_group in repo.users_group_to_perm: | ||
perm = users_group.permission.permission_name | ||||
users_group = users_group.users_group | ||||
r1793 | members.append( | |||
dict( | ||||
r2146 | type="users_group", | |||
r1793 | id=users_group.users_group_id, | |||
name=users_group.users_group_name, | ||||
active=users_group.users_group_active, | ||||
permission=perm | ||||
) | ||||
) | ||||
|
r1584 | |||
r1793 | return dict( | |||
id=repo.repo_id, | ||||
r1843 | repo_name=repo.repo_name, | |||
r1793 | type=repo.repo_type, | |||
r2338 | clone_uri=repo.clone_uri, | |||
private=repo.private, | ||||
created_on=repo.created_on, | ||||
r1793 | description=repo.description, | |||
members=members | ||||
) | ||||
|
r1587 | |||
|
r1593 | @HasPermissionAnyDecorator('hg.admin') | ||
def get_repos(self, apiuser): | ||||
|
r1586 | """" | ||
|
r1587 | Get all repositories | ||
|
r1586 | |||
r1793 | :param apiuser: | |||
|
r1586 | """ | ||
|
r1587 | |||
|
r1586 | result = [] | ||
r2338 | for repo in Repository.getAll(): | |||
r1793 | result.append( | |||
dict( | ||||
r2338 | id=repo.repo_id, | |||
repo_name=repo.repo_name, | ||||
type=repo.repo_type, | ||||
clone_uri=repo.clone_uri, | ||||
private=repo.private, | ||||
created_on=repo.created_on, | ||||
description=repo.description, | ||||
r1793 | ) | |||
) | ||||
|
r1586 | return result | ||
r1810 | @HasPermissionAnyDecorator('hg.admin') | |||
def get_repo_nodes(self, apiuser, repo_name, revision, root_path, | ||||
ret_type='all'): | ||||
""" | ||||
returns a list of nodes and it's children | ||||
for a given path at given revision. It's possible to specify ret_type | ||||
to show only files or dirs | ||||
:param apiuser: | ||||
:param repo_name: name of repository | ||||
:param revision: revision for which listing should be done | ||||
:param root_path: path from which start displaying | ||||
:param ret_type: return type 'all|files|dirs' nodes | ||||
""" | ||||
try: | ||||
_d, _f = ScmModel().get_nodes(repo_name, revision, root_path, | ||||
flat=False) | ||||
_map = { | ||||
'all': _d + _f, | ||||
'files': _f, | ||||
'dirs': _d, | ||||
} | ||||
return _map[ret_type] | ||||
except KeyError: | ||||
raise JSONRPCError('ret_type must be one of %s' % _map.keys()) | ||||
except Exception, e: | ||||
raise JSONRPCError(e) | ||||
|
r1593 | @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository') | ||
r1843 | def create_repo(self, apiuser, repo_name, owner_name, description='', | |||
r2006 | repo_type='hg', private=False, clone_uri=None): | |||
|
r1584 | """ | ||
r2006 | Create repository, if clone_url is given it makes a remote clone | |||
|
r1584 | |||
r1793 | :param apiuser: | |||
r1843 | :param repo_name: | |||
r2006 | :param owner_name: | |||
r1793 | :param description: | |||
r2006 | :param repo_type: | |||
r1793 | :param private: | |||
r2006 | :param clone_uri: | |||
|
r1584 | """ | ||
|
r1586 | try: | ||
r1989 | owner = User.get_by_username(owner_name) | |||
if owner is None: | ||||
raise JSONRPCError('unknown user %s' % owner_name) | ||||
|
r1584 | |||
r1843 | if Repository.get_by_repo_name(repo_name): | |||
raise JSONRPCError("repo %s already exist" % repo_name) | ||||
|
r1589 | |||
r2120 | groups = repo_name.split(Repository.url_sep()) | |||
|
r1589 | real_name = groups[-1] | ||
r2120 | # create structure of groups | |||
group = map_groups(repo_name) | ||||
|
r1589 | |||
r1843 | repo = RepoModel().create( | |||
r1793 | dict( | |||
repo_name=real_name, | ||||
r1843 | repo_name_full=repo_name, | |||
r1793 | description=description, | |||
private=private, | ||||
repo_type=repo_type, | ||||
r2120 | repo_group=group.group_id if group else None, | |||
r2006 | clone_uri=clone_uri | |||
r2378 | ) | |||
r1793 | ) | |||
r1749 | Session.commit() | |||
r1843 | ||||
return dict( | ||||
id=repo.repo_id, | ||||
r2378 | msg="Created new repository %s" % (repo.repo_name), | |||
repo=dict( | ||||
id=repo.repo_id, | ||||
repo_name=repo.repo_name, | ||||
type=repo.repo_type, | ||||
clone_uri=repo.clone_uri, | ||||
private=repo.private, | ||||
created_on=repo.created_on, | ||||
description=repo.description, | ||||
) | ||||
r1843 | ) | |||
|
r1584 | except Exception: | ||
|
r1593 | log.error(traceback.format_exc()) | ||
r1843 | raise JSONRPCError('failed to create repository %s' % repo_name) | |||
|
r1584 | |||
|
r1593 | @HasPermissionAnyDecorator('hg.admin') | ||
r2385 | def fork_repo(self, apiuser, repoid): | |||
repo = RepoModel().get_repo(repoid) | ||||
if repo is None: | ||||
raise JSONRPCError('unknown repository "%s"' % (repo or repoid)) | ||||
RepoModel().create_fork(form_data, cur_user) | ||||
@HasPermissionAnyDecorator('hg.admin') | ||||
r2003 | def delete_repo(self, apiuser, repo_name): | |||
""" | ||||
Deletes a given repository | ||||
:param repo_name: | ||||
""" | ||||
if not Repository.get_by_repo_name(repo_name): | ||||
raise JSONRPCError("repo %s does not exist" % repo_name) | ||||
try: | ||||
RepoModel().delete(repo_name) | ||||
Session.commit() | ||||
return dict( | ||||
msg='Deleted repository %s' % repo_name | ||||
) | ||||
except Exception: | ||||
log.error(traceback.format_exc()) | ||||
raise JSONRPCError('failed to delete repository %s' % repo_name) | ||||
@HasPermissionAnyDecorator('hg.admin') | ||||
r2004 | def grant_user_permission(self, apiuser, repo_name, username, perm): | |||
|
r1586 | """ | ||
r1982 | Grant permission for user on given repository, or update existing one | |||
if found | ||||
|
r1584 | |||
r1793 | :param repo_name: | |||
r1810 | :param username: | |||
r1793 | :param perm: | |||
|
r1586 | """ | ||
try: | ||||
r1749 | repo = Repository.get_by_repo_name(repo_name) | |||
if repo is None: | ||||
r1793 | raise JSONRPCError('unknown repository %s' % repo) | |||
|
r1586 | |||
r1982 | user = User.get_by_username(username) | |||
if user is None: | ||||
raise JSONRPCError('unknown user %s' % username) | ||||
|
r1586 | |||
r1982 | RepoModel().grant_user_permission(repo=repo, user=user, perm=perm) | |||
r1749 | Session.commit() | |||
r1793 | return dict( | |||
r1982 | msg='Granted perm: %s for user: %s in repo: %s' % ( | |||
r1810 | perm, username, repo_name | |||
r1793 | ) | |||
) | ||||
|
r1586 | except Exception: | ||
|
r1593 | log.error(traceback.format_exc()) | ||
r1793 | raise JSONRPCError( | |||
'failed to edit permission %(repo)s for %(user)s' % dict( | ||||
r1810 | user=username, repo=repo_name | |||
r1793 | ) | |||
) | ||||
@HasPermissionAnyDecorator('hg.admin') | ||||
r2004 | def revoke_user_permission(self, apiuser, repo_name, username): | |||
r1982 | """ | |||
Revoke permission for user on given repository | ||||
:param repo_name: | ||||
:param username: | ||||
r1793 | """ | |||
r1982 | ||||
try: | ||||
repo = Repository.get_by_repo_name(repo_name) | ||||
if repo is None: | ||||
raise JSONRPCError('unknown repository %s' % repo) | ||||
user = User.get_by_username(username) | ||||
if user is None: | ||||
raise JSONRPCError('unknown user %s' % username) | ||||
RepoModel().revoke_user_permission(repo=repo_name, user=username) | ||||
r1793 | ||||
r1982 | Session.commit() | |||
return dict( | ||||
msg='Revoked perm for user: %s in repo: %s' % ( | ||||
username, repo_name | ||||
) | ||||
) | ||||
except Exception: | ||||
log.error(traceback.format_exc()) | ||||
raise JSONRPCError( | ||||
'failed to edit permission %(repo)s for %(user)s' % dict( | ||||
user=username, repo=repo_name | ||||
) | ||||
) | ||||
@HasPermissionAnyDecorator('hg.admin') | ||||
r2004 | def grant_users_group_permission(self, apiuser, repo_name, group_name, perm): | |||
r1982 | """ | |||
Grant permission for users group on given repository, or update | ||||
existing one if found | ||||
r1793 | :param repo_name: | |||
:param group_name: | ||||
:param perm: | ||||
""" | ||||
try: | ||||
repo = Repository.get_by_repo_name(repo_name) | ||||
if repo is None: | ||||
raise JSONRPCError('unknown repository %s' % repo) | ||||
|
r1586 | |||
r1982 | user_group = UsersGroup.get_by_group_name(group_name) | |||
if user_group is None: | ||||
r1793 | raise JSONRPCError('unknown users group %s' % user_group) | |||
r1982 | RepoModel().grant_users_group_permission(repo=repo_name, | |||
group_name=group_name, | ||||
perm=perm) | ||||
r1793 | Session.commit() | |||
return dict( | ||||
r1982 | msg='Granted perm: %s for group: %s in repo: %s' % ( | |||
r1793 | perm, group_name, repo_name | |||
) | ||||
) | ||||
except Exception: | ||||
log.error(traceback.format_exc()) | ||||
raise JSONRPCError( | ||||
r1982 | 'failed to edit permission %(repo)s for %(usersgr)s' % dict( | |||
usersgr=group_name, repo=repo_name | ||||
r1793 | ) | |||
) | ||||
r1982 | ||||
@HasPermissionAnyDecorator('hg.admin') | ||||
r2004 | def revoke_users_group_permission(self, apiuser, repo_name, group_name): | |||
r1982 | """ | |||
Revoke permission for users group on given repository | ||||
:param repo_name: | ||||
:param group_name: | ||||
""" | ||||
try: | ||||
repo = Repository.get_by_repo_name(repo_name) | ||||
if repo is None: | ||||
raise JSONRPCError('unknown repository %s' % repo) | ||||
user_group = UsersGroup.get_by_group_name(group_name) | ||||
if user_group is None: | ||||
raise JSONRPCError('unknown users group %s' % user_group) | ||||
RepoModel().revoke_users_group_permission(repo=repo_name, | ||||
group_name=group_name) | ||||
Session.commit() | ||||
return dict( | ||||
msg='Revoked perm for group: %s in repo: %s' % ( | ||||
group_name, repo_name | ||||
) | ||||
) | ||||
except Exception: | ||||
log.error(traceback.format_exc()) | ||||
raise JSONRPCError( | ||||
'failed to edit permission %(repo)s for %(usersgr)s' % dict( | ||||
usersgr=group_name, repo=repo_name | ||||
) | ||||
) | ||||