id : <id_given_in_input>

    result : {

                 "ip_addr_server": <ip_from_client>",

                 "user_ips": [

                                {

                                   "ip_addr": "<ip_with_mask>",

                                   "ip_range": ["<start_ip>", "<end_ip>"],

                                },

                                ...

                             ]

             }

    error :  null

get_user

^^^^^^^^

Get a user by username or userid. The result is empty if user can't be found.

If userid param is skipped, it is set to id of user who is calling this method.

Any userid can be specified when the command is executed using the api_key of a user with admin rights.

Regular users can only specify their own userid.

INPUT::

    id : <id_for_response>

    api_key : "<api_key>"

    method :  "fork_repo"

    args:     {

                "repoid" :          "<reponame or repo_id>",

                "fork_name":        "<forkname>",

                "owner":            "<username or user_id = Optional(=apiuser)>",

                "description":      "<description>",

                "copy_permissions": "<bool>",

                "private":          "<bool>",

                "landing_rev":      "<landing_rev>"

              }

OUTPUT::

    id : <id_given_in_input>

    result: {

              "msg": "Created fork of `<reponame>` as `<forkname>`",

              "success": true

            }

    error:  null

delete_repo

            "msg": "Pulled from `<repository name>`"

            "repository": "<repository name>"

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "Unable to pull changes from `<reponame>`"

          }

        """

        repo = get_repo_or_error(repoid)

        try:

            ScmModel().pull_changes(repo.repo_name,

                                    request.authuser.username,

                                    request.ip_addr,

                                    clone_uri=clone_uri)

            return dict(

                msg='Pulled from `%s`' % repo.repo_name,

                repository=repo.repo_name

            )

        except Exception:

            'added': [<added repository name>,...]

            'removed': [<removed repository name>,...]

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            'Error occurred during rescan repositories action'

          }

        """

        try:

            rm_obsolete = remove_obsolete

            added, removed = repo2db_mapper(ScmModel().repo_scan(),

                                            remove_obsolete=rm_obsolete)

            return {'added': added, 'removed': removed}

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'Error occurred during rescan repositories action'

            )

    def invalidate_cache(self, repoid):

            'msg': Cache for repository `<repository name>` was invalidated,

            'repository': <repository name>

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            'Error occurred during cache invalidation action'

          }

        """

        repo = get_repo_or_error(repoid)

        if not HasPermissionAny('hg.admin')():

            if not HasRepoPermissionLevel('write')(repo.repo_name):

                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

        try:

            ScmModel().mark_for_invalidation(repo.repo_name)

            return dict(

                msg='Cache for repository `%s` was invalidated' % (repoid,),

                repository=repo.repo_name

            )

            server_ip_addr=request.ip_addr,

            user_ips=ips

        )

    # alias for old

    show_ip = get_ip

    @HasPermissionAnyDecorator('hg.admin')

    def get_server_info(self):

        """

        return server info, including Kallithea version and installed packages

        OUTPUT::

          id : <id_given_in_input>

          result : {

            'modules': [ [<module name>, <module version>], ...]

            'py_version': <python version>,

            'platform': <platform type>,

            'kallithea_version': <kallithea version>,

            'git_version': '<git version>',

            'git_path': '<git path>'

          }

          error :  null

                        "emails":       "[<list of all emails including additional ones>]",

                        "active" :      "<bool: user active>",

                        "admin" :       "<bool: user is admin>",

                        "permissions": {

                            "global": ["hg.create.repository",

                                       "repository.read",

                                       "hg.register.manual_activate"],

                            "repositories": {"repo1": "repository.none"},

                            "repositories_groups": {"Group1": "group.read"},

                            "user_groups": { "usrgrp1": "usergroup.admin" }

                         },

                    }

            error:  null

        """

        if not HasPermissionAny('hg.admin')():

            # make sure normal user does not pass someone else userid,

            # he is not allowed to do that

            if userid is not None and userid != request.authuser.user_id:

                raise JSONRPCError(

                    'userid is not the same as your user'

                )

        if userid is None:

            userid = request.authuser.user_id

        user = get_user_or_error(userid)

        data = user.get_api_data()

        data['permissions'] = AuthUser(user_id=user.user_id).permissions

        return data

    @HasPermissionAnyDecorator('hg.admin')

    def get_users(self):

        """

        Lists all existing users. This command can be executed only using api_key

        belonging to user with admin rights.

        OUTPUT::

            id : <id_given_in_input>

            result: [<user_object>, ...]

            error:  null

        """

        return [

            user.get_api_data()

            for user in db.User.query()

                .order_by(db.User.username)

                .filter_by(is_default_user=False)

        ]

    @HasPermissionAnyDecorator('hg.admin')

    def create_user(self, username, email, password='',

                    firstname='', lastname='',

                    active=True, admin=False,

                    extern_type=db.User.DEFAULT_AUTH_TYPE,

        :type firstname: str

        :param lastname: lastname

        :type lastname: str

        :param active: active

        :type active: Optional(bool)

        :param admin: admin

        :type admin: Optional(bool)

        :param extern_name: name of extern

        :type extern_name: Optional(str)

        :param extern_type: extern_type

        :type extern_type: Optional(str)

        OUTPUT::

            id : <id_given_in_input>

            result: {

                      "msg" : "created new user `<username>`",

                      "user": <user_obj>

                    }

            error:  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "user `<username>` already exist"

            or

            "email `<email>` already exist"

            or

            "failed to create user `<username>`"

          }

        """

        if db.User.get_by_username(username):

            raise JSONRPCError("user `%s` already exist" % (username,))

        if db.User.get_by_email(email):

            raise JSONRPCError("email `%s` already exist" % (email,))

        try:

            user = UserModel().create_or_update(

                username=username,

                password=password,

                email=email,

                firstname=firstname,

        :type firstname: Optional(str)

        :param lastname: lastname

        :type lastname: Optional(str)

        :param active: active

        :type active: Optional(bool)

        :param admin: admin

        :type admin: Optional(bool)

        :param extern_name:

        :type extern_name: Optional(str)

        :param extern_type:

        :type extern_type: Optional(str)

        OUTPUT::

            id : <id_given_in_input>

            result: {

                      "msg" : "updated user ID:<userid> <username>",

                      "user": <user_object>,

                    }

            error:  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "failed to update user `<username>`"

          }

        """

        user = get_user_or_error(userid)

        # only non optional arguments will be stored in updates

        updates = {}

        try:

            store_update(updates, username, 'username')

            store_update(updates, password, 'password')

            store_update(updates, email, 'email')

            store_update(updates, firstname, 'name')

            store_update(updates, lastname, 'lastname')

                      "msg" : "deleted user ID:<userid> <username>",

                      "user": null

                    }

            error:  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "failed to delete user ID:<userid> <username>"

          }

        """

        user = get_user_or_error(userid)

        try:

            UserModel().delete(userid)

            meta.Session().commit()

            return dict(

                msg='deleted user ID:%s %s' % (user.user_id, user.username),

                user=None

            )

        except Exception:

            id : <id_given_in_input>

            result : None if group not exist

                     {

                       "users_group_id" : "<id>",

                       "group_name" :     "<groupname>",

                       "group_description": "<description>"

                       "active":          "<bool>",

                       "owner":           "<username>"

                       "members" :  [<user_obj>,...]

                     }

            error : null

        """

        user_group = get_user_group_or_error(usergroupid)

        if not HasPermissionAny('hg.admin')():

            if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):

                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

        data = user_group.get_api_data()

        return data

    # permission check inside

    def get_user_groups(self):

        """

        Lists all existing user groups. This command can be executed only using

        api_key belonging to user with admin rights or user who has at least

        read access to user group.

        OUTPUT::

            id : <id_given_in_input>

            result : [<user_group_obj>,...]

            error : null

        """

        return [

            user_group.get_api_data()

            for user_group in UserGroupList(db.UserGroup.query().all(), perm_level='read')

        ]

    @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')

    def create_user_group(self, group_name, description='',

                          owner=None, active=True):

        """

        Creates new user group. This command can be executed only using api_key

        belonging to user with admin rights or an user who has create user group

        permission

                    }

            error:  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "user group `<group name>` already exist"

            or

            "failed to create group `<group name>`"

          }

        """

        if UserGroupModel().get_by_name(group_name):

            raise JSONRPCError("user group `%s` already exist" % (group_name,))

        try:

            if owner is None:

                owner = request.authuser.user_id

            owner = get_user_or_error(owner)

            ug = UserGroupModel().create(name=group_name, description=description,

                                         owner=owner, active=active)

            meta.Session().commit()

            return dict(

            "msg": 'updated user group ID:<user group id> <user group name>',

            "user_group": <user_group_object>

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "failed to update user group `<user group name>`"

          }

        """

        user_group = get_user_group_or_error(usergroupid)

        if not HasPermissionAny('hg.admin')():

            if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):

                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

        if owner is not None:

            owner = get_user_or_error(owner)

        updates = {}

        store_update(updates, group_name, 'users_group_name')

        store_update(updates, description, 'user_group_description')

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "failed to delete user group ID:<user_group_id> <user_group_name>"

            or

            "RepoGroup assigned to <repo_groups_list>"

          }

        """

        user_group = get_user_group_or_error(usergroupid)

        if not HasPermissionAny('hg.admin')():

            if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):

                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

        try:

            UserGroupModel().delete(user_group)

            meta.Session().commit()

            return dict(

                msg='deleted user group ID:%s %s' %

                    (user_group.users_group_id, user_group.users_group_name),

        :param usergroupid:

        :type usergroupid: str or int

        :param userid:

        :type userid: str or int

        OUTPUT::

          id : <id_given_in_input>

          result : {

              "success": True|False # depends on if member is in group

              "msg": "added member `<username>` to a user group `<groupname>` |

                      User is already in that group"

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "failed to add member to user group `<user_group_name>`"

          }

        """

        user = get_user_or_error(userid)

        user_group = get_user_group_or_error(usergroupid)

        if not HasPermissionAny('hg.admin')():

            if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):

                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

        try:

            ugm = UserGroupModel().add_user_to_group(user_group, user)

            success = True if ugm is not True else False

            msg = 'added member `%s` to user group `%s`' % (

                user.username, user_group.users_group_name

            )

    # permission check inside

    def remove_user_from_user_group(self, usergroupid, userid):

        """

        Removes a user from a user group. If user is not in given group success will

        be `false`. This command can be executed only

        using api_key belonging to user with admin rights or an admin of given user group

        :param usergroupid:

        :param userid:

        OUTPUT::

            id : <id_given_in_input>

            result: {

                      "success":  True|False,  # depends on if member is in group

                      "msg": "removed member <username> from user group <groupname> |

                              User wasn't in group"

                    }

            error:  null

        """

        user = get_user_or_error(userid)

        user_group = get_user_group_or_error(usergroupid)

        if not HasPermissionAny('hg.admin')():

            if not HasUserGroupPermissionLevel('admin')(user_group.users_group_name):

                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

        try:

            success = UserGroupModel().remove_user_from_group(user_group, user)

            msg = 'removed member `%s` from user group `%s`' % (

                user.username, user_group.users_group_name

            )

                            ...

                         },

                 "branches": {

                            "<branchname>": "<raw_id>",

                            ...

                         },

                 "bookmarks": {

                            "<bookmarkname>": "<raw_id>",

                            ...

                         },

          }

          error :  null

        """

        repo = get_repo_or_error(repoid)

        if not HasPermissionAny('hg.admin')():

            if not HasRepoPermissionLevel('read')(repo.repo_name):

                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

        members = []

        for user in repo.repo_to_perm:

            perm = user.permission.permission_name

            user = user.user

            user_data = {

                                 with_pullrequests=with_pullrequests)

        data['members'] = members

        data['followers'] = followers

        return data

    # permission check inside

    def get_repos(self):

        """

        Lists all existing repositories. This command can be executed only using

        api_key belonging to user with admin rights or regular user that have

        admin, write or read access to repository.

        OUTPUT::

            id : <id_given_in_input>

            result: [

                      {

                        "repo_id" :          "<repo_id>",

                        "repo_name" :        "<reponame>"

                        "repo_type" :        "<repo_type>",

                        "clone_uri" :        "<clone_uri>",

                        "private" :          "<bool>",

                        "created_on" :       "<datetimecreated>",

                        "description" :      "<description>",

        `dirs`.  This command can be executed only using api_key belonging to

        user with admin rights or regular user that have at least read access to repository.

        :param repoid: repository name or repository id

        :type repoid: str or int

        :param revision: revision for which listing should be done

        :type revision: str

        :param root_path: path from which start displaying

        :type root_path: str

        :param ret_type: return type 'all|files|dirs' nodes

        :type ret_type: Optional(str)

        OUTPUT::

            id : <id_given_in_input>

            result: [

                      {

                        "name" :        "<name>"

                        "type" :        "<type>",

                      },

                      …

                    ]

            error:  null

        """

                      "msg": "Created new repository `<reponame>`",

                      "success": true

                    }

            error:  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

             'failed to create repository `<repo_name>`

          }

        """

        group_name = None

        repo_name_parts = repo_name.split('/')

        if len(repo_name_parts) > 1:

            group_name = '/'.join(repo_name_parts[:-1])

            repo_group = db.RepoGroup.get_by_group_name(group_name)

            if repo_group is None:

                raise JSONRPCError("repo group `%s` not found" % group_name)

            if not(HasPermissionAny('hg.admin')() or HasRepoGroupPermissionLevel('write')(group_name)):

                raise JSONRPCError("no permission to create repo in %s" % group_name)

        else:

            if not HasPermissionAny('hg.admin', 'hg.create.repository')():

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to create repository `%s`' % (repo_name,))

    # permission check inside

    def update_repo(self, repoid, name=None,

                    owner=None,

                    group=None,

                    description=None, private=None,

                    clone_uri=None, landing_rev=None,

                    enable_statistics=None,

                    enable_downloads=None):

        """

        Updates repo

        :param repoid: repository name or repository id

        :type repoid: str or int

        :param name:

        :param owner:

        :param group:

        :param description:

        :param private:

        :param clone_uri:

        :param landing_rev:

                        "private":          "<bool>",

                        "landing_rev":      "<landing_rev>"

                      }

        OUTPUT::

            id : <id_given_in_input>

            result: {

                      "msg": "Created fork of `<reponame>` as `<forkname>`",

                      "success": true

                    }

            error:  null

        """

        repo = get_repo_or_error(repoid)

        repo_name = repo.repo_name

        _repo = RepoModel().get_by_repo_name(fork_name)

        if _repo:

            type_ = 'fork' if _repo.fork else 'repo'

            raise JSONRPCError("%s `%s` already exist" % (type_, fork_name))

        group_name = None

        fork_name_parts = fork_name.split('/')

        if len(fork_name_parts) > 1:

        :type repoid: str or int

        :param forks: `detach` or `delete`, what do do with attached forks for repo

        :type forks: Optional(str)

        OUTPUT::

            id : <id_given_in_input>

            result: {

                      "msg": "Deleted repository `<reponame>`",

                      "success": true

                    }

            error:  null

        """

        repo = get_repo_or_error(repoid)

        if not HasPermissionAny('hg.admin')():

            if not HasRepoPermissionLevel('admin')(repo.repo_name):

                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

        try:

            handle_forks = forks

            _forks_msg = ''

            _forks = [f for f in repo.forks]

            if handle_forks == 'detach':

        :param repoid: repository name or repository id

        :type repoid: str or int

        :param userid:

        OUTPUT::

            id : <id_given_in_input>

            result: {

                      "msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",

                      "success": true

                    }

            error:  null

        """

        repo = get_repo_or_error(repoid)

        user = get_user_or_error(userid)

        try:

            RepoModel().revoke_user_permission(repo=repo, user=user)

            meta.Session().commit()

            return dict(

                msg='Revoked perm for user: `%s` in repo: `%s`' % (

                    user.username, repo.repo_name

                ),

                success=True

            )

        except Exception:

        :type repoid: str or int

        :param usergroupid: id of usergroup

        :type usergroupid: str or int

        :param perm: (repository.(none|read|write|admin))

        :type perm: str

        OUTPUT::

          id : <id_given_in_input>

          result : {

            "msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",

            "success": true

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "failed to edit permission for user group: `<usergroup>` in repo `<repo>`'

          }

        """

        repo = get_repo_or_error(repoid)

        perm = get_perm_or_error(perm)

        user_group = get_user_group_or_error(usergroupid)

        if not HasPermissionAny('hg.admin')():

            if not HasRepoPermissionLevel('admin')(repo.repo_name):

                raise JSONRPCError('repository `%s` does not exist' % (repoid,))

            if not HasUserGroupPermissionLevel('read')(user_group.users_group_name):

                raise JSONRPCError('user group `%s` does not exist' % (usergroupid,))

        try:

                'permission': perm

            }

            members.append(user_group_data)

        data = repo_group.get_api_data()

        data["members"] = members

        return data

    @HasPermissionAnyDecorator('hg.admin')

    def get_repo_groups(self):

        """

        Returns all repository groups

        """

        return [

            repo_group.get_api_data()

            for repo_group in db.RepoGroup.query()

        ]

    @HasPermissionAnyDecorator('hg.admin')

    def create_repo_group(self, group_name, description='',

                          owner=None,

                          parent=None,

                          copy_permissions=False):

        """

              "msg": "created new repo group `<repo_group_name>`"

              "repo_group": <repogroup_object>

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            failed to create repo group `<repogroupid>`

          }

        """

        if db.RepoGroup.get_by_group_name(group_name):

            raise JSONRPCError("repo group `%s` already exist" % (group_name,))

        if owner is None:

            owner = request.authuser.user_id

        group_description = description

        parent_group = None

        if parent is not None:

            parent_group = get_repo_group_or_error(parent)

        try:

                msg='updated repository group ID:%s %s' % (repo_group.group_id,

                                                           repo_group.group_name),

                repo_group=repo_group.get_api_data()

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError('failed to update repository group `%s`'

                               % (repogroupid,))

    @HasPermissionAnyDecorator('hg.admin')

    def delete_repo_group(self, repogroupid):

        """

        :param repogroupid: name or id of repository group

        :type repogroupid: str or int

        OUTPUT::

          id : <id_given_in_input>

          result : {

            'msg': 'deleted repo group ID:<repogroupid> <repogroupname>

            'repo_group': null

          }

          error :  null

        ERROR OUTPUT::

          id : <id_given_in_input>

          result : null

          error :  {

            "failed to delete repo group ID:<repogroupid> <repogroupname>"

          }

        """

        repo_group = get_repo_group_or_error(repogroupid)

        try:

            RepoGroupModel().delete(repo_group)

            meta.Session().commit()

            return dict(

                msg='deleted repo group ID:%s %s' %