Changeset - 4791487dbec1
Parent rev.
Child rev.
[Not reviewed]
default
0 1 0
Thomas De Schampheleire - 5 years ago 2020-10-09 19:43:18
thomas.de_schampheleire@nokia.com
api: stop using 'Optional', 'OAttr'/'OptionalAttr' classes

There does not seem to be a good reason to use the 'Optional' and
'OptionalAttr' classes.
It makes the code harder to understand. And worse, the 'default value'
specified is not always used, which can thus give false information to
users.

The way Optional was used in the API calls is twofold:

1.either by effectively extracting a value, via Optional.extract(param).
If 'param' was indeed specified by the user, then this would yield that
user-specified value. Otherwise, it would yield the value declared in the
parameter declaration, e.g. param=Optional(defaultvalue).

2.or by checking if a variable is an instance of the Optional class. In case
a user effectively passed a value, this value will not be of the
Optional class. So if a parameter is an object of class Optional, we know
the user did not pass a value, and we can apply some default.

In the declaration of the parameter, the specified default value will only
be used if the 'extract' method is used, i.e. method 1 above.

A simpler way to address this problem of default values is just with Python
default values, using 'None' as magic value if the default will be
calculated inside the method.

The docstrings still specify something like:
type: Optional(bool)
which is humanly readable and does not necessarily refer to a class called
'Optional', so such strings are kept.
1 file changed with 103 insertions and 128 deletions:
kallithea/controllers/api/api.py
103
128
0 comments (0 inline, 0 general)
kallithea/controllers/api/api.py
➞
Show inline comments
 
@@ -27,49 +27,48 @@ Original author and date, and relevant c
 

			




	
	
	
		
 
import logging

			




	
	
	
		
 
import traceback

			




	
	
	
		
 
from datetime import datetime

			




	
	
	
		
 

			




	
	
	
		
 
from tg import request

			




	
	
	
		
 

			




	
	
	
		
 
from kallithea.controllers.api import JSONRPCController, JSONRPCError

			




	
	
	
		
 
from kallithea.lib.auth import (AuthUser, HasPermissionAny, HasPermissionAnyDecorator, HasRepoGroupPermissionLevel, HasRepoPermissionLevel,

			




	
	
	
		
 
                                HasUserGroupPermissionLevel)

			




	
	
	
		
 
from kallithea.lib.exceptions import DefaultUserException, UserGroupsAssignedException

			




	
	
	
		
 
from kallithea.lib.utils import action_logger, repo2db_mapper

			




	
	
	
		
 
from kallithea.lib.utils2 import OAttr, Optional

			




	
	
	
		
 
from kallithea.lib.vcs.backends.base import EmptyChangeset

			




	
	
	
		
 
from kallithea.lib.vcs.exceptions import EmptyRepositoryError

			




	
	
	
		
 
from kallithea.model.changeset_status import ChangesetStatusModel

			




	
	
	
		
 
from kallithea.model.comment import ChangesetCommentsModel

			




	
	
	
		
 
from kallithea.model.db import ChangesetStatus, Gist, Permission, PullRequest, RepoGroup, Repository, Setting, User, UserGroup, UserIpMap

			




	
	
	
		
 
from kallithea.model.gist import GistModel

			




	
	
	
		
 
from kallithea.model.meta import Session

			




	
	
	
		
 
from kallithea.model.pull_request import PullRequestModel

			




	
	
	
		
 
from kallithea.model.repo import RepoModel

			




	
	
	
		
 
from kallithea.model.repo_group import RepoGroupModel

			




	
	
	
		
 
from kallithea.model.scm import ScmModel, UserGroupList

			




	
	
	
		
 
from kallithea.model.user import UserModel

			




	
	
	
		
 
from kallithea.model.user_group import UserGroupModel

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
log = logging.getLogger(__name__)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def store_update(updates, attr, name):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    Stores param in updates dict if it's not instance of Optional

			




	
	
	
		
 
    allows easy updates of passed in params

			




	
	
	
		
 
    Stores param in updates dict if it's not None (i.e. if user explicitly set

			




	
	
	
		
 
    a parameter). This allows easy updates of passed in params.

			




	
	
	
		
 
    """

			




	
	
	
		
 
    if not isinstance(attr, Optional):

			




	
	
	
		
 
    if attr is not None:

			




	
	
	
		
 
        updates[name] = attr

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def get_user_or_error(userid):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    Get user by id or name or return JsonRPCError if not found

			




	
	
	
		
 

			




	
	
	
		
 
    :param userid:

			




	
	
	
		
 
    """

			




	
	
	
		
 
    user = UserModel().get_user(userid)

			




	
	
	
		
 
    if user is None:

			




	
	
	
		
 
        raise JSONRPCError("user `%s` does not exist" % (userid,))

			




	
	
		
 
@@ -152,25 +151,25 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        def func(arg1, arg2,...):

			




	
	
	
		
 
            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=Optional(None)):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -188,37 +187,37 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          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=Optional.extract(clone_uri))

			




	
	
	
		
 
                                    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=Optional(False)):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -231,25 +230,25 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 

			




	
	
	
		
 
        ERROR OUTPUT::

			




	
	
	
		
 

			




	
	
	
		
 
          id : <id_given_in_input>

			




	
	
	
		
 
          result : null

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            'Error occurred during rescan repositories action'

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            rm_obsolete = Optional.extract(remove_obsolete)

			




	
	
	
		
 
            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.

			




	
	
		
 
