Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for tool_shed.webapp.api.users

import logging

import tool_shed.util.shed_util_common as suc
from galaxy import (
    exceptions,
    util,
    web,
)
from galaxy.model.base import transaction
from galaxy.security.validate_user_input import (
    validate_email,
    validate_password,
    validate_publicname,
)
from galaxy.webapps.base.controller import BaseAPIController

log = logging.getLogger(__name__)


[docs]class UsersController(BaseAPIController): """RESTful controller for interactions with users in the Tool Shed."""
[docs] @web.expose_api @web.require_admin def create(self, trans, payload, **kwd): """ POST /api/users Returns a dictionary of information about the created user. : param key: the current Galaxy admin user's API key The following parameters are included in the payload. :param email (required): the email address of the user :param password (required): the password of the user :param username (required): the public username of the user """ # Get the information about the user to be created from the payload. email = payload.get("email", "") password = payload.get("password", "") username = payload.get("username", "") message = self.__validate(trans, email=email, password=password, confirm=password, username=username) if message: raise exceptions.RequestParameterInvalidException(message) # Create the user. user = self.__create_user(trans, email, username, password) user_dict = user.to_dict(view="element", value_mapper=self.__get_value_mapper(trans)) user_dict["message"] = f"User '{str(user.username)}' has been created." user_dict["url"] = web.url_for(controller="users", action="show", id=trans.security.encode_id(user.id)) return user_dict
def __create_user(self, trans, email, username, password): user = trans.app.model.User(email=email) user.set_password_cleartext(password) user.username = username if trans.app.config.user_activation_on: user.active = False else: user.active = True # Activation is off, every new user is active by default. trans.sa_session.add(user) with transaction(trans.sa_session): trans.sa_session.commit() trans.app.security_agent.create_private_user_role(user) return user def __get_value_mapper(self, trans): value_mapper = {"id": trans.security.encode_id} return value_mapper
[docs] @web.expose_api_anonymous_and_sessionless def index(self, trans, deleted=False, **kwd): """ GET /api/users Returns a list of dictionaries that contain information about each user. """ # Example URL: http://localhost:9009/api/users user_dicts = [] deleted = util.asbool(deleted) for user in ( trans.sa_session.query(trans.app.model.User) .filter(trans.app.model.User.table.c.deleted == deleted) .order_by(trans.app.model.User.table.c.username) ): user_dict = user.to_dict(view="collection", value_mapper=self.__get_value_mapper(trans)) user_dict["url"] = web.url_for(controller="users", action="show", id=trans.security.encode_id(user.id)) user_dicts.append(user_dict) return user_dicts
[docs] @web.expose_api_anonymous_and_sessionless def show(self, trans, id, **kwd): """ GET /api/users/{encoded_user_id} GET /api/users/current Returns a dictionary of information about a user. :param id: the encoded id of the User object. """ user = None # user is requesting data about themselves user = trans.user if id == "current" else suc.get_user(trans.app, id) if user is None: user_dict = dict(message=f"Unable to locate user record for id {str(id)}.", status="error") return user_dict user_dict = user.to_dict(view="element", value_mapper=self.__get_value_mapper(trans)) user_dict["url"] = web.url_for(controller="users", action="show", id=trans.security.encode_id(user.id)) return user_dict
def __validate(self, trans, email, password, confirm, username): if username in ["repos"]: return f"The term '{username}' is a reserved word in the Tool Shed, so it cannot be used as a public user name." message = "\n".join( ( validate_email(trans, email), validate_password(trans, password, confirm), validate_publicname(trans, username), ) ).rstrip() return message