fromgalaxy.web.base.controllerimport*fromgalaxy.webapps.communityimportmodelfromgalaxy.model.ormimport*fromgalaxy.web.framework.helpersimporttime_ago,iff,gridsfromgalaxy.utilimportinflectorfromgalaxy.util.shed_utilimportget_configured_uifromcommonimport*fromrepositoryimportRepositoryListGrid,CategoryListGridfromgalaxyimporteggseggs.require('mercurial')frommercurialimporthgimportlogginglog=logging.getLogger(__name__)classUserListGrid(grids.Grid):# TODO: move this to an admin_common controller since it is virtually the same# in the galaxy webapp.classUserLoginColumn(grids.TextColumn):defget_value(self,trans,grid,user):returnuser.emailclassUserNameColumn(grids.TextColumn):defget_value(self,trans,grid,user):ifuser.username:returnuser.usernamereturn'not set'classGroupsColumn(grids.GridColumn):defget_value(self,trans,grid,user):ifuser.groups:returnlen(user.groups)return0classRolesColumn(grids.GridColumn):defget_value(self,trans,grid,user):ifuser.roles:returnlen(user.roles)return0classExternalColumn(grids.GridColumn):defget_value(self,trans,grid,user):ifuser.external:return'yes'return'no'classLastLoginColumn(grids.GridColumn):defget_value(self,trans,grid,user):ifuser.galaxy_sessions:returnself.format(user.galaxy_sessions[0].update_time)return'never'classStatusColumn(grids.GridColumn):defget_value(self,trans,grid,user):ifuser.purged:return"purged"elifuser.deleted:return"deleted"return""classEmailColumn(grids.GridColumn):deffilter(self,trans,user,query,column_filter):ifcolumn_filter=='All':returnqueryreturnquery.filter(and_(model.Tool.table.c.user_id==model.User.table.c.id,model.User.table.c.email==column_filter))# Grid definitionwebapp="community"title="Users"model_class=model.Usertemplate='/admin/user/grid.mako'default_sort_key="email"columns=[UserLoginColumn("Email",key="email",link=(lambdaitem:dict(operation="information",id=item.id,webapp="community")),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',webapp="community"))]operations=[grids.GridOperation("Manage Roles and Groups",condition=(lambdaitem:notitem.deleted),allow_multiple=False,url_args=dict(webapp="community",action="manage_roles_and_groups_for_user")),grids.GridOperation("Reset Password",condition=(lambdaitem:notitem.deleted),allow_multiple=True,allow_popup=False,url_args=dict(webapp="community",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'))]num_rows_per_page=50preserve_state=Falseuse_paging=Truedefget_current_item(self,trans,**kwargs):returntrans.userclassRoleListGrid(grids.Grid):# TODO: move this to an admin_common controller since it is virtually the same# in the galaxy webapp.classNameColumn(grids.TextColumn):defget_value(self,trans,grid,role):returnrole.nameclassDescriptionColumn(grids.TextColumn):defget_value(self,trans,grid,role):ifrole.description:returnrole.descriptionreturn''classTypeColumn(grids.TextColumn):defget_value(self,trans,grid,role):returnrole.typeclassStatusColumn(grids.GridColumn):defget_value(self,trans,grid,role):ifrole.deleted:return"deleted"return""classGroupsColumn(grids.GridColumn):defget_value(self,trans,grid,role):ifrole.groups:returnlen(role.groups)return0classUsersColumn(grids.GridColumn):defget_value(self,trans,grid,role):ifrole.users:returnlen(role.users)return0# Grid definitionwebapp="community"title="Roles"model_class=model.Roletemplate='/admin/dataset_security/role/grid.mako'default_sort_key="name"columns=[NameColumn("Name",key="name",link=(lambdaitem:dict(operation="Manage users and groups",id=item.id,webapp="community")),attach_popup=True,filterable="advanced"),DescriptionColumn("Description",key='description',attach_popup=False,filterable="advanced"),TypeColumn("Type",key='type',attach_popup=False,filterable="advanced"),GroupsColumn("Groups",attach_popup=False),UsersColumn("Users",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],columns[1],columns[2]],key="free-text-search",visible=False,filterable="standard"))global_actions=[grids.GridAction("Add new role",dict(controller='admin',action='roles',operation='create',webapp="community"))]operations=[grids.GridOperation("Rename",condition=(lambdaitem:notitem.deleted),allow_multiple=False,url_args=dict(webapp="community",action="rename_role")),grids.GridOperation("Delete",condition=(lambdaitem:notitem.deleted),allow_multiple=True,url_args=dict(webapp="community",action="mark_role_deleted")),grids.GridOperation("Undelete",condition=(lambdaitem:item.deleted),allow_multiple=True,url_args=dict(webapp="community",action="undelete_role")),grids.GridOperation("Purge",condition=(lambdaitem:item.deleted),allow_multiple=True,url_args=dict(webapp="community",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'))]num_rows_per_page=50preserve_state=Falseuse_paging=Truedefapply_query_filter(self,trans,query,**kwd):returnquery.filter(model.Role.type!=model.Role.types.PRIVATE)classGroupListGrid(grids.Grid):# TODO: move this to an admin_common controller since it is virtually the same# in the galaxy webapp.classNameColumn(grids.TextColumn):defget_value(self,trans,grid,group):returngroup.nameclassStatusColumn(grids.GridColumn):defget_value(self,trans,grid,group):ifgroup.deleted:return"deleted"return""classRolesColumn(grids.GridColumn):defget_value(self,trans,grid,group):ifgroup.roles:returnlen(group.roles)return0classUsersColumn(grids.GridColumn):defget_value(self,trans,grid,group):ifgroup.members:returnlen(group.members)return0# Grid definitionwebapp="community"title="Groups"model_class=model.Grouptemplate='/admin/dataset_security/group/grid.mako'default_sort_key="name"columns=[NameColumn("Name",#key="name",link=(lambdaitem:dict(operation="Manage users and roles",id=item.id,webapp="community")),attach_popup=True#filterable="advanced"),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],columns[1],columns[2]],key="free-text-search",visible=False,filterable="standard"))global_actions=[grids.GridAction("Add new group",dict(controller='admin',action='groups',operation='create',webapp="community"))]operations=[grids.GridOperation("Rename",condition=(lambdaitem:notitem.deleted),allow_multiple=False,url_args=dict(webapp="community",action="rename_group")),grids.GridOperation("Delete",condition=(lambdaitem:notitem.deleted),allow_multiple=True,url_args=dict(webapp="community",action="mark_group_deleted")),grids.GridOperation("Undelete",condition=(lambdaitem:item.deleted),allow_multiple=True,url_args=dict(webapp="community",action="undelete_group")),grids.GridOperation("Purge",condition=(lambdaitem:item.deleted),allow_multiple=True,url_args=dict(webapp="community",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'))]num_rows_per_page=50preserve_state=Falseuse_paging=TrueclassManageCategoryListGrid(CategoryListGrid):columns=[colforcolinCategoryListGrid.columns]# Override the NameColumn to include an Edit linkcolumns[0]=CategoryListGrid.NameColumn("Name",key="Category.name",link=(lambdaitem:dict(operation="Edit",id=item.id,webapp="community")),model_class=model.Category,attach_popup=False)global_actions=[grids.GridAction("Add new category",dict(controller='admin',action='manage_categories',operation='create',webapp="community"))]classAdminRepositoryListGrid(RepositoryListGrid):operations=[operationforoperationinRepositoryListGrid.operations]operations.append(grids.GridOperation("Delete",allow_multiple=False,condition=(lambdaitem:notitem.deleted),async_compatible=False))operations.append(grids.GridOperation("Undelete",allow_multiple=False,condition=(lambdaitem:item.deleted),async_compatible=False))standard_filters=[]classRepositoryMetadataListGrid(grids.Grid):classIdColumn(grids.IntegerColumn):defget_value(self,trans,grid,repository_metadata):returnrepository_metadata.idclassNameColumn(grids.TextColumn):defget_value(self,trans,grid,repository_metadata):returnrepository_metadata.repository.nameclassRevisionColumn(grids.TextColumn):defget_value(self,trans,grid,repository_metadata):repository=repository_metadata.repositoryrepo=hg.repository(get_configured_ui(),repository.repo_path)ctx=get_changectx_for_changeset(repo,repository_metadata.changeset_revision)return"%s:%s"%(str(ctx.rev()),repository_metadata.changeset_revision)classToolsColumn(grids.TextColumn):defget_value(self,trans,grid,repository_metadata):tools_str=''ifrepository_metadata:metadata=repository_metadata.metadataifmetadata:if'tools'inmetadata:fortool_metadata_dictinmetadata['tools']:tools_str+='%s <b>%s</b><br/>'%(tool_metadata_dict['id'],tool_metadata_dict['version'])returntools_strclassDatatypesColumn(grids.TextColumn):defget_value(self,trans,grid,repository_metadata):datatypes_str=''ifrepository_metadata:metadata=repository_metadata.metadataifmetadata:if'datatypes'inmetadata:fordatatype_metadata_dictinmetadata['datatypes']:datatypes_str+='%s<br/>'%datatype_metadata_dict['extension']returndatatypes_strclassWorkflowsColumn(grids.TextColumn):defget_value(self,trans,grid,repository_metadata):workflows_str=''ifrepository_metadata:metadata=repository_metadata.metadataifmetadata:if'workflows'inmetadata: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]forworkflow_tupinworkflow_tups]forworkflow_metadata_dictinworkflow_metadata_dicts:workflows_str+='%s<br/>'%workflow_metadata_dict['name']returnworkflows_strclassMaliciousColumn(grids.BooleanColumn):defget_value(self,trans,grid,repository_metadata):returnrepository_metadata.malicious# Grid definitiontitle="Repository Metadata"model_class=model.RepositoryMetadatatemplate='/webapps/community/repository/grid.mako'default_sort_key="name"columns=[IdColumn("Id",visible=False,attach_popup=False),NameColumn("Name",key="name",model_class=model.Repository,link=(lambdaitem:dict(operation="view_or_manage_repository_revision",id=item.id,webapp="community")),attach_popup=True),RevisionColumn("Revision",attach_popup=False),ToolsColumn("Tools",attach_popup=False),DatatypesColumn("Datatypes",attach_popup=False),WorkflowsColumn("Workflows",attach_popup=False),MaliciousColumn("Malicious",attach_popup=False)]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.")]standard_filters=[]default_filter={}num_rows_per_page=50preserve_state=Falseuse_paging=Truedefbuild_initial_query(self,trans,**kwd):returntrans.sa_session.query(self.model_class) \
.join(model.Repository.table)classAdminController(BaseUIController,Admin):user_list_grid=UserListGrid()role_list_grid=RoleListGrid()group_list_grid=GroupListGrid()manage_category_list_grid=ManageCategoryListGrid()repository_list_grid=AdminRepositoryListGrid()repository_metadata_list_grid=RepositoryMetadataListGrid()@web.expose@web.require_admindefbrowse_repository_metadata(self,trans,**kwd):if'operation'inkwd:operation=kwd['operation'].lower()ifoperation=="delete":returnself.delete_repository_metadata(trans,**kwd)ifoperation=="view_or_manage_repository_revision":# The received id is a RepositoryMetadata object id, so we need to get the# associated Repository and redirect to view_or_manage_repository with the# changeset_revision.repository_metadata=get_repository_metadata_by_id(trans,kwd['id'])repository=repository_metadata.repositorykwd['id']=trans.security.encode_id(repository.id)kwd['changeset_revision']=repository_metadata.changeset_revisionkwd['operation']='view_or_manage_repository'returntrans.response.send_redirect(web.url_for(controller='repository',action='browse_repositories',**kwd))returnself.repository_metadata_list_grid(trans,**kwd)@web.expose@web.require_admindefdelete_repository_metadata(self,trans,**kwd):params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')id=kwd.get('id',None)ifid:ids=util.listify(id)count=0forrepository_metadata_idinids:repository_metadata=get_repository_metadata_by_id(trans,repository_metadata_id)trans.sa_session.delete(repository_metadata)trans.sa_session.flush()count+=1ifcount:message="Deleted %d repository metadata %s"%(count,inflector.cond_plural(len(ids),"record"))else:message="No repository metadata ids received for deleting."status='error'trans.response.send_redirect(web.url_for(controller='admin',action='browse_repository_metadata',message=util.sanitize_text(message),status=status))@web.expose@web.require_admindefreset_all_repository_metadata(self,trans,**kwd):params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')if'reset_all_repository_metadata_button'inkwd:count=0forrepositoryintrans.sa_session.query(trans.model.Repository) \
.filter(trans.model.Repository.table.c.deleted==False):try:reset_all_repository_metadata(trans,trans.security.encode_id(repository.id))log.debug("Reset metadata on repository %s"%repository.name)count+=1exceptException,e:log.debug("Error attempting to reset metadata on repository %s: %s"%(repository.name,str(e)))message="Reset metadata on %d repositories"%counttrans.response.send_redirect(web.url_for(controller='admin',action='browse_repository_metadata',webapp='community',message=util.sanitize_text(message),status=status))returntrans.fill_template('/webapps/community/admin/reset_all_repository_metadata.mako',message=message,status=status)@web.expose@web.require_admindefbrowse_repositories(self,trans,**kwd):# We add params to the keyword dict in this method in order to rename the param# with an "f-" prefix, simulating filtering by clicking a search link. We have# to take this approach because the "-" character is illegal in HTTP requests.if'operation'inkwd:operation=kwd['operation'].lower()ifoperation=="view_or_manage_repository":returntrans.response.send_redirect(web.url_for(controller='repository',action='browse_repositories',**kwd))elifoperation=="edit_repository":returntrans.response.send_redirect(web.url_for(controller='repository',action='edit_repository',**kwd))elifoperation=="repositories_by_user":# Eliminate the current filters if any exist.fork,vinkwd.items():ifk.startswith('f-'):delkwd[k]if'user_id'inkwd:user=get_user(trans,kwd['user_id'])kwd['f-email']=user.emaildelkwd['user_id']else:# The received id is the repository id, so we need to get the id of the user# that uploaded the repository.repository_id=kwd.get('id',None)repository=get_repository(trans,repository_id)kwd['f-email']=repository.user.emailelifoperation=="repositories_by_category":# Eliminate the current filters if any exist.fork,vinkwd.items():ifk.startswith('f-'):delkwd[k]category_id=kwd.get('id',None)category=get_category(trans,category_id)kwd['f-Category.name']=category.nameelifoperation=="receive email alerts":ifkwd['id']:kwd['caller']='browse_repositories'returntrans.response.send_redirect(web.url_for(controller='repository',action='set_email_alerts',**kwd))else:delkwd['operation']elifoperation=='delete':returnself.delete_repository(trans,**kwd)elifoperation=="undelete":returnself.undelete_repository(trans,**kwd)# The changeset_revision_select_field in the RepositoryListGrid performs a refresh_on_change# which sends in request parameters like changeset_revison_1, changeset_revision_2, etc. One# of the many select fields on the grid performed the refresh_on_change, so we loop through # all of the received values to see which value is not the repository tip. If we find it, we# know the refresh_on_change occurred, and we have the necessary repository id and change set# revision to pass on.fork,vinkwd.items():changset_revision_str='changeset_revision_'ifk.startswith(changset_revision_str):repository_id=trans.security.encode_id(int(k.lstrip(changset_revision_str)))repository=get_repository(trans,repository_id)ifrepository.tip!=v:returntrans.response.send_redirect(web.url_for(controller='repository',action='browse_repositories',operation='view_or_manage_repository',id=trans.security.encode_id(repository.id),changeset_revision=v))# Render the list viewreturnself.repository_list_grid(trans,**kwd)@web.expose@web.require_admindefregenerate_statistics(self,trans,**kwd):params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')if'regenerate_statistics_button'inkwd:trans.app.shed_counter.generate_statistics()message="Successfully regenerated statistics"returntrans.fill_template('/webapps/community/admin/statistics.mako',message=message,status=status)@web.expose@web.require_admindefdelete_repository(self,trans,**kwd):params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')id=kwd.get('id',None)ifid:# Deleting multiple items is currently not allowed (allow_multiple=False), so there will only be 1 id. ids=util.listify(id)count=0deleted_repositories=""forrepository_idinids:repository=get_repository(trans,repository_id)ifnotrepository.deleted:repository.deleted=Truetrans.sa_session.add(repository)trans.sa_session.flush()count+=1deleted_repositories+=" %s "%repository.nameifcount:message="Deleted %d%s: %s"%(count,inflector.cond_plural(len(ids),"repository"),deleted_repositories)else:message="All selected repositories were already marked deleted."else:message="No repository ids received for deleting."status='error'trans.response.send_redirect(web.url_for(controller='admin',action='browse_repositories',message=util.sanitize_text(message),status=status))@web.expose@web.require_admindefundelete_repository(self,trans,**kwd):params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')id=kwd.get('id',None)ifid:# Undeleting multiple items is currently not allowed (allow_multiple=False), so there will only be 1 id.ids=util.listify(id)count=0undeleted_repositories=""forrepository_idinids:repository=get_repository(trans,repository_id)ifrepository.deleted:repository.deleted=Falsetrans.sa_session.add(repository)trans.sa_session.flush()count+=1undeleted_repositories+=" %s"%repository.nameifcount:message="Undeleted %d%s: %s"%(count,inflector.cond_plural(count,"repository"),undeleted_repositories)else:message="No selected repositories were marked deleted, so they could not be undeleted."else:message="No repository ids received for undeleting."status='error'trans.response.send_redirect(web.url_for(controller='admin',action='browse_repositories',message=util.sanitize_text(message),status='done'))@web.expose@web.require_admindefmanage_categories(self,trans,**kwd):if'f-free-text-search'inkwd:# Trick to enable searching repository name, description from the CategoryListGrid.# What we've done is rendered the search box for the RepositoryListGrid on the grid.mako# template for the CategoryListGrid. See ~/templates/webapps/community/category/grid.mako.# Since we are searching repositories and not categories, redirect to browse_repositories().returnself.browse_repositories(trans,**kwd)if'operation'inkwd:operation=kwd['operation'].lower()ifoperation=="create":returnself.create_category(trans,**kwd)elifoperation=="delete":returnself.mark_category_deleted(trans,**kwd)elifoperation=="undelete":returnself.undelete_category(trans,**kwd)elifoperation=="purge":returnself.purge_category(trans,**kwd)elifoperation=="edit":returnself.edit_category(trans,**kwd)# Render the list viewreturnself.manage_category_list_grid(trans,**kwd)@web.expose@web.require_admindefcreate_category(self,trans,**kwd):params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')ifparams.get('create_category_button',False):name=util.restore_text(params.name)description=util.restore_text(params.description)error=Falseifnotnameornotdescription:message='Enter a valid name and a description'error=Trueeliftrans.sa_session.query(trans.app.model.Category) \
.filter(trans.app.model.Category.table.c.name==name) \
.first():message='A category with that name already exists'error=Trueiferror:returntrans.fill_template('/webapps/community/category/create_category.mako',name=name,description=description,message=message,status='error')else:# Create the categorycategory=trans.app.model.Category(name=name,description=description)trans.sa_session.add(category)message="Category '%s' has been created"%category.nametrans.sa_session.flush()trans.response.send_redirect(web.url_for(controller='admin',action='manage_categories',message=util.sanitize_text(message),status='done'))trans.response.send_redirect(web.url_for(controller='admin',action='create_category',message=util.sanitize_text(message),status='error'))else:name=''description=''returntrans.fill_template('/webapps/community/category/create_category.mako',name=name,description=description,message=message,status=status)@web.expose@web.require_admindefedit_category(self,trans,**kwd):params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')id=params.get('id',None)ifnotid:message="No category ids received for editing"trans.response.send_redirect(web.url_for(controller='admin',action='manage_categories',message=message,status='error'))category=get_category(trans,id)ifparams.get('edit_category_button',False):new_name=util.restore_text(params.get('name','')).strip()new_description=util.restore_text(params.get('description','')).strip()ifcategory.name!=new_nameorcategory.description!=new_description:ifnotnew_name:message='Enter a valid name'status='error'elifcategory.name!=new_nameand \
trans.sa_session.query(trans.app.model.Category).filter(trans.app.model.Category.table.c.name==new_name).first():message='A category with that name already exists'status='error'else:category.name=new_namecategory.description=new_descriptiontrans.sa_session.add(category)trans.sa_session.flush()message="The information has been saved for category '%s'"%(category.name)returntrans.response.send_redirect(web.url_for(controller='admin',action='manage_categories',message=util.sanitize_text(message),status='done'))returntrans.fill_template('/webapps/community/category/edit_category.mako',category=category,message=message,status=status)@web.expose@web.require_admindefmark_category_deleted(self,trans,**kwd):# TODO: We should probably eliminate the Category.deleted column since it really makes no# sense to mark a category as deleted (category names and descriptions can be changed instead).# If we do this, and the following 2 methods can be eliminated.params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')id=kwd.get('id',None)ifid:ids=util.listify(id)message="Deleted %d categories: "%len(ids)forcategory_idinids:category=get_category(trans,category_id)category.deleted=Truetrans.sa_session.add(category)trans.sa_session.flush()message+=" %s "%category.nameelse:message="No category ids received for deleting."status='error'trans.response.send_redirect(web.url_for(controller='admin',action='manage_categories',message=util.sanitize_text(message),status='done'))@web.expose@web.require_admindefundelete_category(self,trans,**kwd):params=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')id=kwd.get('id',None)ifid:ids=util.listify(id)count=0undeleted_categories=""forcategory_idinids:category=get_category(trans,category_id)ifcategory.deleted:category.deleted=Falsetrans.sa_session.add(category)trans.sa_session.flush()count+=1undeleted_categories+=" %s"%category.namemessage="Undeleted %d categories: %s"%(count,undeleted_categories)else:message="No category ids received for undeleting."status='error'trans.response.send_redirect(web.url_for(controller='admin',action='manage_categories',message=util.sanitize_text(message),status='done'))@web.expose@web.require_admindefpurge_category(self,trans,**kwd):# This method should only be called for a Category that has previously been deleted.# Purging a deleted Category deletes all of the following from the database:# - RepoitoryCategoryAssociations where category_id == Category.idparams=util.Params(kwd)message=util.restore_text(params.get('message',''))status=params.get('status','done')id=kwd.get('id',None)ifid:ids=util.listify(id)count=0purged_categories=""message="Purged %d categories: "%len(ids)forcategory_idinids:category=get_category(trans,category_id)ifcategory.deleted:# Delete RepositoryCategoryAssociationsforrcaincategory.repositories:trans.sa_session.delete(rca)trans.sa_session.flush()purged_categories+=" %s "%category.namemessage="Purged %d categories: %s"%(count,purged_categories)else:message="No category ids received for purging."status='error'trans.response.send_redirect(web.url_for(controller='admin',action='manage_categories',message=util.sanitize_text(message),status='done'))