Changeset - d69c3e30f8b5
[Not reviewed]
stable
0 1 0
Mads Kiilerich (mads) - 3 years ago 2022-10-13 19:37:43
mads@kiilerich.com
Grafted from: 9f919f8b2eac
api docs: update docstrings

Changes contributed by toras <toras9000@gmail.com>.
1 file changed with 22 insertions and 25 deletions:
0 comments (0 inline, 0 general)
kallithea/controllers/api/api.py
Show inline comments
 
@@ -296,280 +296,277 @@ class ApiController(JSONRPCController):
 
    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.
 
        This command can be executed only using api_key belonging to user with
 
        admin rights.
 

	
 
        :param userid: username to show ips for
 
        :type userid: Optional(str or int)
 

	
 
        OUTPUT::
 

	
 
            id : <id_given_in_input>
 
            result : {
 
                         "server_ip_addr": "<ip_from_clien>",
 
                         "user_ips": [
 
                                        {
 
                                           "ip_addr": "<ip_with_mask>",
 
                                           "ip_range": ["<start_ip>", "<end_ip>"],
 
                                        },
 
                                        ...
 
                                     ]
 
            }
 

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

	
 
          id : <id_given_in_input>
 
          result : {
 
            'modules': [<module name>,...]
 
            'modules': [ [<module name>, <module version>], ...]
 
            'py_version': <python version>,
 
            'platform': <platform type>,
 
            'kallithea_version': <kallithea version>
 
            'kallithea_version': <kallithea version>,
 
            'git_version': '<git version>',
 
            'git_path': '<git path>'
 
          }
 
          error :  null
 
        """
 
        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>",
 
                        "api_key" :     "<api_key>",
 
                        "api_keys":     "[<list of all API keys including additional ones>]"
 
                        "username" :    "<username>",
 
                        "firstname":    "<firstname>",
 
                        "lastname" :    "<lastname>",
 
                        "email" :       "<email>",
 
                        "emails":       "[<list of all emails including additional ones>]",
 
                        "ip_addresses": "[<ip_address_for_user>,...]",
 
                        "active" :      "<bool: user active>",
 
                        "admin" :       "<bool: user is admin>",
 
                        "extern_name" : "<extern_name>",
 
                        "extern_type" : "<extern type>
 
                        "last_login":   "<last_login>",
 
                        "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"}
 
                            "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,
 
                    extern_name=''):
 
        """
 
        Creates new user. Returns new user object. This command can
 
        be executed only using api_key belonging to user with admin rights.
 

	
 
        :param username: new username
 
        :type username: str or int
 
        :param email: email
 
        :type email: str
 
        :param password: password
 
        :type password: Optional(str)
 
        :param firstname: firstname
 
        :type firstname: Optional(str)
 
        :type firstname: str
 
        :param lastname: lastname
 
        :type lastname: Optional(str)
 
        :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,
 
                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()
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to create user `%s`' % (username,))
 

	
 
    @HasPermissionAnyDecorator('hg.admin')
 
    def update_user(self, userid, username=None,
 
                    email=None, password=None,
 
                    firstname=None, lastname=None,
 
                    active=None, admin=None,
 
                    extern_type=None, extern_name=None):
 
        """
 
        updates given user if such user exists. This command can
 
        be executed only using api_key belonging to user with admin rights.
 

	
 
        :param userid: userid to update
 
        :type userid: str or int
 
        :param username: new username
 
        :type username: str or int
 
        :type username: Optional(str or int)
 
        :param email: email
 
        :type email: str
 
        :type email: Optional(str)
 
        :param password: password
 
        :type password: Optional(str)
 
        :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::
 

	
 
          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')
 
            store_update(updates, active, 'active')
 
@@ -604,142 +601,144 @@ class ApiController(JSONRPCController):
 
            id : <id_given_in_input>
 
            result: {
 
                      "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:
 

	
 
            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
 
        :type usergroupid: str or int
 

	
 
        OUTPUT::
 

	
 
            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
 

	
 
        :param group_name: name of new user group
 
        :type group_name: str
 
        :param description: group description
 
        :type description: str
 
        :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::
 

	
 
          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='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,
 
@@ -765,151 +764,151 @@ class ApiController(JSONRPCController):
 
          result : {
 
            "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')
 
        store_update(updates, owner, 'owner')
 
        store_update(updates, active, 'users_group_active')
 
        try:
 
            UserGroupModel().update(user_group, updates)
 
            meta.Session().commit()
 
            return dict(
 
                msg='updated user group ID:%s %s' % (user_group.users_group_id,
 
                                                     user_group.users_group_name),
 
                user_group=user_group.get_api_data()
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to update user group `%s`' % (usergroupid,))
 

	
 
    # permission check inside
 
    def delete_user_group(self, usergroupid):
 
        """
 
        Delete given user group by user group id or name.
 
        This command can be executed only using api_key
 
        belonging to user with admin rights or an admin of given user group
 

	
 
        :param usergroupid:
 
        :type usergroupid: int
 
        :type usergroupid: str or int
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            "msg": "deleted user group ID:<user_group_id> <user_group_name>"
 
          }
 
          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),
 
                user_group=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 given user group
 

	
 
        :param usergroupid:
 
        :type usergroupid: int
 
        :type usergroupid: str or int
 
        :param userid:
 
        :type userid: int
 
        :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 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
 
            )
 
            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`' % (
 
                    user_group.users_group_name,
 
                )
 
            )
 

	
 
@@ -1160,98 +1159,97 @@ class ApiController(JSONRPCController):
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError(
 
                'failed to get repo: `%s` nodes' % repo.repo_name
 
            )
 

	
 
    # permission check inside
 
    def create_repo(self, repo_name, owner=None,
 
                    repo_type=None, description='',
 
                    private=False, clone_uri=None,
 
                    landing_rev='rev:tip',
 
                    enable_statistics=None,
 
                    enable_downloads=None,
 
                    copy_permissions=False):
 
        """
 
        Creates a repository. The repository name contains the full path, but the
 
        parent repository group must exist. For example "foo/bar/baz" require the groups
 
        "foo" and "bar" (with "foo" as parent), and create "baz" repository with
 
        "bar" as group. This command can be executed only using api_key
 
        belonging to user with admin rights or regular user that have create
 
        repository permission. Regular users cannot specify owner parameter
 

	
 
        :param repo_name: repository name
 
        :type repo_name: str
 
        :param owner: user_id or username
 
        :type owner: Optional(str)
 
        :param repo_type: 'hg' or 'git'
 
        :type repo_type: Optional(str)
 
        :param description: repository description
 
        :type description: Optional(str)
 
        :param private:
 
        :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,
 
                      "task": "<celery task id or None if done sync>"
 
                      "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')():
 
                raise JSONRPCError("no permission to create top level repo")
 

	
 
        if not HasPermissionAny('hg.admin')():
 
            if owner is not None:
 
                # forbid setting owner for non-admins
 
                raise JSONRPCError(
 
                    'Only Kallithea admin can specify `owner` param'
 
                )
 
        if owner is None:
 
            owner = request.authuser.user_id
 

	
 
        owner = get_user_or_error(owner)
 

	
 
        if RepoModel().get_by_repo_name(repo_name):
 
            raise JSONRPCError("repo `%s` already exist" % repo_name)
 

	
 
        defs = db.Setting.get_default_repo_settings(strip_prefix=True)
 
        if private is None:
 
            private = defs.get('repo_private') or False
 
        if repo_type is None:
 
            repo_type = defs.get('repo_type')
 
        if enable_statistics is None:
 
            enable_statistics = defs.get('repo_enable_statistics')
 
        if enable_downloads is None:
 
            enable_downloads = defs.get('repo_enable_downloads')
 
@@ -1348,98 +1346,97 @@ class ApiController(JSONRPCController):
 
                msg='updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
 
                repository=repo.get_api_data()
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to update repo `%s`' % repoid)
 

	
 
    # permission check inside
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')
 
    def fork_repo(self, repoid, fork_name,
 
                  owner=None,
 
                  description='', copy_permissions=False,
 
                  private=False, landing_rev='rev:tip'):
 
        """
 
        Creates a fork of given repo. In case of using celery this will
 
        immediately return success message, while fork is going to be created
 
        asynchronous. This command can be executed only using api_key belonging to
 
        user with admin rights or regular user that have fork permission, and at least
 
        read access to forking repository. Regular users cannot specify owner parameter.
 

	
 
        :param repoid: repository name or repository id
 
        :type repoid: str or int
 
        :param fork_name:
 
        :param owner:
 
        :param description:
 
        :param copy_permissions:
 
        :param private:
 
        :param landing_rev:
 

	
 
        INPUT::
 

	
 
            id : <id_for_response>
 
            api_key : "<api_key>"
 
            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,
 
                      "task": "<celery task id or None if done sync>"
 
                      "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:
 
            group_name = '/'.join(fork_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')():
 
                raise JSONRPCError("no permission to create top level repo")
 

	
 
        if HasPermissionAny('hg.admin')():
 
            pass
 
        elif HasRepoPermissionLevel('read')(repo.repo_name):
 
            if owner is not None:
 
                # forbid setting owner for non-admins
 
                raise JSONRPCError(
 
                    'Only Kallithea admin can specify `owner` param'
 
                )
 
        else:
 
            raise JSONRPCError('repository `%s` does not exist' % (repoid,))
 

	
 
        if owner is None:
 
            owner = request.authuser.user_id
 

	
 
        owner = get_user_or_error(owner)
 

	
 
        try:
 
            form_data = dict(
 
                repo_name=fork_name_parts[-1],
 
                repo_name_full=fork_name,
 
                repo_group=group_name,
 
                repo_type=repo.repo_type,
 
                description=description,
 
@@ -2174,150 +2171,150 @@ class ApiController(JSONRPCController):
 

	
 
    def get_gists(self, userid=None):
 
        """
 
        Get all gists for given user. If userid is empty returned gists
 
        are for user who called the api
 

	
 
        :param userid: user to get gists for
 
        :type userid: Optional(str or int)
 
        """
 
        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:
 
            user_id = request.authuser.user_id
 
        else:
 
            user_id = get_user_or_error(userid).user_id
 

	
 
        return [
 
            gist.get_api_data()
 
            for gist in db.Gist().query()
 
                .filter_by(is_expired=False)
 
                .filter(db.Gist.owner_id == user_id)
 
                .order_by(db.Gist.created_on.desc())
 
        ]
 

	
 
    def create_gist(self, files, owner=None,
 
                    gist_type=db.Gist.GIST_PUBLIC, lifetime=-1,
 
                    description=''):
 

	
 
        """
 
        Creates new Gist
 

	
 
        :param files: files to be added to gist
 
            {'filename': {'content':'...', 'lexer': null},
 
             'filename2': {'content':'...', 'lexer': null}}
 
        :type files: dict
 
        :param owner: gist owner, defaults to api method caller
 
        :type owner: Optional(str or int)
 
        :param gist_type: type of gist 'public' or 'private'
 
        :type gist_type: Optional(str)
 
        :param lifetime: time in minutes of gist lifetime
 
        :type lifetime: Optional(int)
 
        :param description: gist description
 
        :type description: Optional(str)
 
        :type description: str
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            "msg": "created new gist",
 
            "gist": {}
 
            "gist": <gist_object>
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to create gist"
 
          }
 

	
 
        """
 
        try:
 
            if owner is None:
 
                owner = request.authuser.user_id
 

	
 
            owner = get_user_or_error(owner)
 

	
 
            gist = GistModel().create(description=description,
 
                                      owner=owner,
 
                                      ip_addr=request.ip_addr,
 
                                      gist_mapping=files,
 
                                      gist_type=gist_type,
 
                                      lifetime=lifetime)
 
            meta.Session().commit()
 
            return dict(
 
                msg='created new gist',
 
                gist=gist.get_api_data()
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to create gist')
 

	
 
    # permission check inside
 
    def delete_gist(self, gistid):
 
        """
 
        Deletes existing gist
 

	
 
        :param gistid: id of gist to delete
 
        :type gistid: str
 

	
 
        OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : {
 
            "deleted gist ID: <gist_id>",
 
            "msg": "deleted gist ID: <gist_id>",
 
            "gist": null
 
          }
 
          error :  null
 

	
 
        ERROR OUTPUT::
 

	
 
          id : <id_given_in_input>
 
          result : null
 
          error :  {
 
            "failed to delete gist ID:<gist_id>"
 
          }
 

	
 
        """
 
        gist = get_gist_or_error(gistid)
 
        if not HasPermissionAny('hg.admin')():
 
            if gist.owner_id != request.authuser.user_id:
 
                raise JSONRPCError('gist `%s` does not exist' % (gistid,))
 

	
 
        try:
 
            GistModel().delete(gist)
 
            meta.Session().commit()
 
            return dict(
 
                msg='deleted gist ID:%s' % (gist.gist_access_id,),
 
                gist=None
 
            )
 
        except Exception:
 
            log.error(traceback.format_exc())
 
            raise JSONRPCError('failed to delete gist ID:%s'
 
                               % (gist.gist_access_id,))
 

	
 
    # permission check inside
 
    def get_changesets(self, repoid, start=None, end=None, start_date=None,
 
                       end_date=None, branch_name=None, reverse=False, with_file_list=False, max_revisions=None):
 
        repo = get_repo_or_error(repoid)
 
        if not HasRepoPermissionLevel('read')(repo.repo_name):
 
            raise JSONRPCError('Access denied to repo %s' % repo.repo_name)
 

	
 
        format = "%Y-%m-%dT%H:%M:%S"
 
        try:
 
            return [e.__json__(with_file_list) for e in
 
                repo.scm_instance.get_changesets(start,
 
                                                 end,
 
                                                 datetime.strptime(start_date, format) if start_date else None,
 
                                                 datetime.strptime(end_date, format) if end_date else None,
 
                                                 branch_name,
 
                                                 reverse, max_revisions)]
 
        except EmptyRepositoryError as e:
 
            raise JSONRPCError('Repository is empty')
0 comments (0 inline, 0 general)