Source code for tool_shed.grids.admin_grids

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 = list(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 = list(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)


[docs]class RepositoryMetadataGrid(grids.Grid):
[docs]    class IdColumn(grids.IntegerColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            return repository_metadata.id

[docs]    class NameColumn(grids.TextColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            return escape(repository_metadata.repository.name)

[docs]    class OwnerColumn(grids.TextColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            return escape(repository_metadata.repository.user.username)

[docs]    class RevisionColumn(grids.TextColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            repository = repository_metadata.repository
            return hg_util.get_revision_label(
                trans.app, repository, repository_metadata.changeset_revision, include_date=True, include_hash=True
            )

[docs]    class ToolsColumn(grids.TextColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            tools_str = "0"
            if repository_metadata:
                metadata = repository_metadata.metadata
                if metadata:
                    if "tools" in metadata:
                        # We used to display the following, but grid was too cluttered.
                        # for tool_metadata_dict in metadata[ 'tools' ]:
                        #    tools_str += '%s <b>%s</b><br/>' % ( tool_metadata_dict[ 'id' ], tool_metadata_dict[ 'version' ] )
                        return "{}".format(len(metadata["tools"]))
            return tools_str

[docs]    class DatatypesColumn(grids.TextColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            datatypes_str = "0"
            if repository_metadata:
                metadata = repository_metadata.metadata
                if metadata:
                    if "datatypes" in metadata:
                        # We used to display the following, but grid was too cluttered.
                        # for datatype_metadata_dict in metadata[ 'datatypes' ]:
                        #    datatypes_str += '%s<br/>' % datatype_metadata_dict[ 'extension' ]
                        return "{}".format(len(metadata["datatypes"]))
            return datatypes_str

[docs]    class WorkflowsColumn(grids.TextColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            workflows_str = "0"
            if repository_metadata:
                metadata = repository_metadata.metadata
                if metadata:
                    if "workflows" in metadata:
                        # We used to display the following, but grid was too cluttered.
                        # workflows_str += '<b>Workflows:</b><br/>'
                        # metadata[ 'workflows' ] is a list of tuples where each contained tuple is
                        # [ <relative path to the .ga file in the repository>, <exported workflow dict> ]
                        # workflow_tups = metadata[ 'workflows' ]
                        # workflow_metadata_dicts = [ workflow_tup[1] for workflow_tup in workflow_tups ]
                        # for workflow_metadata_dict in workflow_metadata_dicts:
                        #    workflows_str += '%s<br/>' % workflow_metadata_dict[ 'name' ]
                        return "{}".format(len(metadata["workflows"]))
            return workflows_str

[docs]    class DeletedColumn(grids.BooleanColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            if repository_metadata.repository.deleted:
                return "yes"
            return ""

[docs]    class DeprecatedColumn(grids.BooleanColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            if repository_metadata.repository.deprecated:
                return "yes"
            return ""

[docs]    class MaliciousColumn(grids.BooleanColumn):
[docs]        def get_value(self, trans, grid, repository_metadata):
            if repository_metadata.malicious:
                return "yes"
            return ""

    # Grid definition
    title = "Repository Metadata"
    model_class = model.RepositoryMetadata
    default_sort_key = "name"
    use_hide_message = False
    columns = [
        IdColumn("Id", visible=False, attach_popup=False),
        NameColumn(
            "Name",
            key="name",
            model_class=model.Repository,
            link=(lambda item: dict(operation="view_or_manage_repository_revision", id=item.id)),
            attach_popup=True,
        ),
        OwnerColumn("Owner", attach_popup=False),
        RevisionColumn("Revision", attach_popup=False),
        ToolsColumn("Tools", attach_popup=False),
        DatatypesColumn("Datatypes", attach_popup=False),
        WorkflowsColumn("Workflows", attach_popup=False),
        DeletedColumn("Deleted", attach_popup=False),
        DeprecatedColumn("Deprecated", attach_popup=False),
        MaliciousColumn("Malicious", attach_popup=False),
    ]
    columns.append(
        grids.MulticolFilterColumn(
            "Search repository name",
            cols_to_filter=[columns[1]],
            key="free-text-search",
            visible=False,
            filterable="standard",
        )
    )
    operations = [
        grids.GridOperation(
            "Delete",
            allow_multiple=False,
            allow_popup=True,
            async_compatible=False,
            confirm="Repository metadata records cannot be recovered after they are deleted. Click OK to delete the selected items.",
        )
    ]

[docs]    def build_initial_query(self, trans, **kwd):
        return trans.sa_session.query(model.RepositoryMetadata).join(model.Repository.table)