@@ -286,25 +285,25 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            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=Optional(OAttr('apiuser'))):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -312,25 +311,25 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            result : {

			




	
	
	
		
 
                         "server_ip_addr": "<ip_from_clien>",

			




	
	
	
		
 
                         "user_ips": [

			




	
	
	
		
 
                                        {

			




	
	
	
		
 
                                           "ip_addr": "<ip_with_mask>",

			




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

			




	
	
	
		
 
                                        },

			




	
	
	
		
 
                                        ...

			




	
	
	
		
 
                                     ]

			




	
	
	
		
 
            }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        if isinstance(userid, Optional):

			




	
	
	
		
 
        if userid is None:

			




	
	
	
		
 
            userid = request.authuser.user_id

			




	
	
	
		
 
        user = get_user_or_error(userid)

			




	
	
	
		
 
        ips = UserIpMap.query().filter(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')

			




	
	
		
 
@@ -343,25 +342,25 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 

			




	
	
	
		
 
          id : <id_given_in_input>

			




	
	
	
		
 
          result : {

			




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

			




	
	
	
		
 
            'py_version': <python version>,

			




	
	
	
		
 
            'platform': <platform type>,

			




	
	
	
		
 
            'kallithea_version': <kallithea version>

			




	
	
	
		
 
          }

			




	
	
	
		
 
          error :  null

			




	
	
	
		
 
        """

			




	
	
	
		
 
        return Setting.get_server_info()

			




	
	
	
		
 

			




	
	
	
		
 
    def get_user(self, userid=Optional(OAttr('apiuser'))):

			




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

			




	
	
	
		
 

			




	
	
		
 
@@ -388,30 +387,30 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                                       "hg.register.manual_activate"],

			




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

			




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

			




	
	
	
		
 
                         },

			




	
	
	
		
 
                    }

			




	
	
	
		
 

			




	
	
	
		
 
            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 not isinstance(userid, Optional) and userid != request.authuser.user_id:

			




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

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
	
		
 

			




	
	
	
		
 
        if isinstance(userid, Optional):

			




	
	
	
		
 
        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.

			




	
	
		
 
@@ -423,29 +422,29 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            result: [<user_object>, ...]

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        return [

			




	
	
	
		
 
            user.get_api_data()

			




	
	
	
		
 
            for user in User.query()

			




	
	
	
		
 
                .order_by(User.username)

			




	
	
	
		
 
                .filter_by(is_default_user=False)

			




	
	
	
		
 
        ]

			




	
	
	
		
 

			




	
	
	
		
 
    @HasPermissionAnyDecorator('hg.admin')

			




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

			




	
	
	
		
 
                    firstname=Optional(''), lastname=Optional(''),

			




	
	
	
		
 
                    active=Optional(True), admin=Optional(False),

			




	
	
	
		
 
                    extern_type=Optional(User.DEFAULT_AUTH_TYPE),

			




	
	
	
		
 
                    extern_name=Optional('')):

			




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

			




	
	
	
		
 
                    firstname='', lastname='',

			




	
	
	
		
 
                    active=True, admin=False,

			




	
	
	
		
 
                    extern_type=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)

			




	
	
		
 
@@ -483,49 +482,49 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        if User.get_by_username(username):

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        if User.get_by_email(email):

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            user = UserModel().create_or_update(

			




	
	
	
		
 
                username=Optional.extract(username),

			




	
	
	
		
 
                password=Optional.extract(password),

			




	
	
	
		
 
                email=Optional.extract(email),

			




	
	
	
		
 
                firstname=Optional.extract(firstname),

			




	
	
	
		
 
                lastname=Optional.extract(lastname),

			




	
	
	
		
 
                active=Optional.extract(active),

			




	
	
	
		
 
                admin=Optional.extract(admin),

			




	
	
	
		
 
                extern_type=Optional.extract(extern_type),

			




	
	
	
		
 
                extern_name=Optional.extract(extern_name)

			




	
	
	
		
 
                username=username,

			




	
	
	
		
 
                password=password,

			




	
	
	
		
 
                email=email,

			




	
	
	
		
 
                firstname=firstname,

			




	
	
	
		
 
                lastname=lastname,

			




	
	
	
		
 
                active=active,

			




	
	
	
		
 
                admin=admin,

			




	
	
	
		
 
                extern_type=extern_type,

			




	
	
	
		
 
                extern_name=extern_name

			




	
	
	
		
 
            )

			




	
	
	
		
 
            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=Optional(None),

			




	
	
	
		
 
                    email=Optional(None), password=Optional(None),

			




	
	
	
		
 
                    firstname=Optional(None), lastname=Optional(None),

			




	
	
	
		
 
                    active=Optional(None), admin=Optional(None),

			




	
	
	
		
 
                    extern_type=Optional(None), extern_name=Optional(None)):

			




	
	
	
		
 
    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

			




	
	
	
		
 
        :param email: email

			




	
	
	
		
 
        :type email: str

			




	
	
	
		
 
        :param password: password

			




	
	
	
		
 
        :type password: Optional(str)

			




	
	
		
 
@@ -677,26 +676,26 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 

			




	
	
	
		
 
            id : <id_given_in_input>

			




	
	
	
		
 
            result : [<user_group_obj>,...]

			




	
	
	
		
 
            error : null

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        return [

			




	
	
	
		
 
            user_group.get_api_data()

			




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

			




	
	
	
		
 
        ]

			




	
	
	
		
 

			




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

			




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

			




	
	
	
		
 
                          owner=Optional(OAttr('apiuser')), active=Optional(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

			




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

			




	
	
	
		
 
        :type owner: Optional(str or int)

			




	
	
	
		
 
        :param active: group is active

			




	
	
		
 
@@ -718,45 +717,43 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          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 isinstance(owner, Optional):

			




	
	
	
		
 
            if owner is None:

			




	
	
	
		
 
                owner = request.authuser.user_id

			




	
	
	
		
 

			




	
	
	
		
 
            owner = get_user_or_error(owner)

			




	
	
	
		
 
            active = Optional.extract(active)

			




	
	
	
		
 
            description = Optional.extract(description)

			




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

			




	
	
	
		
 
                                         owner=owner, active=active)

			




	
	
	
		
 
            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=Optional(''),

			




	
	
	
		
 
                          description=Optional(''), owner=Optional(None),

			




	
	
	
		
 
                          active=Optional(True)):

			




	
	
	
		
 
    def update_user_group(self, usergroupid, group_name=None,

			




	
	
	
		
 
                          description=None, owner=None,

			




	
	
	
		
 
                          active=None):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Updates given usergroup.  This command can be executed only using api_key

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        :param usergroupid: id of user group to update

			




	
	
	
		
 
        :type usergroupid: str or int

			




	
	
	
		
 
        :param group_name: name of new user group

			




	
	
	
		
 
        :type group_name: str

			




	
	
	
		
 
        :param description: group description

			




	
	
	
		
 
        :type description: str

			




	
	
	
		
 
        :param owner: owner of group.

			




	
	
	
		
 
        :type owner: Optional(str or int)

			




	
	
		
 
@@ -777,25 +774,25 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          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 not isinstance(owner, Optional):

			




	
	
	
		
 
        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)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




	
	
	
		
 
                msg='updated user group ID:%s %s' % (user_group.users_group_id,

			




	
	
		
 
@@ -954,26 +951,26 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            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=Optional(False),

			




	
	
	
		
 
                 with_pullrequests=Optional(False)):

			




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

			




	
	
	
		
 

			




	
	
	
		
 
          id : <id_given_in_input>

			




	
	
		
 
@@ -1055,26 +1052,26 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            user_group_data = {

			




	
	
	
		
 
                'name': user_group.users_group_name,

			




	
	
	
		
 
                'type': "user_group",

			




	
	
	
		
 
                'permission': perm

			




	
	
	
		
 
            }

			




	
	
	
		
 
            members.append(user_group_data)

			




	
	
	
		
 

			




	
	
	
		
 
        followers = [

			




	
	
	
		
 
            uf.user.get_api_data()

			




	
	
	
		
 
            for uf in repo.followers

			




	
	
	
		
 
        ]

			




	
	
	
		
 

			




	
	
	
		
 
        data = repo.get_api_data(with_revision_names=Optional.extract(with_revision_names),

			




	
	
	
		
 
                                 with_pullrequests=Optional.extract(with_pullrequests))

			




	
	
	
		
 
        data = repo.get_api_data(with_revision_names=with_revision_names,

			




	
	
	
		
 
                                 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.

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
		
 
@@ -1103,25 +1100,25 @@ class ApiController(JSONRPCController):

			




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

			




	
	
	
		
 
            repos = RepoModel().get_all_user_repos(user=request.authuser.user_id)

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            repos = Repository.query()

			




	
	
	
		
 

			




	
	
	
		
 
        return [

			




	
	
	
		
 
            repo.get_api_data()

			




	
	
	
		
 
            for repo in repos

			




	
	
	
		
 
        ]

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def get_repo_nodes(self, repoid, revision, root_path,

			




	
	
	
		
 
                       ret_type=Optional('all')):

			




	
	
	
		
 
                       ret_type='all'):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        returns a list of nodes and it's children in a flat list for a given path

			




	
	
	
		
 
        at given revision. It's possible to specify ret_type to show only `files` or

			




	
	
	
		
 
        `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

			




	
	
		
 
@@ -1138,52 +1135,52 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                        "type" :        "<type>",

			




	
	
	
		
 
                      },

			




	
	
	
		
 
                      …

			




	
	
	
		
 
                    ]

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        ret_type = Optional.extract(ret_type)

			




	
	
	
		
 
        ret_type = ret_type

			




	
	
	
		
 
        _map = {}

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            _d, _f = ScmModel().get_nodes(repo, revision, root_path,

			




	
	
	
		
 
                                          flat=False)

			




	
	
	
		
 
            _map = {

			




	
	
	
		
 
                'all': _d + _f,

			




	
	
	
		
 
                'files': _f,

			




	
	
	
		
 
                'dirs': _d,

			




	
	
	
		
 
            }

			




	
	
	
		
 
            return _map[ret_type]

			




	
	
	
		
 
        except KeyError:

			




	
	
	
		
 
            raise JSONRPCError('ret_type must be one of %s'

			




	
	
	
		
 
                               % (','.join(sorted(_map))))

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise JSONRPCError(

			




	
	
	
		
 
                'failed to get repo: `%s` nodes' % repo.repo_name

			




	
	
	
		
 
            )

			




	
	
	
		
 

			




	
	
	
		
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.create.repository')

			




	
	
	
		
 
    def create_repo(self, repo_name, owner=Optional(OAttr('apiuser')),

			




	
	
	
		
 
                    repo_type=Optional('hg'), description=Optional(''),

			




	
	
	
		
 
                    private=Optional(False), clone_uri=Optional(None),

			




	
	
	
		
 
                    landing_rev=Optional('rev:tip'),

			




	
	
	
		
 
                    enable_statistics=Optional(False),

			




	
	
	
		
 
                    enable_downloads=Optional(False),

			




	
	
	
		
 
                    copy_permissions=Optional(False)):

			




	
	
	
		
 
    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)

			




	
	
		
 
@@ -1216,52 +1213,47 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            error:  null

			




	
	
	
		
 

			




	
	
	
		
 
        ERROR OUTPUT::

			




	
	
	
		
 

			




	
	
	
		
 
          id : <id_given_in_input>

			




	
	
	
		
 
          result : null

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
             'failed to create repository `<repo_name>`

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




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

			




	
	
	
		
 
            if not isinstance(owner, Optional):

			




	
	
	
		
 
            if owner is not None:

			




	
	
	
		
 
                # forbid setting owner for non-admins

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'Only Kallithea admin can specify `owner` param'

			




	
	
	
		
 
                )

			




	
	
	
		
 
        if isinstance(owner, Optional):

			




	
	
	
		
 
        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 = Setting.get_default_repo_settings(strip_prefix=True)

			




	
	
	
		
 
        if isinstance(private, Optional):

			




	
	
	
		
 
            private = defs.get('repo_private') or Optional.extract(private)

			




	
	
	
		
 
        if isinstance(repo_type, Optional):

			




	
	
	
		
 
        if private is None:

			




	
	
	
		
 
            private = defs.get('repo_private') or False

			




	
	
	
		
 
        if repo_type is None:

			




	
	
	
		
 
            repo_type = defs.get('repo_type')

			




	
	
	
		
 
        if isinstance(enable_statistics, Optional):

			




	
	
	
		
 
        if enable_statistics is None:

			




	
	
	
		
 
            enable_statistics = defs.get('repo_enable_statistics')

			




	
	
	
		
 
        if isinstance(enable_downloads, Optional):

			




	
	
	
		
 
        if enable_downloads is None:

			




	
	
	
		
 
            enable_downloads = defs.get('repo_enable_downloads')

			




	
	
	
		
 

			




	
	
	
		
 
        clone_uri = Optional.extract(clone_uri)

			




	
	
	
		
 
        description = Optional.extract(description)

			




	
	
	
		
 
        landing_rev = Optional.extract(landing_rev)

			




	
	
	
		
 
        copy_permissions = Optional.extract(copy_permissions)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            repo_name_parts = repo_name.split('/')

			




	
	
	
		
 
            repo_group = None

			




	
	
	
		
 
            if len(repo_name_parts) > 1:

			




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

			




	
	
	
		
 
                repo_group = RepoGroup.get_by_group_name(group_name)

			




	
	
	
		
 
                if repo_group is None:

			




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

			




	
	
	
		
 
            data = dict(

			




	
	
	
		
 
                repo_name=repo_name_parts[-1],

			




	
	
	
		
 
                repo_name_full=repo_name,

			




	
	
	
		
 
                repo_type=repo_type,

			




	
	
		
 
@@ -1282,31 +1274,31 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            return dict(

			




	
	
	
		
 
                msg="Created new repository `%s`" % (repo_name,),

			




	
	
	
		
 
                success=True,  # cannot return the repo data here since fork

			




	
	
	
		
 
                               # can be done async

			




	
	
	
		
 
                task=task_id

			




	
	
	
		
 
            )

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise JSONRPCError(

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def update_repo(self, repoid, name=Optional(None),

			




	
	
	
		
 
                    owner=Optional(OAttr('apiuser')),

			




	
	
	
		
 
                    group=Optional(None),

			




	
	
	
		
 
                    description=Optional(''), private=Optional(False),

			




	
	
	
		
 
                    clone_uri=Optional(None), landing_rev=Optional('rev:tip'),

			




	
	
	
		
 
                    enable_statistics=Optional(False),

			




	
	
	
		
 
                    enable_downloads=Optional(False)):

			




	
	
	
		
 
    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:

			




	
	
		
 
@@ -1315,33 +1307,33 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        :param enable_downloads:

			




	
	
	
		
 
        """

			




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

			




	
	
	
		
 

			




	
	
	
		
 
            if (name != repo.repo_name and

			




	
	
	
		
 
                not HasPermissionAny('hg.create.repository')()

			




	
	
	
		
 
            ):

			




	
	
	
		
 
                raise JSONRPCError('no permission to create (or move) repositories')

			




	
	
	
		
 

			




	
	
	
		
 
            if not isinstance(owner, Optional):

			




	
	
	
		
 
            if owner is not None:

			




	
	
	
		
 
                # forbid setting owner for non-admins

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'Only Kallithea admin can specify `owner` param'

			




	
	
	
		
 
                )

			




	
	
	
		
 

			




	
	
	
		
 
        updates = {}

			




	
	
	
		
 
        repo_group = group

			




	
	
	
		
 
        if not isinstance(repo_group, Optional):

			




	
	
	
		
 
        if repo_group is not None:

			




	
	
	
		
 
            repo_group = get_repo_group_or_error(repo_group)

			




	
	
	
		
 
            repo_group = repo_group.group_id

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            store_update(updates, name, 'repo_name')

			




	
	
	
		
 
            store_update(updates, repo_group, 'repo_group')

			




	
	
	
		
 
            store_update(updates, owner, 'owner')

			




	
	
	
		
 
            store_update(updates, description, 'repo_description')

			




	
	
	
		
 
            store_update(updates, private, 'repo_private')

			




	
	
	
		
 
            store_update(updates, clone_uri, 'clone_uri')

			




	
	
	
		
 
            store_update(updates, landing_rev, 'repo_landing_rev')

			




	
	
	
		
 
            store_update(updates, enable_statistics, 'repo_enable_statistics')

			




	
	
	
		
 
            store_update(updates, enable_downloads, 'repo_enable_downloads')

			




	
	
		
 
