api_key : "<api_key>"

    method :  "get_repo"

    args:     {

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

                "with_revision_names": "<bool> = Optional(False)",

                "with_pullrequests": "<bool> = Optional(False)",

              }

OUTPUT::

    id : <id_given_in_input>

    result: None if repository does not exist or

            {

                "repo_id" :          "<repo_id>",

                "repo_name" :        "<reponame>"

                "repo_type" :        "<repo_type>",

                "clone_uri" :        "<clone_uri>",

                "enable_downloads":  "<bool>",

                "enable_statistics": "<bool>",

                "private":           "<bool>",

                "created_on" :       "<date_time_created>",

                "description" :      "<description>",

                "landing_rev":       "<landing_rev>",

                "last_changeset":    {

                                     },

                "owner":             "<repo_owner>",

                "fork_of":           "<name_of_fork_parent>",

                                  {

                                    "type":        "user",

                                    "user_id" :    "<user_id>",

                                    "api_key" :    "<api_key>",

                                    "username" :   "<username>",

                                    "firstname":   "<firstname>",

                                    "lastname" :   "<lastname>",

                                    "email" :      "<email>",

                                    "emails":      "<list_of_all_additional_emails>",

                                    "active" :     "<bool>",

                                    "admin" :      "<bool>",

                                    "ldap_dn" :    "<ldap_dn>",

                                    "last_login":  "<last_login>",

                                    "permission" : "repository.(read|write|admin)"

                                  },

                                  …

                                  {

                                    "type":      "users_group",

                                    "id" :       "<usersgroupid>",

                                    "name" :     "<usersgroupname>",

                                    "active":    "<bool>",

                                    "permission" : "repository.(read|write|admin)"

                                  },

                                  …

                                  {

                                    "user_id" :     "<user_id>",

                                    "username" :    "<username>",

                                    "api_key" :     "<api_key>",

                                    "firstname":    "<firstname>",

                                    "lastname" :    "<lastname>",

                                    "email" :       "<email>",

                                    "emails":       "<list_of_all_additional_emails>",

                                    "ip_addresses": "<list_of_ip_addresses_for_user>",

                                    "active" :      "<bool>",

                                    "admin" :       "<bool>",

                                    "ldap_dn" :     "<ldap_dn>",

                                    "last_login":   "<last_login>",

                                  },

                                  …

                            "<tagname>": "<raw_id>",

                            ...

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

                            ...

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

                            ...

                <if with_pullrequests == True>

                "pull_requests": [

                  {

                    "status": "<pull_request_status>",

                    "pull_request_id": <pull_request_id>,

                    "description": "<pull_request_description>",

                    "title": "<pull_request_title>",

                    "url": "<pull_request_url>",

                    "reviewers": [

                      {

                        "username": "<user_id>",

                      },

                      ...

                    ],

                    "org_repo_url": "<repo_url>",

                    "org_ref_parts": [

                      "<ref_type>",

                      "<ref_name>",

                      "<raw_id>"

                    ],

                    "other_ref_parts": [

                      "<ref_type>",

                      "<ref_name>",

                      "<raw_id>"

            pass

    Each function should also **raise** JSONRPCError for any

    errors that happens.

    """

    @HasPermissionAnyDecorator('hg.admin')

    def test(self, args):

        return args

    @HasPermissionAnyDecorator('hg.admin')

    def pull(self, repoid, clone_uri=None):

        """

        Triggers a pull from remote location on given repo. Can be used to

        automatically keep remote repos up to date. This command can be executed

        only using api_key belonging to user with admin rights

        :param repoid: repository name or repository id

        :type repoid: str or int

        :param clone_uri: repository URI to pull from (optional)

        :type clone_uri: str

        OUTPUT::

        ERROR OUTPUT::

        """

        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:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'Unable to pull changes from `%s`' % repo.repo_name

            )

    @HasPermissionAnyDecorator('hg.admin')

    def rescan_repos(self, remove_obsolete=False):

        """

        Triggers rescan repositories action. If remove_obsolete is set

        than also delete repos that are in database but not in the filesystem.

        aka "clean zombies". This command can be executed only using api_key

        belonging to user with admin rights.

        :param remove_obsolete: deletes repositories from

            database that are not found on the filesystem

        :type remove_obsolete: Optional(bool)

        OUTPUT::

        ERROR OUTPUT::

        """

        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):

        """

        Invalidate cache for repository.

        This command can be executed only using api_key belonging to user with admin

        rights or regular user that have write or admin or write access to repository.

        :param repoid: repository name or repository id

        :type repoid: str or int

        OUTPUT::

        ERROR OUTPUT::

        """

        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

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'Error occurred during cache invalidation action'

            )

    @HasPermissionAnyDecorator('hg.admin')

    def get_ip(self, userid=None):

        """

        Shows IP address as seen from Kallithea server, together with all

        defined IP addresses for given user. If userid is not passed data is

        returned for user who's calling this function.

                                        ...

                                     ]

            }

            error : null

        """

        if userid is None:

            userid = request.authuser.user_id

        user = get_user_or_error(userid)

        ips = db.UserIpMap.query().filter(db.UserIpMap.user == user).all()

        return dict(

            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::

        """

        return db.Setting.get_server_info()

    def get_user(self, userid=None):

        """

        Gets a user by username or user_id, Returns empty result if user is

        not found. If userid param is skipped it is set to id of user who is

        calling this method. This command can be executed only using api_key

        belonging to user with admin rights, or regular users that cannot

        specify different userid than theirs

        :param userid: user to get data for

        :type userid: Optional(str or int)

        OUTPUT::

            id : <id_given_in_input>

            result: None if user does not exist or

                    {

                        "user_id" :     "<user_id>",

                        "username" :    "<username>",

                        "firstname":    "<firstname>",

                        "lastname" :    "<lastname>",

                        "email" :       "<email>",

        :param firstname: firstname

        :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::

        """

        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,

                lastname=lastname,

                active=active,

                admin=admin,

                extern_type=extern_type,

                extern_name=extern_name

            )

            meta.Session().commit()

            return dict(

                msg='created new user `%s`' % username,

                user=user.get_api_data()

            )

        :param 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::

        """

        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')

            store_update(updates, active, 'active')

            store_update(updates, admin, 'admin')

            store_update(updates, extern_name, 'extern_name')

            store_update(updates, extern_type, 'extern_type')

            user = UserModel().update_user(user, **updates)

            meta.Session().commit()

            return dict(

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

                user=user.get_api_data()

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError('failed to update user `%s`' % (userid,))

    @HasPermissionAnyDecorator('hg.admin')

    def delete_user(self, userid):

        """

        deletes given user if such user exists. This command can

        be executed only using api_key belonging to user with admin rights.

        :param userid: user to delete

        :type userid: str or int

        OUTPUT::

            id : <id_given_in_input>

            result: {

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

                      "user": null

                    }

            error:  null

        ERROR OUTPUT::

        """

        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:

            log.error(traceback.format_exc())

            raise JSONRPCError('failed to delete user ID:%s %s'

                               % (user.user_id, user.username))

    # permission check inside

    def get_user_group(self, usergroupid):

        """

        Gets an existing user group. 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.

        :param usergroupid: id of user_group to edit

        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

        :param group_name: name of new user group

        :type group_name: str

        :param description: group description

        :type description: Optional(str)

        :param owner: owner of group. If not passed apiuser is the owner

        :type owner: Optional(str or int)

        :param active: group is active

        :type active: Optional(bool)

        OUTPUT::

            id : <id_given_in_input>

            result: {

                      "msg": "created new user group `<groupname>`",

                      "user_group": <user_group_object>

                    }

            error:  null

        ERROR OUTPUT::

        """

        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='created new user group `%s`' % group_name,

                user_group=ug.get_api_data()

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError('failed to create group `%s`' % (group_name,))

    # permission check inside

    def update_user_group(self, usergroupid, group_name=None,

                          description=None, owner=None,

                          active=None):

        except UserGroupsAssignedException as e:

            log.error(traceback.format_exc())

            raise JSONRPCError(str(e))

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError('failed to delete user group ID:%s %s' %

                               (user_group.users_group_id,

                                user_group.users_group_name)

                               )

    # permission check inside

    def add_user_to_user_group(self, usergroupid, userid):

        """

        Adds a user to a user group. If user exists in that group success will be

        `false`. This command can be executed only using api_key

        belonging to user with admin rights or an admin of a given user group

        :param usergroupid:

        :type usergroupid: str or int

        :param userid:

        :type userid: str or int

        OUTPUT::

        ERROR OUTPUT::

        """

        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

            )

            msg = msg if success else 'User is already in that group'

            meta.Session().commit()

            return dict(

                success=success,

                msg=msg

            )

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to add member to user group `%s`' % (

            return dict(success=success, msg=msg)

        except Exception:

            log.error(traceback.format_exc())

            raise JSONRPCError(

                'failed to remove member from user group `%s`' % (

                    user_group.users_group_name,

                )

            )

    # permission check inside

    def get_repo(self, repoid,

                 with_revision_names=False,

                 with_pullrequests=False):

        """

        Gets an existing repository by it's name or repository_id. Members will return

        either users_group or user associated to that repository. 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

        OUTPUT::

        """

        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 = {

                'name': user.username,

                'type': "user",

                'permission': perm

            }

            members.append(user_data)

        for user_group in repo.users_group_to_perm:

            perm = user_group.permission.permission_name

            user_group = user_group.users_group

            user_group_data = {

                'name': user_group.users_group_name,

                'type': "user_group",

        :type private: bool

        :param clone_uri:

        :type clone_uri: str

        :param landing_rev: <rev_type>:<rev>

        :type landing_rev: str

        :param enable_downloads:

        :type enable_downloads: bool

        :param enable_statistics:

        :type enable_statistics: bool

        :param copy_permissions: Copy permission from group that repository is

            being created.

        :type copy_permissions: bool

        OUTPUT::

            id : <id_given_in_input>

            result: {

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

                      "success": true

                    }