mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Merge pull request #11418 from davelopez/refactor_group_users_api
Refactor group_users API logic into GroupUsersManager
This commit is contained in:
John Chilton
@@ -0,0 +1,113 @@
|
||||
import logging
|
||||
from typing import (
|
||||
Any,
|
||||
Dict,
|
||||
List,
|
||||
Optional,
|
||||
)
|
||||
|
||||
from galaxy import model
|
||||
from galaxy.app import StructuredApp
|
||||
from galaxy.exceptions import (
|
||||
ObjectNotFound,
|
||||
)
|
||||
from galaxy.managers.base import decode_id
|
||||
from galaxy.managers.context import ProvidesAppContext
|
||||
from galaxy.schema.fields import EncodedDatabaseIdField
|
||||
from galaxy.web import url_for
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GroupUsersManager:
|
||||
"""Interface/service object shared by controllers for interacting with group users."""
|
||||
|
||||
def __init__(self, app: StructuredApp) -> None:
|
||||
self._app = app
|
||||
|
||||
def index(self, trans: ProvidesAppContext, group_id: EncodedDatabaseIdField) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
Returns a collection (list) with some information about users associated with the given group.
|
||||
"""
|
||||
group = self._get_group(trans, group_id)
|
||||
rval = []
|
||||
for uga in group.users:
|
||||
group_user = self._serialize_group_user(group_id, uga.user)
|
||||
rval.append(group_user)
|
||||
return rval
|
||||
|
||||
def show(self, trans: ProvidesAppContext, id: EncodedDatabaseIdField, group_id: EncodedDatabaseIdField) -> Dict[str, Any]:
|
||||
"""
|
||||
Returns information about a group user.
|
||||
"""
|
||||
user_id = id
|
||||
group = self._get_group(trans, group_id)
|
||||
user = self._get_user(trans, user_id)
|
||||
group_user = self._get_group_user(trans, group, user)
|
||||
if group_user is None:
|
||||
raise ObjectNotFound(f"User {user.email} not in group {group.name}")
|
||||
|
||||
return self._serialize_group_user(group_id, user)
|
||||
|
||||
def update(self, trans: ProvidesAppContext, id: EncodedDatabaseIdField, group_id: EncodedDatabaseIdField):
|
||||
"""
|
||||
Adds a user to a group.
|
||||
"""
|
||||
user_id = id
|
||||
group = self._get_group(trans, group_id)
|
||||
user = self._get_user(trans, user_id)
|
||||
group_user = self._get_group_user(trans, group, user)
|
||||
if group_user is None:
|
||||
self._add_user_to_group(trans, group, user)
|
||||
|
||||
return self._serialize_group_user(group_id, user)
|
||||
|
||||
def delete(self, trans: ProvidesAppContext, id: EncodedDatabaseIdField, group_id: EncodedDatabaseIdField):
|
||||
"""
|
||||
Removes a user from a group.
|
||||
"""
|
||||
user_id = id
|
||||
group = self._get_group(trans, group_id)
|
||||
user = self._get_user(trans, user_id)
|
||||
group_user = self._get_group_user(trans, group, user)
|
||||
if group_user is None:
|
||||
raise ObjectNotFound(f"User {user.email} not in group {group.name}")
|
||||
self._remove_user_from_group(trans, group_user)
|
||||
return self._serialize_group_user(group_id, user)
|
||||
|
||||
def _get_group(self, trans: ProvidesAppContext, encoded_group_id: EncodedDatabaseIdField) -> Any:
|
||||
decoded_group_id = decode_id(self._app, encoded_group_id)
|
||||
group = trans.sa_session.query(model.Group).get(decoded_group_id)
|
||||
if group is None:
|
||||
raise ObjectNotFound(f"Group with id {encoded_group_id} was not found.")
|
||||
return group
|
||||
|
||||
def _get_user(self, trans: ProvidesAppContext, encoded_user_id: EncodedDatabaseIdField) -> model.User:
|
||||
decoded_user_id = decode_id(self._app, encoded_user_id)
|
||||
user = trans.sa_session.query(model.User).get(decoded_user_id)
|
||||
if user is None:
|
||||
raise ObjectNotFound(f"User with id {encoded_user_id} was not found.")
|
||||
return user
|
||||
|
||||
def _get_group_user(self, trans: ProvidesAppContext, group: model.Group, user: model.User) -> Optional[model.UserGroupAssociation]:
|
||||
return trans.sa_session.query(model.UserGroupAssociation).filter(
|
||||
model.UserGroupAssociation.user == user,
|
||||
model.UserGroupAssociation.group == group
|
||||
).one_or_none()
|
||||
|
||||
def _add_user_to_group(self, trans: ProvidesAppContext, group: model.Group, user: model.User):
|
||||
gra = model.UserGroupAssociation(user, group)
|
||||
trans.sa_session.add(gra)
|
||||
trans.sa_session.flush()
|
||||
|
||||
def _remove_user_from_group(self, trans: ProvidesAppContext, group_user: model.UserGroupAssociation):
|
||||
trans.sa_session.delete(group_user)
|
||||
trans.sa_session.flush()
|
||||
|
||||
def _serialize_group_user(self, encoded_group_id: EncodedDatabaseIdField, user: model.User):
|
||||
encoded_user_id = self._app.security.encode_id(user.id)
|
||||
return {
|
||||
"id": encoded_user_id,
|
||||
"email": user.email,
|
||||
"url": url_for('group_user', group_id=encoded_group_id, id=encoded_user_id)
|
||||
}
|
||||
@@ -3,125 +3,56 @@ API operations on Group objects.
|
||||
"""
|
||||
import logging
|
||||
|
||||
from galaxy import web
|
||||
from galaxy.util import unicodify
|
||||
from galaxy.webapps.base.controller import BaseAPIController, url_for
|
||||
from galaxy.managers.context import ProvidesAppContext
|
||||
from galaxy.managers.group_users import GroupUsersManager
|
||||
from galaxy.schema.fields import EncodedDatabaseIdField
|
||||
from galaxy.web import (
|
||||
expose_api,
|
||||
require_admin,
|
||||
)
|
||||
from galaxy.webapps.base.controller import BaseAPIController
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GroupUsersAPIController(BaseAPIController):
|
||||
|
||||
@web.require_admin
|
||||
@web.legacy_expose_api
|
||||
def index(self, trans, group_id, **kwd):
|
||||
def __init__(self, app):
|
||||
super().__init__(app)
|
||||
self.manager = GroupUsersManager(app)
|
||||
|
||||
@require_admin
|
||||
@expose_api
|
||||
def index(self, trans: ProvidesAppContext, group_id: EncodedDatabaseIdField, **kwd):
|
||||
"""
|
||||
GET /api/groups/{encoded_group_id}/users
|
||||
Displays a collection (list) of groups.
|
||||
"""
|
||||
decoded_group_id = trans.security.decode_id(group_id)
|
||||
try:
|
||||
group = trans.sa_session.query(trans.app.model.Group).get(decoded_group_id)
|
||||
except Exception:
|
||||
group = None
|
||||
if not group:
|
||||
trans.response.status = 400
|
||||
return "Invalid group id ( %s ) specified." % str(group_id)
|
||||
rval = []
|
||||
try:
|
||||
for uga in group.users:
|
||||
user = uga.user
|
||||
encoded_id = trans.security.encode_id(user.id)
|
||||
rval.append(dict(id=encoded_id,
|
||||
email=user.email,
|
||||
url=url_for('group_user', group_id=group_id, id=encoded_id, )))
|
||||
except Exception as e:
|
||||
rval = "Error in group API at listing users"
|
||||
log.error(rval + ": %s", unicodify(e))
|
||||
trans.response.status = 500
|
||||
return rval
|
||||
return self.manager.index(trans, group_id)
|
||||
|
||||
@web.require_admin
|
||||
@web.legacy_expose_api
|
||||
def show(self, trans, id, group_id, **kwd):
|
||||
@require_admin
|
||||
@expose_api
|
||||
def show(self, trans: ProvidesAppContext, id: EncodedDatabaseIdField, group_id: EncodedDatabaseIdField, **kwd):
|
||||
"""
|
||||
GET /api/groups/{encoded_group_id}/users/{encoded_user_id}
|
||||
Displays information about a group user.
|
||||
"""
|
||||
user_id = id
|
||||
decoded_group_id = trans.security.decode_id(group_id)
|
||||
decoded_user_id = trans.security.decode_id(user_id)
|
||||
item = None
|
||||
try:
|
||||
group = trans.sa_session.query(trans.app.model.Group).get(decoded_group_id)
|
||||
user = trans.sa_session.query(trans.app.model.User).get(decoded_user_id)
|
||||
for uga in group.users:
|
||||
if uga.user == user:
|
||||
item = dict(id=user_id,
|
||||
email=user.email,
|
||||
url=url_for('group_user', group_id=group_id, id=user_id)) # TODO Fix This
|
||||
if not item:
|
||||
item = f"user {user.email} not in group {group.name}"
|
||||
except Exception as e:
|
||||
item = f"Error in group_user API group {group.name} user {user.email}"
|
||||
log.error(item + ": %s", unicodify(e))
|
||||
return item
|
||||
return self.manager.show(trans, id, group_id)
|
||||
|
||||
@web.require_admin
|
||||
@web.legacy_expose_api
|
||||
def update(self, trans, id, group_id, **kwd):
|
||||
@require_admin
|
||||
@expose_api
|
||||
def update(self, trans: ProvidesAppContext, id: EncodedDatabaseIdField, group_id: EncodedDatabaseIdField, **kwd):
|
||||
"""
|
||||
PUT /api/groups/{encoded_group_id}/users/{encoded_user_id}
|
||||
Adds a user to a group
|
||||
"""
|
||||
user_id = id
|
||||
decoded_group_id = trans.security.decode_id(group_id)
|
||||
decoded_user_id = trans.security.decode_id(user_id)
|
||||
item = None
|
||||
try:
|
||||
group = trans.sa_session.query(trans.app.model.Group).get(decoded_group_id)
|
||||
user = trans.sa_session.query(trans.app.model.User).get(decoded_user_id)
|
||||
for uga in group.users:
|
||||
if uga.user == user:
|
||||
item = dict(id=user_id,
|
||||
email=user.email,
|
||||
url=url_for('group_user', group_id=group_id, id=user_id))
|
||||
if not item:
|
||||
uga = trans.app.model.UserGroupAssociation(user, group)
|
||||
# Add UserGroupAssociations
|
||||
trans.sa_session.add(uga)
|
||||
trans.sa_session.flush()
|
||||
item = dict(id=user_id,
|
||||
email=user.email,
|
||||
url=url_for('group_user', group_id=group_id, id=user_id))
|
||||
except Exception as e:
|
||||
item = f"Error in group_user API Adding user {user.email} to group {group.name}"
|
||||
log.error(item + ": %s", unicodify(e))
|
||||
return item
|
||||
return self.manager.update(trans, id, group_id)
|
||||
|
||||
@web.require_admin
|
||||
@web.legacy_expose_api
|
||||
def delete(self, trans, id, group_id, **kwd):
|
||||
@require_admin
|
||||
@expose_api
|
||||
def delete(self, trans: ProvidesAppContext, id: EncodedDatabaseIdField, group_id: EncodedDatabaseIdField, **kwd):
|
||||
"""
|
||||
DELETE /api/groups/{encoded_group_id}/users/{encoded_user_id}
|
||||
Removes a user from a group
|
||||
"""
|
||||
user_id = id
|
||||
decoded_group_id = trans.security.decode_id(group_id)
|
||||
decoded_user_id = trans.security.decode_id(user_id)
|
||||
try:
|
||||
group = trans.sa_session.query(trans.app.model.Group).get(decoded_group_id)
|
||||
user = trans.sa_session.query(trans.app.model.User).get(decoded_user_id)
|
||||
for uga in group.users:
|
||||
if uga.user == user:
|
||||
trans.sa_session.delete(uga)
|
||||
trans.sa_session.flush()
|
||||
item = dict(id=user_id,
|
||||
email=user.email,
|
||||
url=url_for('group_user', group_id=group_id, id=user_id))
|
||||
if not item:
|
||||
item = f"user {user.email} not in group {group.name}"
|
||||
except Exception as e:
|
||||
item = f"Error in group_user API Removing user {user.email} from group {group.name}"
|
||||
log.error(item + ": %s", unicodify(e))
|
||||
return item
|
||||
return self.manager.delete(trans, id, group_id)
|
||||
|
||||
@@ -0,0 +1,135 @@
|
||||
from typing import List
|
||||
|
||||
from galaxy_test.base.populators import DatasetPopulator
|
||||
from ._framework import ApiTestCase
|
||||
|
||||
|
||||
class GroupUsersApiTestCase(ApiTestCase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
|
||||
|
||||
def test_index(self, group_name: str = None):
|
||||
group_name = group_name or "test-group"
|
||||
group = self._create_group(group_name)
|
||||
encoded_group_id = group["id"]
|
||||
|
||||
group_users = self._get_group_users(encoded_group_id)
|
||||
|
||||
assert isinstance(group_users, list)
|
||||
assert len(group_users) > 0
|
||||
for group_user in group_users:
|
||||
self._assert_valid_group_user(group_user)
|
||||
|
||||
def test_index_only_admin(self):
|
||||
encoded_group_id = "any-group-id"
|
||||
response = self._get(f"groups/{encoded_group_id}/users")
|
||||
self._assert_status_code_is(response, 403)
|
||||
|
||||
def test_index_unknown_group_raises_400(self):
|
||||
encoded_group_id = "unknown-group-id"
|
||||
response = self._get(f"groups/{encoded_group_id}/users", admin=True)
|
||||
self._assert_status_code_is(response, 400)
|
||||
|
||||
def test_show(self):
|
||||
encoded_user_id = self.dataset_populator.user_id()
|
||||
group = self._create_group("test-group-show-user", encoded_user_ids=[encoded_user_id])
|
||||
encoded_group_id = group["id"]
|
||||
response = self._get(f"groups/{encoded_group_id}/users/{encoded_user_id}", admin=True)
|
||||
self._assert_status_code_is(response, 200)
|
||||
group_user = response.json()
|
||||
self._assert_valid_group_user(group_user)
|
||||
|
||||
def test_show_only_admin(self):
|
||||
encoded_group_id = "any-group-id"
|
||||
encoded_user_id = "any-user-id"
|
||||
response = self._get(f"groups/{encoded_group_id}/users/{encoded_user_id}")
|
||||
self._assert_status_code_is(response, 403)
|
||||
|
||||
def test_show_unknown_raises_400(self):
|
||||
group = self._create_group("test-group-with-unknown-user")
|
||||
encoded_group_id = group["id"]
|
||||
encoded_user_id = "unknown-user-id"
|
||||
response = self._get(f"groups/{encoded_group_id}/users/{encoded_user_id}", admin=True)
|
||||
self._assert_status_code_is(response, 400)
|
||||
|
||||
def test_update(self):
|
||||
group_name = "group-without-users"
|
||||
group = self._create_group(group_name, encoded_user_ids=[])
|
||||
encoded_group_id = group["id"]
|
||||
|
||||
group_users = self._get_group_users(encoded_group_id)
|
||||
assert len(group_users) == 0
|
||||
|
||||
encoded_user_id = self.dataset_populator.user_id()
|
||||
update_response = self._put(f"groups/{encoded_group_id}/users/{encoded_user_id}", admin=True)
|
||||
self._assert_status_code_is_ok(update_response)
|
||||
group_user = update_response.json()
|
||||
self._assert_valid_group_user(group_user, assert_id=encoded_user_id)
|
||||
assert group_user["url"] == f"/api/groups/{encoded_group_id}/users/{encoded_user_id}"
|
||||
|
||||
def test_update_only_admin(self):
|
||||
encoded_group_id = "any-group-id"
|
||||
encoded_user_id = "any-user-id"
|
||||
response = self._put(f"groups/{encoded_group_id}/users/{encoded_user_id}")
|
||||
self._assert_status_code_is(response, 403)
|
||||
|
||||
def test_delete(self):
|
||||
group_name = "group-with-user-to-delete"
|
||||
encoded_user_id = self.dataset_populator.user_id()
|
||||
group = self._create_group(group_name, encoded_user_ids=[encoded_user_id])
|
||||
encoded_group_id = group["id"]
|
||||
|
||||
group_users = self._get_group_users(encoded_group_id)
|
||||
assert len(group_users) == 1
|
||||
|
||||
delete_response = self._delete(f"groups/{encoded_group_id}/users/{encoded_user_id}", admin=True)
|
||||
self._assert_status_code_is_ok(delete_response)
|
||||
group_user = delete_response.json()
|
||||
self._assert_valid_group_user(group_user, assert_id=encoded_user_id)
|
||||
|
||||
group_users = self._get_group_users(encoded_group_id)
|
||||
assert len(group_users) == 0
|
||||
|
||||
def test_delete_only_admin(self):
|
||||
encoded_group_id = "any-group-id"
|
||||
encoded_user_id = "any-user-id"
|
||||
response = self._delete(f"groups/{encoded_group_id}/users/{encoded_user_id}")
|
||||
self._assert_status_code_is(response, 403)
|
||||
|
||||
def test_delete_unknown_raises_400(self):
|
||||
group_name = "group-without-user-to-delete"
|
||||
group = self._create_group(group_name, encoded_user_ids=[])
|
||||
encoded_group_id = group["id"]
|
||||
|
||||
group_users = self._get_group_users(encoded_group_id)
|
||||
assert len(group_users) == 0
|
||||
|
||||
encoded_user_id = "unknown-user-id"
|
||||
delete_response = self._delete(f"groups/{encoded_group_id}/users/{encoded_user_id}", admin=True)
|
||||
self._assert_status_code_is(delete_response, 400)
|
||||
|
||||
def _create_group(self, group_name: str, encoded_user_ids: List[str] = None):
|
||||
if encoded_user_ids is None:
|
||||
encoded_user_ids = [self.dataset_populator.user_id()]
|
||||
user_ids = encoded_user_ids
|
||||
payload = {
|
||||
"name": group_name,
|
||||
"user_ids": user_ids,
|
||||
}
|
||||
response = self._post("groups", payload, admin=True, json=True)
|
||||
self._assert_status_code_is(response, 200)
|
||||
group = response.json()[0] # POST /api/groups returns a list
|
||||
return group
|
||||
|
||||
def _get_group_users(self, encoded_group_id: str):
|
||||
response = self._get(f"groups/{encoded_group_id}/users", admin=True)
|
||||
self._assert_status_code_is(response, 200)
|
||||
group_users = response.json()
|
||||
return group_users
|
||||
|
||||
def _assert_valid_group_user(self, user, assert_id=None):
|
||||
self._assert_has_keys(user, "id", "email", "url")
|
||||
if assert_id is not None:
|
||||
assert user["id"] == assert_id
|
||||
Reference in New Issue
Block a user