@@ -1349,27 +1341,27 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            RepoModel().update(repo, **updates)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




	
	
	
		
 
                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)

			




	
	
	
		
 

			




	
	
	
		
 
    @HasPermissionAnyDecorator('hg.admin', 'hg.fork.repository')

			




	
	
	
		
 
    def fork_repo(self, repoid, fork_name,

			




	
	
	
		
 
                  owner=Optional(OAttr('apiuser')),

			




	
	
	
		
 
                  description=Optional(''), copy_permissions=Optional(False),

			




	
	
	
		
 
                  private=Optional(False), landing_rev=Optional('rev:tip')):

			




	
	
	
		
 
                  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:

			




	
	
		
 
@@ -1404,80 +1396,80 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        """

			




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

			




	
	
	
		
 

			




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

			




	
	
	
		
 
            pass

			




	
	
	
		
 
        elif HasRepoPermissionLevel('read')(repo.repo_name):

			




	
	
	
		
 
            if not isinstance(owner, Optional):

			




	
	
	
		
 
            if owner is not None:

			




	
	
	
		
 
                # forbid setting owner for non-admins

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'Only Kallithea admin can specify `owner` param'

			




	
	
	
		
 
                )

			




	
	
	
		
 

			




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

			




	
	
	
		
 
                raise JSONRPCError('no permission to create repositories')

			




	
	
	
		
 
        else:

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        if isinstance(owner, Optional):

			




	
	
	
		
 
        if owner is None:

			




	
	
	
		
 
            owner = request.authuser.user_id

			




	
	
	
		
 

			




	
	
	
		
 
        owner = get_user_or_error(owner)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            fork_name_parts = fork_name.split('/')

			




	
	
	
		
 
            repo_group = None

			




	
	
	
		
 
            if len(fork_name_parts) > 1:

			




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

			




	
	
	
		
 
                repo_group = RepoGroup.get_by_group_name(group_name)

			




	
	
	
		
 
                if repo_group is None:

			




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

			




	
	
	
		
 

			




	
	
	
		
 
            form_data = dict(

			




	
	
	
		
 
                repo_name=fork_name_parts[-1],

			




	
	
	
		
 
                repo_name_full=fork_name,

			




	
	
	
		
 
                repo_group=repo_group,

			




	
	
	
		
 
                repo_type=repo.repo_type,

			




	
	
	
		
 
                description=Optional.extract(description),

			




	
	
	
		
 
                private=Optional.extract(private),

			




	
	
	
		
 
                copy_permissions=Optional.extract(copy_permissions),

			




	
	
	
		
 
                landing_rev=Optional.extract(landing_rev),

			




	
	
	
		
 
                description=description,

			




	
	
	
		
 
                private=private,

			




	
	
	
		
 
                copy_permissions=copy_permissions,

			




	
	
	
		
 
                landing_rev=landing_rev,

			




	
	
	
		
 
                update_after_clone=False,

			




	
	
	
		
 
                fork_parent_id=repo.repo_id,

			




	
	
	
		
 
            )

			




	
	
	
		
 
            task = RepoModel().create_fork(form_data, cur_user=owner)

			




	
	
	
		
 
            # no commit, it's done in RepoModel, or async via celery

			




	
	
	
		
 
            task_id = task.task_id

			




	
	
	
		
 
            return dict(

			




	
	
	
		
 
                msg='Created fork of `%s` as `%s`' % (repo.repo_name,

			




	
	
	
		
 
                                                      fork_name),

			




	
	
	
		
 
                success=True,  # cannot return the repo data here since fork

			




	
	
	
		
 
                               # can be done async

			




	
	
	
		
 
                task=task_id

			




	
	
	
		
 
            )

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise JSONRPCError(

			




	
	
	
		
 
                'failed to fork repository `%s` as `%s`' % (repo_name,

			




	
	
	
		
 
                                                            fork_name)

			




	
	
	
		
 
            )

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def delete_repo(self, repoid, forks=Optional('')):

			




	
	
	
		
 
    def delete_repo(self, repoid, forks=''):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Deletes a repository. This command can be executed only using api_key belonging

			




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

			




	
	
	
		
 
        When `forks` param is set it's possible to detach or delete forks of deleting

			




	
	
	
		
 
        repository

			




	
	
	
		
 

			




	
	
	
		
 
        :param repoid: repository name or repository id

			




	
	
	
		
 
        :type repoid: str or int

			




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

			




	
	
	
		
 
        :type forks: Optional(str)

			




	
	
	
		
 

			




	
	
	
		
 
        OUTPUT::

			




	
	
		
 
@@ -1488,25 +1480,25 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
                      "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 = Optional.extract(forks)

			




	
	
	
		
 
            handle_forks = forks

			




	
	
	
		
 
            _forks_msg = ''

			




	
	
	
		
 
            _forks = [f for f in repo.forks]

			




	
	
	
		
 
            if handle_forks == 'detach':

			




	
	
	
		
 
                _forks_msg = ' ' + 'Detached %s forks' % len(_forks)

			




	
	
	
		
 
            elif handle_forks == 'delete':

			




	
	
	
		
 
                _forks_msg = ' ' + 'Deleted %s forks' % len(_forks)

			




	
	
	
		
 
            elif _forks:

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'Cannot delete `%s` it still contains attached forks' %

			




	
	
	
		
 
                    (repo.repo_name,)

			




	
	
	
		
 
                )

			




	
	
	
		
 

			




	
	
		
 
@@ -1759,28 +1751,28 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
    @HasPermissionAnyDecorator('hg.admin')

			




	
	
	
		
 
    def get_repo_groups(self):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Returns all repository groups

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        return [

			




	
	
	
		
 
            repo_group.get_api_data()

			




	
	
	
		
 
            for repo_group in RepoGroup.query()

			




	
	
	
		
 
        ]

			




	
	
	
		
 

			




	
	
	
		
 
    @HasPermissionAnyDecorator('hg.admin')

			




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

			




	
	
	
		
 
                          owner=Optional(OAttr('apiuser')),

			




	
	
	
		
 
                          parent=Optional(None),

			




	
	
	
		
 
                          copy_permissions=Optional(False)):

			




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

			




	
	
	
		
 
                          owner=None,

			




	
	
	
		
 
                          parent=None,

			




	
	
	
		
 
                          copy_permissions=False):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Creates a repository group. This command can be executed only using

			




	
	
	
		
 
        api_key belonging to user with admin rights.

			




	
	
	
		
 

			




	
	
	
		
 
        :param group_name:

			




	
	
	
		
 
        :type group_name:

			




	
	
	
		
 
        :param description:

			




	
	
	
		
 
        :type description:

			




	
	
	
		
 
        :param owner:

			




	
	
	
		
 
        :type owner:

			




	
	
	
		
 
        :param parent:

			




	
	
	
		
 
        :type parent:

			




	
	
		
 
@@ -1799,55 +1791,54 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        ERROR OUTPUT::

			




	
	
	
		
 

			




	
	
	
		
 
          id : <id_given_in_input>

			




	
	
	
		
 
          result : null

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            failed to create repo group `<repogroupid>`

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        if RepoGroup.get_by_group_name(group_name):

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        if isinstance(owner, Optional):

			




	
	
	
		
 
        if owner is None:

			




	
	
	
		
 
            owner = request.authuser.user_id

			




	
	
	
		
 
        group_description = Optional.extract(description)

			




	
	
	
		
 
        parent_group = Optional.extract(parent)

			




	
	
	
		
 
        if not isinstance(parent, Optional):

			




	
	
	
		
 
            parent_group = get_repo_group_or_error(parent_group)

			




	
	
	
		
 
        group_description = description

			




	
	
	
		
 
        parent_group = None

			




	
	
	
		
 
        if parent is not None:

			




	
	
	
		
 
            parent_group = get_repo_group_or_error(parent)

			




	
	
	
		
 

			




	
	
	
		
 
        copy_permissions = Optional.extract(copy_permissions)

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            repo_group = RepoGroupModel().create(

			




	
	
	
		
 
                group_name=group_name,

			




	
	
	
		
 
                group_description=group_description,

			




	
	
	
		
 
                owner=owner,

			




	
	
	
		
 
                parent=parent_group,

			




	
	
	
		
 
                copy_permissions=copy_permissions

			




	
	
	
		
 
            )

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




	
	
	
		
 
                msg='created new repo group `%s`' % group_name,

			




	
	
	
		
 
                repo_group=repo_group.get_api_data()

			




	
	
	
		
 
            )

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    @HasPermissionAnyDecorator('hg.admin')

			




	
	
	
		
 
    def update_repo_group(self, repogroupid, group_name=Optional(''),

			




	
	
	
		
 
                          description=Optional(''),

			




	
	
	
		
 
                          owner=Optional(OAttr('apiuser')),

			




	
	
	
		
 
                          parent=Optional(None)):

			




	
	
	
		
 
    def update_repo_group(self, repogroupid, group_name=None,

			




	
	
	
		
 
                          description=None,

			




	
	
	
		
 
                          owner=None,

			




	
	
	
		
 
                          parent=None):

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 

			




	
	
	
		
 
        updates = {}

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            store_update(updates, group_name, 'group_name')

			




	
	
	
		
 
            store_update(updates, description, 'group_description')

			




	
	
	
		
 
            store_update(updates, owner, 'owner')

			




	
	
	
		
 
            store_update(updates, parent, 'parent_group')

			




	
	
	
		
 
            repo_group = RepoGroupModel().update(repo_group, updates)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




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

			




	
	
		
 
@@ -1893,25 +1884,25 @@ class ApiController(JSONRPCController):

			




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

			




	
	
	
		
 
                    (repo_group.group_id, repo_group.group_name),

			




	
	
	
		
 
                repo_group=None

			




	
	
	
		
 
            )

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




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

			




	
	
	
		
 
                               (repo_group.group_id, repo_group.group_name)

			




	
	
	
		
 
                               )

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def grant_user_permission_to_repo_group(self, repogroupid, userid,

			




	
	
	
		
 
                                            perm, apply_to_children=Optional('none')):

			




	
	
	
		
 
                                            perm, apply_to_children='none'):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Grant permission for user on given repository group, or update existing

			




	
	
	
		
 
        one if found. This command can be executed only using api_key belonging

			




	
	
	
		
 
        to user with admin rights, or user who has admin right to given repository

			




	
	
	
		
 
        group.

			




	
	
	
		
 

			




	
	
	
		
 
        :param repogroupid: name or id of repository group

			




	
	
	
		
 
        :type repogroupid: str or int

			




	
	
	
		
 
        :param userid:

			




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

			




	
	
	
		
 
        :type perm: str

			




	
	
	
		
 
        :param apply_to_children: 'none', 'repos', 'groups', 'all'

			




	
	
		
 
@@ -1935,48 +1926,47 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 

			




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

			




	
	
	
		
 
            if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        user = get_user_or_error(userid)

			




	
	
	
		
 
        perm = get_perm_or_error(perm, prefix='group.')

			




	
	
	
		
 
        apply_to_children = Optional.extract(apply_to_children)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            RepoGroupModel().add_permission(repo_group=repo_group,

			




	
	
	
		
 
                                            obj=user,

			




	
	
	
		
 
                                            obj_type="user",

			




	
	
	
		
 
                                            perm=perm,

			




	
	
	
		
 
                                            recursive=apply_to_children)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




	
	
	
		
 
                msg='Granted perm: `%s` (recursive:%s) for user: `%s` in repo group: `%s`' % (

			




	
	
	
		
 
                    perm.permission_name, apply_to_children, user.username, repo_group.name

			




	
	
	
		
 
                ),

			




	
	
	
		
 
                success=True

			




	
	
	
		
 
            )

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise JSONRPCError(

			




	
	
	
		
 
                'failed to edit permission for user: `%s` in repo group: `%s`' % (

			




	
	
	
		
 
                    userid, repo_group.name))

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def revoke_user_permission_from_repo_group(self, repogroupid, userid,

			




	
	
	
		
 
                                               apply_to_children=Optional('none')):

			




	
	
	
		
 
                                               apply_to_children='none'):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Revoke permission for user on given repository group. This command can

			




	
	
	
		
 
        be executed only using api_key belonging to user with admin rights, or

			




	
	
	
		
 
        user who has admin right to given repository group.

			




	
	
	
		
 

			




	
	
	
		
 
        :param repogroupid: name or id of repository group

			




	
	
	
		
 
        :type repogroupid: str or int

			




	
	
	
		
 
        :param userid:

			




	
	
	
		
 
        :type userid:

			




	
	
	
		
 
        :param apply_to_children: 'none', 'repos', 'groups', 'all'

			




	
	
	
		
 
        :type apply_to_children: str

			




	
	
	
		
 

			




	
	
		
 
@@ -1997,49 +1987,48 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
            "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 

			




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

			




	
	
	
		
 
            if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        user = get_user_or_error(userid)

			




	
	
	
		
 
        apply_to_children = Optional.extract(apply_to_children)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            RepoGroupModel().delete_permission(repo_group=repo_group,

			




	
	
	
		
 
                                               obj=user,

			




	
	
	
		
 
                                               obj_type="user",

			




	
	
	
		
 
                                               recursive=apply_to_children)

			




	
	
	
		
 

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




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

			




	
	
	
		
 
                    apply_to_children, user.username, repo_group.name

			




	
	
	
		
 
                ),

			




	
	
	
		
 
                success=True

			




	
	
	
		
 
            )

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise JSONRPCError(

			




	
	
	
		
 
                'failed to edit permission for user: `%s` in repo group: `%s`' % (

			




	
	
	
		
 
                    userid, repo_group.name))

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def grant_user_group_permission_to_repo_group(

			




	
	
	
		
 
            self, repogroupid, usergroupid, perm,

			




	
	
	
		
 
            apply_to_children=Optional('none')):

			




	
	
	
		
 
            apply_to_children='none'):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Grant permission for user group on given repository group, or update

			




	
	
	
		
 
        existing one if found. This command can be executed only using

			




	
	
	
		
 
        api_key belonging to user with admin rights, or user who has admin

			




	
	
	
		
 
        right to given repository group.

			




	
	
	
		
 

			




	
	
	
		
 
        :param repogroupid: name or id of repository group

			




	
	
	
		
 
        :type repogroupid: str or int

			




	
	
	
		
 
        :param usergroupid: id of usergroup

			




	
	
	
		
 
        :type usergroupid: str or int

			




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

			




	
	
	
		
 
        :type perm: str

			




	
	
		
 
@@ -2068,26 +2057,24 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 
        perm = get_perm_or_error(perm, prefix='group.')

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




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

			




	
	
	
		
 
            if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'repository group `%s` does not exist' % (repogroupid,))

			




	
	
	
		
 

			




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

			




	
	
	
		
 
                raise JSONRPCError(

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        apply_to_children = Optional.extract(apply_to_children)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            RepoGroupModel().add_permission(repo_group=repo_group,

			




	
	
	
		
 
                                            obj=user_group,

			




	
	
	
		
 
                                            obj_type="user_group",

			




	
	
	
		
 
                                            perm=perm,

			




	
	
	
		
 
                                            recursive=apply_to_children)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




	
	
	
		
 
                msg='Granted perm: `%s` (recursive:%s) for user group: `%s` in repo group: `%s`' % (

			




	
	
	
		
 
                    perm.permission_name, apply_to_children,

			




	
	
	
		
 
                    user_group.users_group_name, repo_group.name

			




	
	
	
		
 
                ),

			




	
	
		
 
@@ -2096,25 +2083,25 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        except Exception:

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 
            raise JSONRPCError(

			




	
	
	
		
 
                'failed to edit permission for user group: `%s` in '

			




	
	
	
		
 
                'repo group: `%s`' % (

			




	
	
	
		
 
                    usergroupid, repo_group.name

			




	
	
	
		
 
                )

			




	
	
	
		
 
            )

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def revoke_user_group_permission_from_repo_group(

			




	
	
	
		
 
            self, repogroupid, usergroupid,

			




	
	
	
		
 
            apply_to_children=Optional('none')):

			




	
	
	
		
 
            apply_to_children='none'):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Revoke permission for user group on given repository. This command can be

			




	
	
	
		
 
        executed only using api_key belonging to user with admin rights, or

			




	
	
	
		
 
        user who has admin right to given repository group.

			




	
	
	
		
 

			




	
	
	
		
 
        :param repogroupid: name or id of repository group

			




	
	
	
		
 
        :type repogroupid: str or int

			




	
	
	
		
 
        :param usergroupid:

			




	
	
	
		
 
        :param apply_to_children: 'none', 'repos', 'groups', 'all'

			




	
	
	
		
 
        :type apply_to_children: str

			




	
	
	
		
 

			




	
	
	
		
 
        OUTPUT::

			




	
	
		
 
@@ -2138,26 +2125,24 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        repo_group = get_repo_group_or_error(repogroupid)

			




	
	
	
		
 
        user_group = get_user_group_or_error(usergroupid)

			




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

			




	
	
	
		
 
            if not HasRepoGroupPermissionLevel('admin')(repo_group.group_name):

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'repository group `%s` does not exist' % (repogroupid,))

			




	
	
	
		
 

			




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

			




	
	
	
		
 
                raise JSONRPCError(

			




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

			




	
	
	
		
 

			




	
	
	
		
 
        apply_to_children = Optional.extract(apply_to_children)

			




	
	
	
		
 

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            RepoGroupModel().delete_permission(repo_group=repo_group,

			




	
	
	
		
 
                                               obj=user_group,

			




	
	
	
		
 
                                               obj_type="user_group",

			




	
	
	
		
 
                                               recursive=apply_to_children)

			




	
	
	
		
 
            Session().commit()

			




	
	
	
		
 
            return dict(

			




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

			




	
	
	
		
 
                    apply_to_children, user_group.users_group_name, repo_group.name

			




	
	
	
		
 
                ),

			




	
	
	
		
 
                success=True

			




	
	
	
		
 
            )

			




	
	
		
 
@@ -2173,56 +2158,56 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Get given gist by id

			




	
	
	
		
 

			




	
	
	
		
 
        :param gistid: id of private or public gist

			




	
	
	
		
 
        :type gistid: str

			




	
	
	
		
 
        """

			




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

			




	
	
	
		
 
        return gist.get_api_data()

			




	
	
	
		
 

			




	
	
	
		
 
    def get_gists(self, userid=Optional(OAttr('apiuser'))):

			




	
	
	
		
 
    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 not isinstance(userid, Optional) and userid != request.authuser.user_id:

			




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

			




	
	
	
		
 
                raise JSONRPCError(

			




	
	
	
		
 
                    'userid is not the same as your user'

			




	
	
	
		
 
                )

			




	
	
	
		
 

			




	
	
	
		
 
        if isinstance(userid, Optional):

			




	
	
	
		
 
        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 Gist().query()

			




	
	
	
		
 
                .filter_by(is_expired=False)

			




	
	
	
		
 
                .filter(Gist.owner_id == user_id)

			




	
	
	
		
 
                .order_by(Gist.created_on.desc())

			




	
	
	
		
 
        ]

			




	
	
	
		
 

			




	
	
	
		
 
    def create_gist(self, files, owner=Optional(OAttr('apiuser')),

			




	
	
	
		
 
                    gist_type=Optional(Gist.GIST_PUBLIC), lifetime=Optional(-1),

			




	
	
	
		
 
                    description=Optional('')):

			




	
	
	
		
 
    def create_gist(self, files, owner=None,

			




	
	
	
		
 
                    gist_type=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)

			




	
	
		
 
