##// END OF EJS Templates
chore(dev): removed appenlight client js
chore(dev): removed appenlight client js

File last commit:

r5095:aa627a5f default
r5240:4378f944 default
Show More
user_group.py
752 lines | 28.9 KiB | text/x-python | PythonLexer
copyrights: updated for 2023
r5088 # Copyright (C) 2011-2023 RhodeCode GmbH
project: added all source files and assets
r1 #
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License, version 3
# (only), as published by the Free Software Foundation.
#
# 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 Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
# This program is dual-licensed. If you wish to learn more about the
# RhodeCode Enterprise Edition, including its added features, Support services,
# and proprietary license terms, please see https://rhodecode.com/licenses/
import logging
import traceback
models: major update for python3,...
r5070 from rhodecode.lib.utils2 import safe_str
security: escape always the provided user data like firstname/lastname.
r1780 from rhodecode.lib.exceptions import (
UserGroupAssignedException, RepoGroupAssignmentError)
from rhodecode.lib.utils2 import (
get_current_rhodecode_user, action_logger_generic)
project: added all source files and assets
r1 from rhodecode.model import BaseModel
user-groups: moved get_user_groups from RepoModel to UserGroupModel.
r1676 from rhodecode.model.scm import UserGroupList
pull-request-reviewers: added option to add reviewers by picking an user group for pull requests....
r1678 from rhodecode.model.db import (
api: add consistent permissions_summary data for both user and user_groups that expose...
r2437 joinedload, true, func, User, UserGroupMember, UserGroup,
pull-request-reviewers: added option to add reviewers by picking an user group for pull requests....
r1678 UserGroupRepoToPerm, Permission, UserGroupToPerm, UserUserGroupToPerm,
UserGroupUserGroupToPerm, UserGroupRepoGroupToPerm)
security: escape always the provided user data like firstname/lastname.
r1780
project: added all source files and assets
r1
log = logging.getLogger(__name__)
class UserGroupModel(BaseModel):
cls = UserGroup
def _get_user_group(self, user_group):
return self._get_instance(UserGroup, user_group,
callback=UserGroup.get_by_group_name)
def _create_default_perms(self, user_group):
# create default permission
default_perm = 'usergroup.read'
def_user = User.get_default_user()
for p in def_user.user_perms:
if p.permission.permission_name.startswith('usergroup.'):
default_perm = p.permission.permission_name
break
user_group_to_perm = UserUserGroupToPerm()
user_group_to_perm.permission = Permission.get_by_key(default_perm)
user_group_to_perm.user_group = user_group
models: major update for python3,...
r5070 user_group_to_perm.user = def_user
project: added all source files and assets
r1 return user_group_to_perm
user-groups: rewrote the app to pyramid...
r2068 def update_permissions(
self, user_group, perm_additions=None, perm_updates=None,
perm_deletions=None, check_perms=True, cur_user=None):
project: added all source files and assets
r1 from rhodecode.lib.auth import HasUserGroupPermissionAny
if not perm_additions:
perm_additions = []
if not perm_updates:
perm_updates = []
if not perm_deletions:
perm_deletions = []
req_perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin')
user-groups: rewrote the app to pyramid...
r2068 changes = {
'added': [],
'updated': [],
'deleted': []
}
api: added audit logs for user-group related calls....
r2686 change_obj = user_group.get_api_data()
project: added all source files and assets
r1 # update permissions
for member_id, perm, member_type in perm_updates:
member_id = int(member_id)
if member_type == 'user':
user-groups: rewrote the app to pyramid...
r2068 member_name = User.get(member_id).username
project: added all source files and assets
r1 # this updates existing one
self.grant_user_permission(
user_group=user_group, user=member_id, perm=perm
)
tests: added tests for permission update views to catch obvious form errors.
r2827 elif member_type == 'user_group':
project: added all source files and assets
r1 # check if we have permissions to alter this usergroup
member_name = UserGroup.get(member_id).users_group_name
user-groups: rewrote the app to pyramid...
r2068 if not check_perms or HasUserGroupPermissionAny(
*req_perms)(member_name, user=cur_user):
project: added all source files and assets
r1 self.grant_user_group_permission(
user-groups: rewrote the app to pyramid...
r2068 target_user_group=user_group, user_group=member_id, perm=perm)
tests: added tests for permission update views to catch obvious form errors.
r2827 else:
raise ValueError("member_type must be 'user' or 'user_group' "
"got {} instead".format(member_type))
user-groups: rewrote the app to pyramid...
r2068
api: added audit logs for user-group related calls....
r2686 changes['updated'].append({
'change_obj': change_obj,
'type': member_type, 'id': member_id,
'name': member_name, 'new_perm': perm})
project: added all source files and assets
r1
# set new permissions
for member_id, perm, member_type in perm_additions:
member_id = int(member_id)
if member_type == 'user':
user-groups: rewrote the app to pyramid...
r2068 member_name = User.get(member_id).username
project: added all source files and assets
r1 self.grant_user_permission(
user-groups: rewrote the app to pyramid...
r2068 user_group=user_group, user=member_id, perm=perm)
tests: added tests for permission update views to catch obvious form errors.
r2827 elif member_type == 'user_group':
project: added all source files and assets
r1 # check if we have permissions to alter this usergroup
member_name = UserGroup.get(member_id).users_group_name
user-groups: rewrote the app to pyramid...
r2068 if not check_perms or HasUserGroupPermissionAny(
*req_perms)(member_name, user=cur_user):
project: added all source files and assets
r1 self.grant_user_group_permission(
user-groups: rewrote the app to pyramid...
r2068 target_user_group=user_group, user_group=member_id, perm=perm)
tests: added tests for permission update views to catch obvious form errors.
r2827 else:
raise ValueError("member_type must be 'user' or 'user_group' "
"got {} instead".format(member_type))
user-groups: rewrote the app to pyramid...
r2068
api: added audit logs for user-group related calls....
r2686 changes['added'].append({
'change_obj': change_obj,
'type': member_type, 'id': member_id,
'name': member_name, 'new_perm': perm})
project: added all source files and assets
r1
# delete permissions
for member_id, perm, member_type in perm_deletions:
member_id = int(member_id)
if member_type == 'user':
user-groups: rewrote the app to pyramid...
r2068 member_name = User.get(member_id).username
project: added all source files and assets
r1 self.revoke_user_permission(user_group=user_group, user=member_id)
tests: added tests for permission update views to catch obvious form errors.
r2827 elif member_type == 'user_group':
pull-request-reviewers: added option to add reviewers by picking an user group for pull requests....
r1678 # check if we have permissions to alter this usergroup
project: added all source files and assets
r1 member_name = UserGroup.get(member_id).users_group_name
user-groups: rewrote the app to pyramid...
r2068 if not check_perms or HasUserGroupPermissionAny(
*req_perms)(member_name, user=cur_user):
project: added all source files and assets
r1 self.revoke_user_group_permission(
user-groups: rewrote the app to pyramid...
r2068 target_user_group=user_group, user_group=member_id)
tests: added tests for permission update views to catch obvious form errors.
r2827 else:
raise ValueError("member_type must be 'user' or 'user_group' "
"got {} instead".format(member_type))
user-groups: rewrote the app to pyramid...
r2068
api: added audit logs for user-group related calls....
r2686 changes['deleted'].append({
'change_obj': change_obj,
'type': member_type, 'id': member_id,
'name': member_name, 'new_perm': perm})
user-groups: rewrote the app to pyramid...
r2068 return changes
project: added all source files and assets
r1
def get(self, user_group_id, cache=False):
return UserGroup.get(user_group_id)
def get_group(self, user_group):
return self._get_user_group(user_group)
def get_by_name(self, name, cache=False, case_insensitive=False):
return UserGroup.get_by_group_name(name, cache, case_insensitive)
def create(self, name, description, owner, active=True, group_data=None):
try:
new_user_group = UserGroup()
new_user_group.user = self._get_user(owner)
new_user_group.users_group_name = name
new_user_group.user_group_description = description
new_user_group.users_group_active = active
if group_data:
new_user_group.group_data = group_data
self.sa.add(new_user_group)
perm_obj = self._create_default_perms(new_user_group)
self.sa.add(perm_obj)
self.grant_user_permission(user_group=new_user_group,
user=owner, perm='usergroup.admin')
return new_user_group
except Exception:
log.error(traceback.format_exc())
raise
def _get_memberships_for_user_ids(self, user_group, user_id_list):
members = []
for user_id in user_id_list:
member = self._get_membership(user_group.users_group_id, user_id)
members.append(member)
return members
def _get_added_and_removed_user_ids(self, user_group, user_id_list):
current_members = user_group.members or []
current_members_ids = [m.user.user_id for m in current_members]
added_members = [
user_id for user_id in user_id_list
if user_id not in current_members_ids]
if user_id_list == []:
# all members were deleted
deleted_members = current_members_ids
else:
deleted_members = [
user_id for user_id in current_members_ids
if user_id not in user_id_list]
audit-logs: implemented full audit logs across application....
r1829 return added_members, deleted_members
project: added all source files and assets
r1
def _set_users_as_members(self, user_group, user_ids):
user_group.members = []
self.sa.flush()
members = self._get_memberships_for_user_ids(
user_group, user_ids)
user_group.members = members
self.sa.add(user_group)
def _update_members_from_user_ids(self, user_group, user_ids):
added, removed = self._get_added_and_removed_user_ids(
user_group, user_ids)
self._set_users_as_members(user_group, user_ids)
self._log_user_changes('added to', user_group, added)
self._log_user_changes('removed from', user_group, removed)
audit-logs: implemented full audit logs across application....
r1829 return added, removed
project: added all source files and assets
r1
def _clean_members_data(self, members_data):
if not members_data:
members_data = []
user-groups: new selector for user group memebers....
r1089 members = []
for user in members_data:
uid = int(user['member_user_id'])
if uid not in members and user['type'] in ['new', 'existing']:
members.append(uid)
return members
project: added all source files and assets
r1
api: enable setting sync flag for user groups on create/edit.
r2660 def update(self, user_group, form_data, group_data=None):
project: added all source files and assets
r1 user_group = self._get_user_group(user_group)
if 'users_group_name' in form_data:
user_group.users_group_name = form_data['users_group_name']
if 'users_group_active' in form_data:
user_group.users_group_active = form_data['users_group_active']
if 'user_group_description' in form_data:
user_group.user_group_description = form_data[
'user_group_description']
# handle owner change
if 'user' in form_data:
owner = form_data['user']
py3: remove use of pyramid.compat
r4908 if isinstance(owner, str):
dan
bugfix: fixed #3965 and updated a bunch of api tests which...
r68 owner = User.get_by_username(form_data['user'])
if not isinstance(owner, User):
raise ValueError(
'invalid owner for user group: %s' % form_data['user'])
user_group.user = owner
project: added all source files and assets
r1
audit-logs: implemented full audit logs across application....
r1829 added_user_ids = []
removed_user_ids = []
project: added all source files and assets
r1 if 'users_group_members' in form_data:
members_id_list = self._clean_members_data(
form_data['users_group_members'])
audit-logs: implemented full audit logs across application....
r1829 added_user_ids, removed_user_ids = \
self._update_members_from_user_ids(user_group, members_id_list)
project: added all source files and assets
r1
api: enable setting sync flag for user groups on create/edit.
r2660 if group_data:
new_group_data = {}
new_group_data.update(group_data)
user_group.group_data = new_group_data
project: added all source files and assets
r1 self.sa.add(user_group)
audit-logs: implemented full audit logs across application....
r1829 return user_group, added_user_ids, removed_user_ids
project: added all source files and assets
r1
def delete(self, user_group, force=False):
"""
Deletes repository group, unless force flag is used
raises exception if there are members in that group, else deletes
group and users
:param user_group:
:param force:
"""
user_group = self._get_user_group(user_group)
user-groups: rewrote the app to pyramid...
r2068 if not user_group:
return
project: added all source files and assets
r1 try:
# check if this group is not assigned to repo
assigned_to_repo = [x.repository for x in UserGroupRepoToPerm.query()\
.filter(UserGroupRepoToPerm.users_group == user_group).all()]
# check if this group is not assigned to repo
assigned_to_repo_group = [x.group for x in UserGroupRepoGroupToPerm.query()\
.filter(UserGroupRepoGroupToPerm.users_group == user_group).all()]
if (assigned_to_repo or assigned_to_repo_group) and not force:
assigned = ','.join(map(safe_str,
assigned_to_repo+assigned_to_repo_group))
raise UserGroupAssignedException(
modernize: updates for python3
r5095 'UserGroup assigned to {}'.format(assigned))
project: added all source files and assets
r1 self.sa.delete(user_group)
except Exception:
log.error(traceback.format_exc())
raise
def _log_user_changes(self, action, user_group, user_or_users):
users = user_or_users
if not isinstance(users, (list, tuple)):
users = [users]
logging: use logger_generic in all suitable places, and fix get_current_rhodecode_user()...
r2108
project: added all source files and assets
r1 group_name = user_group.users_group_name
for user_or_user_id in users:
user = self._get_user(user_or_user_id)
log_text = 'User {user} {action} {group}'.format(
action=action, user=user.username, group=group_name)
logging: use logger_generic in all suitable places, and fix get_current_rhodecode_user()...
r2108 action_logger_generic(log_text)
project: added all source files and assets
r1
def _find_user_in_group(self, user, user_group):
user_group_member = None
for m in user_group.members:
if m.user_id == user.user_id:
# Found this user's membership row
user_group_member = m
break
return user_group_member
def _get_membership(self, user_group_id, user_id):
user_group_member = UserGroupMember(user_group_id, user_id)
return user_group_member
def add_user_to_group(self, user_group, user):
user_group = self._get_user_group(user_group)
user = self._get_user(user)
user_member = self._find_user_in_group(user, user_group)
if user_member:
# user already in the group, skip
return True
member = self._get_membership(
user_group.users_group_id, user.user_id)
user_group.members.append(member)
try:
self.sa.add(member)
except Exception:
# what could go wrong here?
log.error(traceback.format_exc())
raise
self._log_user_changes('added to', user_group, user)
return member
def remove_user_from_group(self, user_group, user):
user_group = self._get_user_group(user_group)
user = self._get_user(user)
user_group_member = self._find_user_in_group(user, user_group)
if not user_group_member:
# User isn't in that group
return False
try:
self.sa.delete(user_group_member)
except Exception:
log.error(traceback.format_exc())
raise
self._log_user_changes('removed from', user_group, user)
return True
def has_perm(self, user_group, perm):
user_group = self._get_user_group(user_group)
perm = self._get_perm(perm)
return UserGroupToPerm.query()\
.filter(UserGroupToPerm.users_group == user_group)\
.filter(UserGroupToPerm.permission == perm).scalar() is not None
def grant_perm(self, user_group, perm):
user_group = self._get_user_group(user_group)
perm = self._get_perm(perm)
# if this permission is already granted skip it
_perm = UserGroupToPerm.query()\
.filter(UserGroupToPerm.users_group == user_group)\
.filter(UserGroupToPerm.permission == perm)\
.scalar()
if _perm:
return
new = UserGroupToPerm()
new.users_group = user_group
new.permission = perm
self.sa.add(new)
return new
def revoke_perm(self, user_group, perm):
user_group = self._get_user_group(user_group)
perm = self._get_perm(perm)
obj = UserGroupToPerm.query()\
.filter(UserGroupToPerm.users_group == user_group)\
.filter(UserGroupToPerm.permission == perm).scalar()
if obj:
self.sa.delete(obj)
def grant_user_permission(self, user_group, user, perm):
"""
Grant permission for user on given user group, or update
existing one if found
:param user_group: Instance of UserGroup, users_group_id,
or users_group_name
:param user: Instance of User, user_id or username
:param perm: Instance of Permission, or permission_name
"""
api: added audit logs for user-group related calls....
r2686 changes = {
'added': [],
'updated': [],
'deleted': []
}
project: added all source files and assets
r1
user_group = self._get_user_group(user_group)
user = self._get_user(user)
permission = self._get_perm(perm)
api: added audit logs for user-group related calls....
r2686 perm_name = permission.permission_name
member_id = user.user_id
member_name = user.username
project: added all source files and assets
r1
# check if we have that permission already
obj = self.sa.query(UserUserGroupToPerm)\
.filter(UserUserGroupToPerm.user == user)\
.filter(UserUserGroupToPerm.user_group == user_group)\
.scalar()
if obj is None:
# create new !
obj = UserUserGroupToPerm()
obj.user_group = user_group
obj.user = user
obj.permission = permission
self.sa.add(obj)
log.debug('Granted perm %s to %s on %s', perm, user, user_group)
action_logger_generic(
'granted permission: {} to user: {} on usergroup: {}'.format(
perm, user, user_group), namespace='security.usergroup')
api: added audit logs for user-group related calls....
r2686 changes['added'].append({
'change_obj': user_group.get_api_data(),
'type': 'user', 'id': member_id,
'name': member_name, 'new_perm': perm_name})
return changes
project: added all source files and assets
r1
def revoke_user_permission(self, user_group, user):
"""
Revoke permission for user on given user group
:param user_group: Instance of UserGroup, users_group_id,
or users_group name
:param user: Instance of User, user_id or username
"""
api: added audit logs for user-group related calls....
r2686 changes = {
'added': [],
'updated': [],
'deleted': []
}
project: added all source files and assets
r1
user_group = self._get_user_group(user_group)
user = self._get_user(user)
api: added audit logs for user-group related calls....
r2686 perm_name = 'usergroup.none'
member_id = user.user_id
member_name = user.username
project: added all source files and assets
r1
obj = self.sa.query(UserUserGroupToPerm)\
.filter(UserUserGroupToPerm.user == user)\
.filter(UserUserGroupToPerm.user_group == user_group)\
.scalar()
if obj:
self.sa.delete(obj)
log.debug('Revoked perm on %s on %s', user_group, user)
action_logger_generic(
'revoked permission from user: {} on usergroup: {}'.format(
user, user_group), namespace='security.usergroup')
api: added audit logs for user-group related calls....
r2686 changes['deleted'].append({
'change_obj': user_group.get_api_data(),
'type': 'user', 'id': member_id,
'name': member_name, 'new_perm': perm_name})
return changes
project: added all source files and assets
r1 def grant_user_group_permission(self, target_user_group, user_group, perm):
"""
Grant user group permission for given target_user_group
:param target_user_group:
:param user_group:
:param perm:
"""
api: added audit logs for user-group related calls....
r2686 changes = {
'added': [],
'updated': [],
'deleted': []
}
project: added all source files and assets
r1 target_user_group = self._get_user_group(target_user_group)
user_group = self._get_user_group(user_group)
permission = self._get_perm(perm)
api: added audit logs for user-group related calls....
r2686 perm_name = permission.permission_name
member_id = user_group.users_group_id
member_name = user_group.users_group_name
project: added all source files and assets
r1 # forbid assigning same user group to itself
if target_user_group == user_group:
raise RepoGroupAssignmentError('target repo:%s cannot be '
'assigned to itself' % target_user_group)
# check if we have that permission already
obj = self.sa.query(UserGroupUserGroupToPerm)\
.filter(UserGroupUserGroupToPerm.target_user_group == target_user_group)\
.filter(UserGroupUserGroupToPerm.user_group == user_group)\
.scalar()
if obj is None:
# create new !
obj = UserGroupUserGroupToPerm()
obj.user_group = user_group
obj.target_user_group = target_user_group
obj.permission = permission
self.sa.add(obj)
log.debug(
'Granted perm %s to %s on %s', perm, target_user_group, user_group)
action_logger_generic(
'granted permission: {} to usergroup: {} on usergroup: {}'.format(
perm, user_group, target_user_group),
namespace='security.usergroup')
api: added audit logs for user-group related calls....
r2686 changes['added'].append({
'change_obj': target_user_group.get_api_data(),
'type': 'user_group', 'id': member_id,
'name': member_name, 'new_perm': perm_name})
return changes
project: added all source files and assets
r1
def revoke_user_group_permission(self, target_user_group, user_group):
"""
Revoke user group permission for given target_user_group
:param target_user_group:
:param user_group:
"""
api: added audit logs for user-group related calls....
r2686 changes = {
'added': [],
'updated': [],
'deleted': []
}
project: added all source files and assets
r1 target_user_group = self._get_user_group(target_user_group)
user_group = self._get_user_group(user_group)
api: added audit logs for user-group related calls....
r2686 perm_name = 'usergroup.none'
member_id = user_group.users_group_id
member_name = user_group.users_group_name
project: added all source files and assets
r1
obj = self.sa.query(UserGroupUserGroupToPerm)\
.filter(UserGroupUserGroupToPerm.target_user_group == target_user_group)\
.filter(UserGroupUserGroupToPerm.user_group == user_group)\
.scalar()
if obj:
self.sa.delete(obj)
log.debug(
'Revoked perm on %s on %s', target_user_group, user_group)
action_logger_generic(
'revoked permission from usergroup: {} on usergroup: {}'.format(
user_group, target_user_group),
namespace='security.repogroup')
api: added audit logs for user-group related calls....
r2686 changes['deleted'].append({
'change_obj': target_user_group.get_api_data(),
'type': 'user_group', 'id': member_id,
'name': member_name, 'new_perm': perm_name})
return changes
api: add consistent permissions_summary data for both user and user_groups that expose...
r2437 def get_perms_summary(self, user_group_id):
permissions = {
'repositories': {},
'repositories_groups': {},
}
ugroup_repo_perms = UserGroupRepoToPerm.query()\
.options(joinedload(UserGroupRepoToPerm.permission))\
.options(joinedload(UserGroupRepoToPerm.repository))\
.filter(UserGroupRepoToPerm.users_group_id == user_group_id)\
.all()
for gr in ugroup_repo_perms:
permissions['repositories'][gr.repository.repo_name] \
= gr.permission.permission_name
ugroup_group_perms = UserGroupRepoGroupToPerm.query()\
.options(joinedload(UserGroupRepoGroupToPerm.permission))\
.options(joinedload(UserGroupRepoGroupToPerm.group))\
.filter(UserGroupRepoGroupToPerm.users_group_id == user_group_id)\
.all()
for gr in ugroup_group_perms:
permissions['repositories_groups'][gr.group.group_name] \
= gr.permission.permission_name
return permissions
project: added all source files and assets
r1 def enforce_groups(self, user, groups, extern_type=None):
user = self._get_user(user)
current_groups = user.group_member
user-groups: fix potential problem with group sync of external plugins....
r2193 # find the external created groups, i.e automatically created
log.debug('Enforcing user group set `%s` on user %s', groups, user)
project: added all source files and assets
r1 # calculate from what groups user should be removed
user-groups: fix potential problem with group sync of external plugins....
r2193 # external_groups that are not in groups
for gr in [x.users_group for x in current_groups]:
managed = gr.group_data.get('extern_type')
if managed:
if gr.users_group_name not in groups:
log.debug('Removing user %s from user group %s. '
'Group sync managed by: %s', user, gr, managed)
self.remove_user_from_group(gr, user)
else:
log.debug('Skipping removal from group %s since it is '
logging: use lazy parameter evaluation in log calls.
r3061 'not set to be automatically synchronized', gr)
project: added all source files and assets
r1
# now we calculate in which groups user should be == groups params
refactor: renamed get_first_admin to get_first_super_admin which...
r278 owner = User.get_first_super_admin().username
project: added all source files and assets
r1 for gr in set(groups):
existing_group = UserGroup.get_by_group_name(gr)
if not existing_group:
desc = 'Automatically created from plugin:%s' % extern_type
# we use first admin account to set the owner of the group
user-groups: added info, and new panel to control user group synchronization....
r1600 existing_group = UserGroupModel().create(
gr, desc, owner, group_data={'extern_type': extern_type})
project: added all source files and assets
r1
user-groups: fix potential problem with group sync of external plugins....
r2193 # we can only add users to groups which have set sync flag via
# extern_type attribute.
# This is either set and created via plugins, or manually
managed = existing_group.group_data.get('extern_type')
project: added all source files and assets
r1 if managed:
log.debug('Adding user %s to user group %s', user, gr)
UserGroupModel().add_user_to_group(existing_group, user)
else:
log.debug('Skipping addition to group %s since it is '
logging: use lazy parameter evaluation in log calls.
r3061 'not set to be automatically synchronized', gr)
Bartłomiej Wołyńczyk
admin-users: add view for user groups managment...
r1556
def change_groups(self, user, groups):
"""
This method changes user group assignment
:param user: User
:param groups: array of UserGroupModel
"""
user = self._get_user(user)
log.debug('Changing user(%s) assignment to groups(%s)', user, groups)
current_groups = user.group_member
current_groups = [x.users_group for x in current_groups]
# calculate from what groups user should be removed/add
groups = set(groups)
current_groups = set(current_groups)
groups_to_remove = current_groups - groups
groups_to_add = groups - current_groups
users: ported controllers from pylons into pyramid views.
r2114 removed_from_groups = []
added_to_groups = []
Bartłomiej Wołyńczyk
admin-users: add view for user groups managment...
r1556 for gr in groups_to_remove:
user-groups: rewrote the app to pyramid...
r2068 log.debug('Removing user %s from user group %s',
user.username, gr.users_group_name)
users: ported controllers from pylons into pyramid views.
r2114 removed_from_groups.append(gr.users_group_id)
Bartłomiej Wołyńczyk
admin-users: add view for user groups managment...
r1556 self.remove_user_from_group(gr.users_group_name, user.username)
for gr in groups_to_add:
user-groups: rewrote the app to pyramid...
r2068 log.debug('Adding user %s to user group %s',
user.username, gr.users_group_name)
users: ported controllers from pylons into pyramid views.
r2114 added_to_groups.append(gr.users_group_id)
user-groups: rewrote the app to pyramid...
r2068 UserGroupModel().add_user_to_group(
gr.users_group_name, user.username)
Bartłomiej Wołyńczyk
admin-users: add view for user groups managment...
r1556
users: ported controllers from pylons into pyramid views.
r2114 return added_to_groups, removed_from_groups
Bartłomiej Wołyńczyk
admin-users: add view for user groups managment...
r1556
pull-request-reviewers: added option to add reviewers by picking an user group for pull requests....
r1678 def _serialize_user_group(self, user_group):
import rhodecode.lib.helpers as h
return {
'id': user_group.users_group_id,
# TODO: marcink figure out a way to generate the url for the
# icon
'icon_link': '',
'value_display': 'Group: %s (%d members)' % (
user_group.users_group_name, len(user_group.members),),
'value': user_group.users_group_name,
'description': user_group.user_group_description,
'owner': user_group.user.username,
'owner_icon': h.gravatar_url(user_group.user.email, 30),
'value_display_owner': h.person(user_group.user.email),
'value_type': 'user_group',
'active': user_group.users_group_active,
}
user-groups: moved get_user_groups from RepoModel to UserGroupModel.
r1676 def get_user_groups(self, name_contains=None, limit=20, only_active=True,
expand_groups=False):
query = self.sa.query(UserGroup)
if only_active:
query = query.filter(UserGroup.users_group_active == true())
if name_contains:
modernize: updates for python3
r5095 ilike_expression = f'%{safe_str(name_contains)}%'
user-groups: moved get_user_groups from RepoModel to UserGroupModel.
r1676 query = query.filter(
UserGroup.users_group_name.ilike(ilike_expression))\
.order_by(func.length(UserGroup.users_group_name))\
.order_by(UserGroup.users_group_name)
query = query.limit(limit)
user_groups = query.all()
perm_set = ['usergroup.read', 'usergroup.write', 'usergroup.admin']
user_groups = UserGroupList(user_groups, perm_set=perm_set)
pull-request-reviewers: added option to add reviewers by picking an user group for pull requests....
r1678 # store same serialize method to extract data from User
from rhodecode.model.user import UserModel
serialize_user = UserModel()._serialize_user
user-groups: moved get_user_groups from RepoModel to UserGroupModel.
r1676
pull-request-reviewers: added option to add reviewers by picking an user group for pull requests....
r1678 _groups = []
for group in user_groups:
entry = self._serialize_user_group(group)
if expand_groups:
expanded_members = []
for member in group.members:
expanded_members.append(serialize_user(member.user))
entry['members'] = expanded_members
_groups.append(entry)
user-groups: moved get_user_groups from RepoModel to UserGroupModel.
r1676 return _groups
Bartłomiej Wołyńczyk
admin-users: add view for user groups managment...
r1556 @staticmethod
def get_user_groups_as_dict(user_group):
import rhodecode.lib.helpers as h
data = {
'users_group_id': user_group.users_group_id,
user-groups: expose public profile link to user group in membership view
r2793 'group_name': h.link_to_group(user_group.users_group_name),
Bartłomiej Wołyńczyk
admin-users: add view for user groups managment...
r1556 'group_description': user_group.user_group_description,
'active': user_group.users_group_active,
"owner": user_group.user.username,
'owner_icon': h.gravatar_url(user_group.user.email, 30),
user-groups: moved get_user_groups from RepoModel to UserGroupModel.
r1676 "owner_data": {
'owner': user_group.user.username,
'owner_icon': h.gravatar_url(user_group.user.email, 30)}
Bartłomiej Wołyńczyk
admin-users: add view for user groups managment...
r1556 }
return data