import logging
from markupsafe import escape
from sqlalchemy import and_
from galaxy.web.framework.helpers import time_ago
from galaxy.web.legacy_framework import grids
from tool_shed.grids.repository_grids import (
CategoryGrid,
RepositoryGrid,
)
from tool_shed.util import hg_util
from tool_shed.webapp import model
log = logging.getLogger(__name__)
[docs]class UserGrid(grids.Grid):
[docs] class UserLoginColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, user):
return escape(user.email)
[docs] class UserNameColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, user):
if user.username:
return escape(user.username)
return "not set"
[docs] class GroupsColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user):
if user.groups:
return len(user.groups)
return 0
[docs] class RolesColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user):
if user.roles:
return len(user.roles)
return 0
[docs] class ExternalColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user):
if user.external:
return "yes"
return "no"
[docs] class LastLoginColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user):
if user.current_galaxy_session:
return self.format(user.current_galaxy_session.update_time)
return "never"
[docs] class StatusColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, user):
if user.purged:
return "purged"
elif user.deleted:
return "deleted"
return ""
[docs] class EmailColumn(grids.GridColumn):
[docs] def filter(self, trans, user, query, column_filter):
if column_filter == "All":
return query
return query.filter(
and_(model.Tool.table.c.user_id == model.User.table.c.id, model.User.table.c.email == column_filter)
)
title = "Users"
model_class = model.User
default_sort_key = "email"
columns = [
UserLoginColumn(
"Email",
key="email",
link=(lambda item: dict(operation="information", id=item.id)),
attach_popup=True,
filterable="advanced",
),
UserNameColumn("User Name", key="username", attach_popup=False, filterable="advanced"),
GroupsColumn("Groups", attach_popup=False),
RolesColumn("Roles", attach_popup=False),
ExternalColumn("External", attach_popup=False),
LastLoginColumn("Last Login", format=time_ago),
StatusColumn("Status", attach_popup=False),
# Columns that are valid for filtering but are not visible.
EmailColumn("Email", key="email", visible=False),
]
columns.append(
grids.MulticolFilterColumn(
"Search",
cols_to_filter=[columns[0], columns[1]],
key="free-text-search",
visible=False,
filterable="standard",
)
)
global_actions = [grids.GridAction("Create new user", dict(controller="admin", action="users", operation="create"))]
operations = [
grids.GridOperation(
"Manage Roles and Groups",
condition=(lambda item: not item.deleted),
allow_multiple=False,
url_args=dict(action="manage_roles_and_groups_for_user"),
),
grids.GridOperation(
"Reset Password",
condition=(lambda item: not item.deleted),
allow_multiple=True,
allow_popup=False,
url_args=dict(action="reset_user_password"),
),
]
standard_filters = [
grids.GridColumnFilter("Active", args=dict(deleted=False)),
grids.GridColumnFilter("Deleted", args=dict(deleted=True, purged=False)),
grids.GridColumnFilter("Purged", args=dict(purged=True)),
grids.GridColumnFilter("All", args=dict(deleted="All")),
]
use_paging = False
[docs] def get_current_item(self, trans, **kwargs):
return trans.user
[docs]class RoleGrid(grids.Grid):
[docs] class NameColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, role):
return escape(str(role.name))
[docs] class DescriptionColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, role):
if role.description:
return str(role.description)
return ""
[docs] class TypeColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, role):
return str(role.type)
[docs] class StatusColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, role):
if role.deleted:
return "deleted"
return ""
[docs] class GroupsColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, role):
if role.groups:
return len(role.groups)
return 0
[docs] class RepositoriesColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, role):
if role.repositories:
return len(role.repositories)
return 0
[docs] class UsersColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, role):
if role.users:
return len(role.users)
return 0
title = "Roles"
model_class = model.Role
default_sort_key = "name"
columns = [
NameColumn(
"Name",
key="name",
link=(lambda item: dict(operation="Manage role associations", id=item.id)),
attach_popup=True,
filterable="advanced",
),
DescriptionColumn("Description", key="description", attach_popup=False, filterable="advanced"),
GroupsColumn("Groups", attach_popup=False),
RepositoriesColumn("Repositories", attach_popup=False),
UsersColumn("Users", attach_popup=False),
# Columns that are valid for filtering but are not visible.
grids.DeletedColumn("Deleted", key="deleted", visible=False, filterable="advanced"),
]
columns.append(
grids.MulticolFilterColumn(
"Search", cols_to_filter=[columns[0]], key="free-text-search", visible=False, filterable="standard"
)
)
global_actions = [grids.GridAction("Add new role", dict(controller="admin", action="roles", operation="create"))]
# Repository admin roles currently do not have any operations since they are managed automatically based
# on other events. For example, if a repository is renamed, its associated admin role is automatically
# renamed accordingly and if a repository is deleted its associated admin role is automatically deleted.
operations = [
grids.GridOperation(
"Rename",
condition=(lambda item: not item.deleted and not item.is_repository_admin_role),
allow_multiple=False,
url_args=dict(action="rename_role"),
),
grids.GridOperation(
"Delete",
condition=(lambda item: not item.deleted and not item.is_repository_admin_role),
allow_multiple=True,
url_args=dict(action="mark_role_deleted"),
),
grids.GridOperation(
"Undelete",
condition=(lambda item: item.deleted and not item.is_repository_admin_role),
allow_multiple=True,
url_args=dict(action="undelete_role"),
),
grids.GridOperation(
"Purge",
condition=(lambda item: item.deleted and not item.is_repository_admin_role),
allow_multiple=True,
url_args=dict(action="purge_role"),
),
]
standard_filters = [
grids.GridColumnFilter("Active", args=dict(deleted=False)),
grids.GridColumnFilter("Deleted", args=dict(deleted=True)),
grids.GridColumnFilter("All", args=dict(deleted="All")),
]
use_paging = False
[docs] def apply_query_filter(self, trans, query, **kwd):
return query.filter(model.Role.type != model.Role.types.PRIVATE)
[docs]class GroupGrid(grids.Grid):
[docs] class NameColumn(grids.TextColumn):
[docs] def get_value(self, trans, grid, group):
return str(group.name)
[docs] class StatusColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, group):
if group.deleted:
return "deleted"
return ""
[docs] class RolesColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, group):
if group.roles:
return len(group.roles)
return 0
[docs] class UsersColumn(grids.GridColumn):
[docs] def get_value(self, trans, grid, group):
if group.users:
return len(group.users)
return 0
title = "Groups"
model_class = model.Group
default_sort_key = "name"
columns = [
NameColumn(
"Name",
key="name",
link=(lambda item: dict(operation="Manage users and roles", id=item.id)),
attach_popup=True,
),
UsersColumn("Users", attach_popup=False),
RolesColumn("Roles", attach_popup=False),
StatusColumn("Status", attach_popup=False),
# Columns that are valid for filtering but are not visible.
grids.DeletedColumn("Deleted", key="deleted", visible=False, filterable="advanced"),
]
columns.append(
grids.MulticolFilterColumn(
"Search", cols_to_filter=[columns[0]], key="free-text-search", visible=False, filterable="standard"
)
)
global_actions = [grids.GridAction("Add new group", dict(controller="admin", action="groups", operation="create"))]
operations = [
grids.GridOperation(
"Rename",
condition=(lambda item: not item.deleted),
allow_multiple=False,
url_args=dict(action="rename_group"),
),
grids.GridOperation(
"Delete",
condition=(lambda item: not item.deleted),
allow_multiple=True,
url_args=dict(action="mark_group_deleted"),
),
grids.GridOperation(
"Undelete",
condition=(lambda item: item.deleted),
allow_multiple=True,
url_args=dict(action="undelete_group"),
),
grids.GridOperation(
"Purge", condition=(lambda item: item.deleted), allow_multiple=True, url_args=dict(action="purge_group")
),
]
standard_filters = [
grids.GridColumnFilter("Active", args=dict(deleted=False)),
grids.GridColumnFilter("Deleted", args=dict(deleted=True)),
grids.GridColumnFilter("All", args=dict(deleted="All")),
]
use_paging = False
[docs]class ManageCategoryGrid(CategoryGrid):
columns = [col for col in CategoryGrid.columns]
# Override the NameColumn to include an Edit link
columns[0] = CategoryGrid.NameColumn(
"Name",
key="Category.name",
link=(lambda item: dict(operation="Edit", id=item.id)),
model_class=model.Category,
attach_popup=False,
)
global_actions = [
grids.GridAction("Add new category", dict(controller="admin", action="manage_categories", operation="create"))
]
[docs]class AdminRepositoryGrid(RepositoryGrid):
[docs] class DeletedColumn(grids.BooleanColumn):
[docs] def get_value(self, trans, grid, repository):
if repository.deleted:
return "yes"
return ""
columns = [
RepositoryGrid.NameColumn(
"Name",
key="name",
link=(lambda item: dict(operation="view_or_manage_repository", id=item.id)),
attach_popup=True,
),
RepositoryGrid.HeadsColumn("Heads"),
RepositoryGrid.UserColumn(
"Owner",
model_class=model.User,
link=(lambda item: dict(operation="repositories_by_user", id=item.id)),
attach_popup=False,
key="User.username",
),
RepositoryGrid.DeprecatedColumn("Deprecated", key="deprecated", attach_popup=False),
# Columns that are valid for filtering but are not visible.
DeletedColumn("Deleted", key="deleted", attach_popup=False),
]
columns.append(
grids.MulticolFilterColumn(
"Search repository name",
cols_to_filter=[columns[0]],
key="free-text-search",
visible=False,
filterable="standard",
)
)
operations = [operation for operation in RepositoryGrid.operations]
operations.append(
grids.GridOperation(
"Delete", allow_multiple=False, condition=(lambda item: not item.deleted), async_compatible=False
)
)
operations.append(
grids.GridOperation(
"Undelete", allow_multiple=False, condition=(lambda item: item.deleted), async_compatible=False
)
)
[docs] def build_initial_query(self, trans, **kwd):
return trans.sa_session.query(model.Repository).join(model.User.table)