@@ -2241,53 +2226,44 @@ class ApiController(JSONRPCController):

			




	
	
	
		
 
          error :  null

			




	
	
	
		
 

			




	
	
	
		
 
        ERROR OUTPUT::

			




	
	
	
		
 

			




	
	
	
		
 
          id : <id_given_in_input>

			




	
	
	
		
 
          result : null

			




	
	
	
		
 
          error :  {

			




	
	
	
		
 
            "failed to create gist"

			




	
	
	
		
 
          }

			




	
	
	
		
 

			




	
	
	
		
 
        """

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            if isinstance(owner, Optional):

			




	
	
	
		
 
            if owner is None:

			




	
	
	
		
 
                owner = request.authuser.user_id

			




	
	
	
		
 

			




	
	
	
		
 
            owner = get_user_or_error(owner)

			




	
	
	
		
 
            description = Optional.extract(description)

			




	
	
	
		
 
            gist_type = Optional.extract(gist_type)

			




	
	
	
		
 
            lifetime = Optional.extract(lifetime)

			




	
	
	
		
 

			




	
	
	
		
 
            gist = GistModel().create(description=description,

			




	
	
	
		
 
                                      owner=owner,

			




	
	
	
		
 
                                      ip_addr=request.ip_addr,

			




	
	
	
		
 
                                      gist_mapping=files,

			




	
	
	
		
 
                                      gist_type=gist_type,

			




	
	
	
		
 
                                      lifetime=lifetime)

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    # def update_gist(self, gistid, files, owner=Optional(OAttr('apiuser')),

			




	
	
	
		
 
    #                 gist_type=Optional(Gist.GIST_PUBLIC),

			




	
	
	
		
 
    #                 gist_lifetime=Optional(-1), gist_description=Optional('')):

			




	
	
	
		
 
    #     gist = get_gist_or_error(gistid)

			




	
	
	
		
 
    #     updates = {}

			




	
	
	
		
 

			




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

			




	
	
		
 
@@ -2333,35 +2309,34 @@ class ApiController(JSONRPCController):

			




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

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def get_changeset(self, repoid, raw_id, with_reviews=Optional(False)):

			




	
	
	
		
 
    def get_changeset(self, repoid, raw_id, with_reviews=False):

			




	
	
	
		
 
        repo = get_repo_or_error(repoid)

			




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

			




	
	
	
		
 
            raise JSONRPCError('Access denied to repo %s' % repo.repo_name)

			




	
	
	
		
 
        changeset = repo.get_changeset(raw_id)

			




	
	
	
		
 
        if isinstance(changeset, EmptyChangeset):

			




	
	
	
		
 
            raise JSONRPCError('Changeset %s does not exist' % raw_id)

			




	
	
	
		
 

			




	
	
	
		
 
        info = dict(changeset.as_dict())

			




	
	
	
		
 

			




	
	
	
		
 
        with_reviews = Optional.extract(with_reviews)

			




	
	
	
		
 
        if with_reviews:

			




	
	
	
		
 
            reviews = ChangesetStatusModel().get_statuses(

			




	
	
	
		
 
                                repo.repo_name, raw_id)

			




	
	
	
		
 
            info["reviews"] = reviews

			




	
	
	
		
 

			




	
	
	
		
 
        return info

			




	
	
	
		
 

			




	
	
	
		
 
    # permission check inside

			




	
	
	
		
 
    def get_pullrequest(self, pullrequest_id):

			




	
	
	
		
 
        """

			




	
	
	
		
 
        Get given pull request by id

			




	
	
	
		
 
        """

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        Server instance: clio-17285
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.
            – Support