rhodecode/controllers/api/api.py

import traceback
import logging

from rhodecode.controllers.api import JSONRPCController, JSONRPCError
from rhodecode.lib.auth import HasPermissionAllDecorator, \
    HasPermissionAnyDecorator
from rhodecode.model.scm import ScmModel

from rhodecode.model.db import User, UsersGroup, RepoGroup, Repository
from rhodecode.model.repo import RepoModel
from rhodecode.model.user import UserModel
from rhodecode.model.repo_permission import RepositoryPermissionModel
from rhodecode.model.users_group import UsersGroupModel
from rhodecode.model import users_group
from rhodecode.model.repos_group import ReposGroupModel
from sqlalchemy.orm.exc import NoResultFound

log = logging.getLogger(__name__)


class ApiController(JSONRPCController):
    """
    API Controller


    Each method needs to have USER as argument this is then based on given
    API_KEY propagated as instance of user object

    Preferably this should be first argument also


    Each function should also **raise** JSONRPCError for any
    errors that happens

    """

    @HasPermissionAllDecorator('hg.admin')
    def pull(self, apiuser, repo):
        """
        Dispatch pull action on given repo


        :param user:
        :param repo:
        """

        if Repository.is_valid(repo) is False:
            raise JSONRPCError('Unknown repo "%s"' % repo)

        try:
            ScmModel().pull_changes(repo, self.rhodecode_user.username)
            return 'Pulled from %s' % repo
        except Exception:
            raise JSONRPCError('Unable to pull changes from "%s"' % repo)

    @HasPermissionAllDecorator('hg.admin')
    def get_user(self, apiuser, username):
        """"
        Get a user by username

        :param apiuser
        :param username
        """

        user = User.get_by_username(username)
        if not user:
            return None

        return dict(id=user.user_id,
                    username=user.username,
                    firstname=user.name,
                    lastname=user.lastname,
                    email=user.email,
                    active=user.active,
                    admin=user.admin,
                    ldap=user.ldap_dn)

    @HasPermissionAllDecorator('hg.admin')
    def get_users(self, apiuser):
        """"
        Get all users

        :param apiuser
        """

        result = []
        for user in User.getAll():
            result.append(dict(id=user.user_id,
                                username=user.username,
                                firstname=user.name,
                                lastname=user.lastname,
                                email=user.email,
                                active=user.active,
                                admin=user.admin,
                                ldap=user.ldap_dn))
        return result

    @HasPermissionAllDecorator('hg.admin')
    def create_user(self, apiuser, username, password, firstname,
                    lastname, email, active=True, admin=False, ldap_dn=None):
        """
        Create new user

        :param apiuser:
        :param username:
        :param password:
        :param name:
        :param lastname:
        :param email:
        :param active:
        :param admin:
        :param ldap_dn:
        """

        if User.get_by_username(username):
            raise JSONRPCError("user %s already exist" % username)

        try:
            UserModel().create_or_update(username, password, email, firstname, 
                                         lastname, active, admin, ldap_dn)
            return dict(msg='created new user %s' % username)
        except Exception:
            log.error(traceback.format_exc())
            raise JSONRPCError('failed to create user %s' % username)

    @HasPermissionAllDecorator('hg.admin')
    def get_users_group(self, apiuser, group_name):
        """"
        Get users group by name

        :param apiuser
        :param group_name
        """

        users_group = UsersGroup.get_by_group_name(group_name)
        if not users_group:
            return None

        members = []
        for user in users_group.members:
            user = user.user
            members.append(dict(id=user.user_id,
                            username=user.username,
                            firstname=user.name,
                            lastname=user.lastname,
                            email=user.email,
                            active=user.active,
                            admin=user.admin,
                            ldap=user.ldap_dn))

        return dict(id=users_group.users_group_id,
                    name=users_group.users_group_name,
                    active=users_group.users_group_active,
                    members=members)

    @HasPermissionAllDecorator('hg.admin')
    def get_users_groups(self, apiuser):
        """"
        Get all users groups

        :param apiuser
        """

        result = []
        for users_group in UsersGroup.getAll():
            members = []
            for user in users_group.members:
                user = user.user
                members.append(dict(id=user.user_id,
                                username=user.username,
                                firstname=user.name,
                                lastname=user.lastname,
                                email=user.email,
                                active=user.active,
                                admin=user.admin,
                                ldap=user.ldap_dn))

            result.append(dict(id=users_group.users_group_id,
                                name=users_group.users_group_name,
                                active=users_group.users_group_active,
                                members=members))
        return result

    @HasPermissionAllDecorator('hg.admin')
    def create_users_group(self, apiuser, name, active=True):
        """
        Creates an new usergroup

        :param name:
        :param active:
        """

        if self.get_users_group(apiuser, name):
            raise JSONRPCError("users group %s already exist" % name)

        try:
            form_data = dict(users_group_name=name,
                             users_group_active=active)
            ug = UsersGroup.create(form_data)
            return dict(id=ug.users_group_id,
                        msg='created new users group %s' % name)
        except Exception:
            log.error(traceback.format_exc())
            raise JSONRPCError('failed to create group %s' % name)

    @HasPermissionAllDecorator('hg.admin')
    def add_user_to_users_group(self, apiuser, group_name, user_name):
        """"
        Add a user to a group

        :param apiuser
        :param group_name
        :param user_name
        """

        try:
            users_group = UsersGroup.get_by_group_name(group_name)
            if not users_group:
                raise JSONRPCError('unknown users group %s' % group_name)

            try:
                user = User.get_by_username(user_name)
            except NoResultFound:
                raise JSONRPCError('unknown user %s' % user_name)

            ugm = UsersGroupModel().add_user_to_group(users_group, user)

            return dict(id=ugm.users_group_member_id,
                        msg='created new users group member')
        except Exception:
            log.error(traceback.format_exc())
            raise JSONRPCError('failed to create users group member')

    @HasPermissionAnyDecorator('hg.admin')
    def get_repo(self, apiuser, repo_name):
        """"
        Get repository by name

        :param apiuser
        :param repo_name
        """

        try:
            repo = Repository.get_by_repo_name(repo_name)
        except NoResultFound:
            return None

        members = []
        for user in repo.repo_to_perm:
            perm = user.permission.permission_name
            user = user.user
            members.append(dict(type_="user",
                                    id=user.user_id,
                                    username=user.username,
                                    firstname=user.name,
                                    lastname=user.lastname,
                                    email=user.email,
                                    active=user.active,
                                    admin=user.admin,
                                    ldap=user.ldap_dn,
                                    permission=perm))
        for users_group in repo.users_group_to_perm:
            perm = users_group.permission.permission_name
            users_group = users_group.users_group
            members.append(dict(type_="users_group",
                                    id=users_group.users_group_id,
                                    name=users_group.users_group_name,
                                    active=users_group.users_group_active,
                                    permission=perm))

        return dict(id=repo.repo_id,
                    name=repo.repo_name,
                    type=repo.repo_type,
                    description=repo.description,
                    members=members)

    @HasPermissionAnyDecorator('hg.admin')
    def get_repos(self, apiuser):
        """"
        Get all repositories

        :param apiuser
        """

        result = []
        for repository in Repository.getAll():
            result.append(dict(id=repository.repo_id,
                                name=repository.repo_name,
                                type=repository.repo_type,
                                description=repository.description))
        return result

    @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')
    def create_repo(self, apiuser, name, owner_name, description='', 
                    repo_type='hg', private=False):
        """
        Create a repository

        :param apiuser
        :param name
        :param description
        :param type
        :param private
        :param owner_name
        """

        try:
            try:
                owner = User.get_by_username(owner_name)
            except NoResultFound:
                raise JSONRPCError('unknown user %s' % owner)

            if self.get_repo(apiuser, name):
                raise JSONRPCError("repo %s already exist" % name)

            groups = name.split('/')
            real_name = groups[-1]
            groups = groups[:-1]
            parent_id = None
            for g in groups:
                group = RepoGroup.get_by_group_name(g)
                if not group:
                    group = ReposGroupModel().create(dict(group_name=g,
                                                  group_description='',
                                                  group_parent_id=parent_id))
                parent_id = group.group_id

            RepoModel().create(dict(repo_name=real_name,
                                     repo_name_full=name,
                                     description=description,
                                     private=private,
                                     repo_type=repo_type,
                                     repo_group=parent_id,
                                     clone_uri=None), owner)
        except Exception:
            log.error(traceback.format_exc())
            raise JSONRPCError('failed to create repository %s' % name)

    @HasPermissionAnyDecorator('hg.admin')
    def add_user_to_repo(self, apiuser, repo_name, user_name, perm):
        """
        Add permission for a user to a repository

        :param apiuser
        :param repo_name
        :param user_name
        :param perm
        """

        try:
            try:
                repo = Repository.get_by_repo_name(repo_name)
            except NoResultFound:
                raise JSONRPCError('unknown repository %s' % repo)

            try:
                user = User.get_by_username(user_name)
            except NoResultFound:
                raise JSONRPCError('unknown user %s' % user)

            RepositoryPermissionModel()\
                .update_or_delete_user_permission(repo, user, perm)
        except Exception:
            log.error(traceback.format_exc())
            raise JSONRPCError('failed to edit permission %(repo)s for %(user)s'
                            % dict(user=user_name, repo=repo_name))