api: modernize code for python3

rhodecode/api/__init__.py

@@ -1,576 +1,574 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import itertools
 import logging
 import sys
 import fnmatch
 
 import decorator
 import typing
 import venusian
 from collections import OrderedDict
 
 from pyramid.exceptions import ConfigurationError
 from pyramid.renderers import render
 from pyramid.response import Response
 from pyramid.httpexceptions import HTTPNotFound
 
 from rhodecode.api.exc import (
     JSONRPCBaseError, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
 from rhodecode.apps._base import TemplateArgs
 from rhodecode.lib.auth import AuthUser
 from rhodecode.lib.base import get_ip_addr, attach_context_attributes
 from rhodecode.lib.exc_tracking import store_exception
 from rhodecode.lib import ext_json
 from rhodecode.lib.utils2 import safe_str
 from rhodecode.lib.plugins.utils import get_plugin_settings
 from rhodecode.model.db import User, UserApiKeys
 
 log = logging.getLogger(__name__)
 
 DEFAULT_RENDERER = 'jsonrpc_renderer'
 DEFAULT_URL = '/_admin/apiv2'
 
 
 def find_methods(jsonrpc_methods, pattern):
     matches = OrderedDict()
     if not isinstance(pattern, (list, tuple)):
         pattern = [pattern]
 
     for single_pattern in pattern:
         for method_name, method in jsonrpc_methods.items():
             if fnmatch.fnmatch(method_name, single_pattern):
                 matches[method_name] = method
     return matches
 
 
 class ExtJsonRenderer(object):
     """
     Custom renderer that makes use of our ext_json lib
 
     """
 
     def __init__(self):
         self.serializer = ext_json.formatted_json
 
     def __call__(self, info):
         """ Returns a plain JSON-encoded string with content-type
         ``application/json``. The content-type may be overridden by
         setting ``request.response.content_type``."""
 
         def _render(value, system):
             request = system.get('request')
             if request is not None:
                 response = request.response
                 ct = response.content_type
                 if ct == response.default_content_type:
                     response.content_type = 'application/json'
 
             return self.serializer(value)
 
         return _render
 
 
 def jsonrpc_response(request, result):
     rpc_id = getattr(request, 'rpc_id', None)
 
     ret_value = ''
     if rpc_id:
         ret_value = {'id': rpc_id, 'result': result, 'error': None}
 
         # fetch deprecation warnings, and store it inside results
         deprecation = getattr(request, 'rpc_deprecation', None)
         if deprecation:
             ret_value['DEPRECATION_WARNING'] = deprecation
 
     raw_body = render(DEFAULT_RENDERER, ret_value, request=request)
     content_type = 'application/json'
     content_type_header = 'Content-Type'
     headers = {
         content_type_header: content_type
     }
     return Response(
         body=raw_body,
         content_type=content_type,
         headerlist=[(k, v) for k, v in headers.items()]
     )
 
 
-def jsonrpc_error(request, message, retid=None, code: typing.Optional[int] = None, headers: typing.Optional[dict] = None):
+def jsonrpc_error(request, message, retid=None, code: int | None = None, headers: dict | None = None):
     """
     Generate a Response object with a JSON-RPC error body
     """
     headers = headers or {}
     content_type = 'application/json'
     content_type_header = 'Content-Type'
     if content_type_header not in headers:
         headers[content_type_header] = content_type
 
     err_dict = {'id': retid, 'result': None, 'error': message}
     raw_body = render(DEFAULT_RENDERER, err_dict, request=request)
 
     return Response(
         body=raw_body,
         status=code,
         content_type=content_type,
         headerlist=[(k, v) for k, v in headers.items()]
     )
 
 
 def exception_view(exc, request):
     rpc_id = getattr(request, 'rpc_id', None)
 
     if isinstance(exc, JSONRPCError):
         fault_message = safe_str(exc.message)
         log.debug('json-rpc error rpc_id:%s "%s"', rpc_id, fault_message)
     elif isinstance(exc, JSONRPCValidationError):
         colander_exc = exc.colander_exception
         # TODO(marcink): think maybe of nicer way to serialize errors ?
         fault_message = colander_exc.asdict()
         log.debug('json-rpc colander error rpc_id:%s "%s"', rpc_id, fault_message)
     elif isinstance(exc, JSONRPCForbidden):
         fault_message = 'Access was denied to this resource.'
         log.warning('json-rpc forbidden call rpc_id:%s "%s"', rpc_id, fault_message)
     elif isinstance(exc, HTTPNotFound):
         method = request.rpc_method
         log.debug('json-rpc method `%s` not found in list of '
                   'api calls: %s, rpc_id:%s',
                   method, list(request.registry.jsonrpc_methods.keys()), rpc_id)
 
         similar = 'none'
         try:
             similar_paterns = [f'*{x}*' for x in method.split('_')]
             similar_found = find_methods(
                 request.registry.jsonrpc_methods, similar_paterns)
             similar = ', '.join(similar_found.keys()) or similar
         except Exception:
             # make the whole above block safe
             pass
 
         fault_message = "No such method: {}. Similar methods: {}".format(
             method, similar)
     else:
         fault_message = 'undefined error'
         exc_info = exc.exc_info()
         store_exception(id(exc_info), exc_info, prefix='rhodecode-api')
 
         statsd = request.registry.statsd
         if statsd:
-            exc_type = "{}.{}".format(exc.__class__.__module__, exc.__class__.__name__)
+            exc_type = f"{exc.__class__.__module__}.{exc.__class__.__name__}"
             statsd.incr('rhodecode_exception_total',
-                        tags=["exc_source:api", "type:{}".format(exc_type)])
+                        tags=["exc_source:api", f"type:{exc_type}"])
 
     return jsonrpc_error(request, fault_message, rpc_id)
 
 
 def request_view(request):
     """
     Main request handling method. It handles all logic to call a specific
     exposed method
     """
     # cython compatible inspect
     from rhodecode.config.patches import inspect_getargspec
     inspect = inspect_getargspec()
 
     # check if we can find this session using api_key, get_by_auth_token
     # search not expired tokens only
     try:
         api_user = User.get_by_auth_token(request.rpc_api_key)
 
         if api_user is None:
             return jsonrpc_error(
                 request, retid=request.rpc_id, message='Invalid API KEY')
 
         if not api_user.active:
             return jsonrpc_error(
                 request, retid=request.rpc_id,
                 message='Request from this user not allowed')
 
         # check if we are allowed to use this IP
         auth_u = AuthUser(
             api_user.user_id, request.rpc_api_key, ip_addr=request.rpc_ip_addr)
         if not auth_u.ip_allowed:
             return jsonrpc_error(
                 request, retid=request.rpc_id,
-                message='Request from IP:%s not allowed' % (
-                    request.rpc_ip_addr,))
+                message='Request from IP:{} not allowed'.format(
+                    request.rpc_ip_addr))
         else:
             log.info('Access for IP:%s allowed', request.rpc_ip_addr)
 
         # register our auth-user
         request.rpc_user = auth_u
         request.environ['rc_auth_user_id'] = str(auth_u.user_id)
 
         # now check if token is valid for API
         auth_token = request.rpc_api_key
         token_match = api_user.authenticate_by_token(
             auth_token, roles=[UserApiKeys.ROLE_API])
         invalid_token = not token_match
 
         log.debug('Checking if API KEY is valid with proper role')
         if invalid_token:
             return jsonrpc_error(
                 request, retid=request.rpc_id,
                 message='API KEY invalid or, has bad role for an API call')
 
     except Exception:
         log.exception('Error on API AUTH')
         return jsonrpc_error(
             request, retid=request.rpc_id, message='Invalid API KEY')
 
     method = request.rpc_method
     func = request.registry.jsonrpc_methods[method]
 
     # now that we have a method, add request._req_params to
     # self.kargs and dispatch control to WGIController
 
     argspec = inspect.getargspec(func)
     arglist = argspec[0]
     defs = argspec[3] or []
     defaults = [type(a) for a in defs]
     default_empty = type(NotImplemented)
 
     # kw arguments required by this method
     func_kwargs = dict(itertools.zip_longest(
         reversed(arglist), reversed(defaults), fillvalue=default_empty))
 
     # This attribute will need to be first param of a method that uses
     # api_key, which is translated to instance of user at that name
     user_var = 'apiuser'
     request_var = 'request'
 
     for arg in [user_var, request_var]:
         if arg not in arglist:
             return jsonrpc_error(
                 request,
                 retid=request.rpc_id,
                 message='This method [%s] does not support '
                         'required parameter `%s`' % (func.__name__, arg))
 
     # get our arglist and check if we provided them as args
     for arg, default in func_kwargs.items():
         if arg in [user_var, request_var]:
             # user_var and request_var are pre-hardcoded parameters and we
             # don't need to do any translation
             continue
 
         # skip the required param check if it's default value is
         # NotImplementedType (default_empty)
         if default == default_empty and arg not in request.rpc_params:
             return jsonrpc_error(
                 request,
                 retid=request.rpc_id,
                 message=('Missing non optional `%s` arg in JSON DATA' % arg)
             )
 
     # sanitize extra passed arguments
     for k in list(request.rpc_params.keys()):
         if k not in func_kwargs:
             del request.rpc_params[k]
 
     call_params = request.rpc_params
     call_params.update({
         'request': request,
         'apiuser': auth_u
     })
 
     # register some common functions for usage
     attach_context_attributes(TemplateArgs(), request, request.rpc_user.user_id)
 
     statsd = request.registry.statsd
 
     try:
         ret_value = func(**call_params)
         resp = jsonrpc_response(request, ret_value)
         if statsd:
             statsd.incr('rhodecode_api_call_success_total')
         return resp
     except JSONRPCBaseError:
         raise
     except Exception:
         log.exception('Unhandled exception occurred on api call: %s', func)
         exc_info = sys.exc_info()
         exc_id, exc_type_name = store_exception(
             id(exc_info), exc_info, prefix='rhodecode-api')
         error_headers = {
             'RhodeCode-Exception-Id': str(exc_id),
             'RhodeCode-Exception-Type': str(exc_type_name)
         }
         err_resp = jsonrpc_error(
             request, retid=request.rpc_id, message='Internal server error',
             headers=error_headers)
         if statsd:
             statsd.incr('rhodecode_api_call_fail_total')
         return err_resp
 
 
 def setup_request(request):
     """
     Parse a JSON-RPC request body. It's used inside the predicates method
     to validate and bootstrap requests for usage in rpc calls.
 
     We need to raise JSONRPCError here if we want to return some errors back to
     user.
     """
 
     log.debug('Executing setup request: %r', request)
     request.rpc_ip_addr = get_ip_addr(request.environ)
     # TODO(marcink): deprecate GET at some point
     if request.method not in ['POST', 'GET']:
         log.debug('unsupported request method "%s"', request.method)
         raise JSONRPCError(
             'unsupported request method "%s". Please use POST' % request.method)
 
     if 'CONTENT_LENGTH' not in request.environ:
         log.debug("No Content-Length")
         raise JSONRPCError("Empty body, No Content-Length in request")
 
     else:
         length = request.environ['CONTENT_LENGTH']
         log.debug('Content-Length: %s', length)
 
         if length == 0:
             log.debug("Content-Length is 0")
             raise JSONRPCError("Content-Length is 0")
 
     raw_body = request.body
     log.debug("Loading JSON body now")
     try:
         json_body = ext_json.json.loads(raw_body)
     except ValueError as e:
         # catch JSON errors Here
-        raise JSONRPCError("JSON parse error ERR:%s RAW:%r" % (e, raw_body))
+        raise JSONRPCError("JSON parse error ERR:{} RAW:{!r}".format(e, raw_body))
 
     request.rpc_id = json_body.get('id')
     request.rpc_method = json_body.get('method')
 
     # check required base parameters
     try:
         api_key = json_body.get('api_key')
         if not api_key:
             api_key = json_body.get('auth_token')
 
         if not api_key:
             raise KeyError('api_key or auth_token')
 
         # TODO(marcink): support passing in token in request header
 
         request.rpc_api_key = api_key
         request.rpc_id = json_body['id']
         request.rpc_method = json_body['method']
         request.rpc_params = json_body['args'] \
             if isinstance(json_body['args'], dict) else {}
 
         log.debug('method: %s, params: %.10240r', request.rpc_method, request.rpc_params)
     except KeyError as e:
         raise JSONRPCError(f'Incorrect JSON data. Missing {e}')
 
     log.debug('setup complete, now handling method:%s rpcid:%s',
               request.rpc_method, request.rpc_id, )
 
 
 class RoutePredicate(object):
     def __init__(self, val, config):
         self.val = val
 
     def text(self):
         return f'jsonrpc route = {self.val}'
 
     phash = text
 
     def __call__(self, info, request):
         if self.val:
             # potentially setup and bootstrap our call
             setup_request(request)
 
             # Always return True so that even if it isn't a valid RPC it
             # will fall through to the underlaying handlers like notfound_view
             return True
 
 
 class NotFoundPredicate(object):
     def __init__(self, val, config):
         self.val = val
         self.methods = config.registry.jsonrpc_methods
 
     def text(self):
         return f'jsonrpc method not found = {self.val}'
 
     phash = text
 
     def __call__(self, info, request):
         return hasattr(request, 'rpc_method')
 
 
 class MethodPredicate(object):
     def __init__(self, val, config):
         self.method = val
 
     def text(self):
         return f'jsonrpc method = {self.method}'
 
     phash = text
 
     def __call__(self, context, request):
         # we need to explicitly return False here, so pyramid doesn't try to
         # execute our view directly. We need our main handler to execute things
         return getattr(request, 'rpc_method') == self.method
 
 
 def add_jsonrpc_method(config, view, **kwargs):
     # pop the method name
     method = kwargs.pop('method', None)
 
     if method is None:
         raise ConfigurationError(
             'Cannot register a JSON-RPC method without specifying the "method"')
 
     # we define custom predicate, to enable to detect conflicting methods,
     # those predicates are kind of "translation" from the decorator variables
     # to internal predicates names
 
     kwargs['jsonrpc_method'] = method
 
     # register our view into global view store for validation
     config.registry.jsonrpc_methods[method] = view
 
     # we're using our main request_view handler, here, so each method
     # has a unified handler for itself
     config.add_view(request_view, route_name='apiv2', **kwargs)
 
 
 class jsonrpc_method(object):
     """
     decorator that works similar to @add_view_config decorator,
     but tailored for our JSON RPC
     """
 
     venusian = venusian  # for testing injection
 
     def __init__(self, method=None, **kwargs):
         self.method = method
         self.kwargs = kwargs
 
     def __call__(self, wrapped):
         kwargs = self.kwargs.copy()
         kwargs['method'] = self.method or wrapped.__name__
         depth = kwargs.pop('_depth', 0)
 
         def callback(context, name, ob):
             config = context.config.with_package(info.module)
             config.add_jsonrpc_method(view=ob, **kwargs)
 
         info = venusian.attach(wrapped, callback, category='pyramid',
                                depth=depth + 1)
         if info.scope == 'class':
             # ensure that attr is set if decorating a class method
             kwargs.setdefault('attr', wrapped.__name__)
 
         kwargs['_info'] = info.codeinfo  # fbo action_method
         return wrapped
 
 
 class jsonrpc_deprecated_method(object):
     """
     Marks method as deprecated, adds log.warning, and inject special key to
     the request variable to mark method as deprecated.
     Also injects special docstring that extract_docs will catch to mark
     method as deprecated.
 
     :param use_method: specify which method should be used instead of
         the decorated one
 
     Use like::
 
         @jsonrpc_method()
         @jsonrpc_deprecated_method(use_method='new_func', deprecated_at_version='3.0.0')
         def old_func(request, apiuser, arg1, arg2):
             ...
     """
 
     def __init__(self, use_method, deprecated_at_version):
         self.use_method = use_method
         self.deprecated_at_version = deprecated_at_version
         self.deprecated_msg = ''
 
     def __call__(self, func):
         self.deprecated_msg = 'Please use method `{method}` instead.'.format(
             method=self.use_method)
 
         docstring = """\n
         .. deprecated:: {version}
 
            {deprecation_message}
 
         {original_docstring}
         """
         func.__doc__ = docstring.format(
             version=self.deprecated_at_version,
             deprecation_message=self.deprecated_msg,
             original_docstring=func.__doc__)
         return decorator.decorator(self.__wrapper, func)
 
     def __wrapper(self, func, *fargs, **fkwargs):
         log.warning('DEPRECATED API CALL on function %s, please '
                     'use `%s` instead', func, self.use_method)
         # alter function docstring to mark as deprecated, this is picked up
         # via fabric file that generates API DOC.
         result = func(*fargs, **fkwargs)
 
         request = fargs[0]
         request.rpc_deprecation = 'DEPRECATED METHOD ' + self.deprecated_msg
         return result
 
 
 def add_api_methods(config):
     from rhodecode.api.views import (
         deprecated_api, gist_api, pull_request_api, repo_api, repo_group_api,
         server_api, search_api, testing_api, user_api, user_group_api)
 
     config.scan('rhodecode.api.views')
 
 
 def includeme(config):
     plugin_module = 'rhodecode.api'
     plugin_settings = get_plugin_settings(
         plugin_module, config.registry.settings)
 
     if not hasattr(config.registry, 'jsonrpc_methods'):
         config.registry.jsonrpc_methods = OrderedDict()
 
     # match filter by given method only
     config.add_view_predicate('jsonrpc_method', MethodPredicate)
     config.add_view_predicate('jsonrpc_method_not_found', NotFoundPredicate)
 
     config.add_renderer(DEFAULT_RENDERER, ExtJsonRenderer())
     config.add_directive('add_jsonrpc_method', add_jsonrpc_method)
 
     config.add_route_predicate(
         'jsonrpc_call', RoutePredicate)
 
     config.add_route(
         'apiv2', plugin_settings.get('url', DEFAULT_URL), jsonrpc_call=True)
 
     # register some exception handling view
     config.add_view(exception_view, context=JSONRPCBaseError)
     config.add_notfound_view(exception_view, jsonrpc_method_not_found=True)
 
     add_api_methods(config)

rhodecode/api/exc.py

@@ -1,42 +1,40 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 class JSONRPCBaseError(Exception):
     def __init__(self, message='', *args):
         self.message = message
-        super(JSONRPCBaseError, self).__init__(message, *args)
+        super().__init__(message, *args)
 
 
 class JSONRPCError(JSONRPCBaseError):
     pass
 
 
 class JSONRPCValidationError(JSONRPCBaseError):
 
     def __init__(self, *args, **kwargs):
         self.colander_exception = kwargs.pop('colander_exc')
-        super(JSONRPCValidationError, self).__init__(
+        super().__init__(
             message=self.colander_exception, *args)
 
 
 class JSONRPCForbidden(JSONRPCBaseError):
     pass
 

rhodecode/api/utils.py

@@ -1,458 +1,456 b''
-
-
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 JSON RPC utils
 """
 
 import collections
 import logging
 
 from rhodecode.api.exc import JSONRPCError
 from rhodecode.lib.auth import (
     HasPermissionAnyApi, HasRepoPermissionAnyApi, HasRepoGroupPermissionAnyApi)
 from rhodecode.lib.str_utils import safe_str
 from rhodecode.lib.vcs.exceptions import RepositoryError
 from rhodecode.lib.view_utils import get_commit_from_ref_name
 from rhodecode.lib.utils2 import str2bool
 
 log = logging.getLogger(__name__)
 
 
 class OAttr(object):
     """
     Special Option that defines other attribute, and can default to them
 
     Example::
 
         def test(apiuser, userid=Optional(OAttr('apiuser')):
             user = Optional.extract(userid, evaluate_locals=local())
             #if we pass in userid, we get it, else it will default to apiuser
             #attribute
     """
 
     def __init__(self, attr_name):
         self.attr_name = attr_name
 
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
 
     def __call__(self):
         return self
 
 
 class Optional(object):
     """
     Defines an optional parameter::
 
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
 
     is equivalent of::
 
         param = Optional.extract(param)
 
     """
 
     def __init__(self, type_):
         self.type_ = type_
 
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
 
     def __call__(self):
         return self.getval()
 
     def getval(self, evaluate_locals=None):
         """
         returns value from this Optional instance
         """
         if isinstance(self.type_, OAttr):
             param_name = self.type_.attr_name
             if evaluate_locals:
                 return evaluate_locals[param_name]
             # use params name
             return param_name
         return self.type_
 
     @classmethod
     def extract(cls, val, evaluate_locals=None, binary=None):
         """
         Extracts value from Optional() instance
 
         :param val:
         :return: original value if it's not Optional instance else
             value of instance
         """
         if isinstance(val, cls):
             val = val.getval(evaluate_locals)
 
         if binary:
             val = str2bool(val)
 
         return val
 
 
 def parse_args(cli_args, key_prefix=''):
     from rhodecode.lib.utils2 import (escape_split)
     kwargs = collections.defaultdict(dict)
     for el in escape_split(cli_args, ','):
         kv = escape_split(el, '=', 1)
         if len(kv) == 2:
             k, v = kv
             kwargs[key_prefix + k] = v
     return kwargs
 
 
 def get_origin(obj):
     """
     Get origin of permission from object.
 
     :param obj:
     """
     origin = 'permission'
 
     if getattr(obj, 'owner_row', '') and getattr(obj, 'admin_row', ''):
         # admin and owner case, maybe we should use dual string ?
         origin = 'owner'
     elif getattr(obj, 'owner_row', ''):
         origin = 'owner'
     elif getattr(obj, 'admin_row', ''):
         origin = 'super-admin'
     return origin
 
 
 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
     """
     if not isinstance(attr, Optional):
         updates[name] = attr
 
 
 def has_superadmin_permission(apiuser):
     """
     Return True if apiuser is admin or return False
 
     :param apiuser:
     """
     if HasPermissionAnyApi('hg.admin')(user=apiuser):
         return True
     return False
 
 
 def validate_repo_permissions(apiuser, repoid, repo, perms):
     """
     Raise JsonRPCError if apiuser is not authorized or return True
 
     :param apiuser:
     :param repoid:
     :param repo:
     :param perms:
     """
     if not HasRepoPermissionAnyApi(*perms)(
             user=apiuser, repo_name=repo.repo_name):
         raise JSONRPCError('repository `%s` does not exist' % repoid)
 
     return True
 
 
 def validate_repo_group_permissions(apiuser, repogroupid, repo_group, perms):
     """
     Raise JsonRPCError if apiuser is not authorized or return True
 
     :param apiuser:
     :param repogroupid: just the id of repository group
     :param repo_group: instance of repo_group
     :param perms:
     """
     if not HasRepoGroupPermissionAnyApi(*perms)(
             user=apiuser, group_name=repo_group.group_name):
         raise JSONRPCError(
             'repository group `%s` does not exist' % repogroupid)
 
     return True
 
 
 def validate_set_owner_permissions(apiuser, owner):
     if isinstance(owner, Optional):
         owner = get_user_or_error(apiuser.user_id)
     else:
         if has_superadmin_permission(apiuser):
             owner = get_user_or_error(owner)
         else:
             # forbid setting owner for non-admins
             raise JSONRPCError(
                 'Only RhodeCode super-admin can specify `owner` param')
     return owner
 
 
 def get_user_or_error(userid):
     """
     Get user by id or name or return JsonRPCError if not found
 
     :param userid:
     """
     from rhodecode.model.user import UserModel
     user_model = UserModel()
 
     if isinstance(userid, int):
         try:
             user = user_model.get_user(userid)
         except ValueError:
             user = None
     else:
         user = user_model.get_by_username(userid)
 
     if user is None:
         raise JSONRPCError(
-            'user `%s` does not exist' % (userid,))
+            'user `{}` does not exist'.format(userid))
     return user
 
 
 def get_repo_or_error(repoid):
     """
     Get repo by id or name or return JsonRPCError if not found
 
     :param repoid:
     """
     from rhodecode.model.repo import RepoModel
     repo_model = RepoModel()
 
     if isinstance(repoid, int):
         try:
             repo = repo_model.get_repo(repoid)
         except ValueError:
             repo = None
     else:
         repo = repo_model.get_by_repo_name(repoid)
 
     if repo is None:
         raise JSONRPCError(
-            'repository `%s` does not exist' % (repoid,))
+            'repository `{}` does not exist'.format(repoid))
     return repo
 
 
 def get_repo_group_or_error(repogroupid):
     """
     Get repo group by id or name or return JsonRPCError if not found
 
     :param repogroupid:
     """
     from rhodecode.model.repo_group import RepoGroupModel
     repo_group_model = RepoGroupModel()
 
     if isinstance(repogroupid, int):
         try:
             repo_group = repo_group_model._get_repo_group(repogroupid)
         except ValueError:
             repo_group = None
     else:
         repo_group = repo_group_model.get_by_group_name(repogroupid)
 
     if repo_group is None:
         raise JSONRPCError(
-            'repository group `%s` does not exist' % (repogroupid,))
+            'repository group `{}` does not exist'.format(repogroupid))
     return repo_group
 
 
 def get_user_group_or_error(usergroupid):
     """
     Get user group by id or name or return JsonRPCError if not found
 
     :param usergroupid:
     """
     from rhodecode.model.user_group import UserGroupModel
     user_group_model = UserGroupModel()
 
     if isinstance(usergroupid, int):
         try:
             user_group = user_group_model.get_group(usergroupid)
         except ValueError:
             user_group = None
     else:
         user_group = user_group_model.get_by_name(usergroupid)
 
     if user_group is None:
         raise JSONRPCError(
-            'user group `%s` does not exist' % (usergroupid,))
+            'user group `{}` does not exist'.format(usergroupid))
     return user_group
 
 
 def get_perm_or_error(permid, prefix=None):
     """
     Get permission by id or name or return JsonRPCError if not found
 
     :param permid:
     """
     from rhodecode.model.permission import PermissionModel
 
     perm = PermissionModel.cls.get_by_key(permid)
     if perm is None:
-        msg = 'permission `{}` does not exist.'.format(permid)
+        msg = f'permission `{permid}` does not exist.'
         if prefix:
-            msg += ' Permission should start with prefix: `{}`'.format(prefix)
+            msg += f' Permission should start with prefix: `{prefix}`'
         raise JSONRPCError(msg)
 
     if prefix:
         if not perm.permission_name.startswith(prefix):
             raise JSONRPCError('permission `%s` is invalid, '
                                'should start with %s' % (permid, prefix))
     return perm
 
 
 def get_gist_or_error(gistid):
     """
     Get gist by id or gist_access_id or return JsonRPCError if not found
 
     :param gistid:
     """
     from rhodecode.model.gist import GistModel
 
     gist = GistModel.cls.get_by_access_id(gistid)
     if gist is None:
-        raise JSONRPCError('gist `%s` does not exist' % (gistid,))
+        raise JSONRPCError('gist `{}` does not exist'.format(gistid))
     return gist
 
 
 def get_pull_request_or_error(pullrequestid):
     """
     Get pull request by id or return JsonRPCError if not found
 
     :param pullrequestid:
     """
     from rhodecode.model.pull_request import PullRequestModel
 
     try:
         pull_request = PullRequestModel().get(int(pullrequestid))
     except ValueError:
         raise JSONRPCError('pullrequestid must be an integer')
     if not pull_request:
-        raise JSONRPCError('pull request `%s` does not exist' % (
-            pullrequestid,))
+        raise JSONRPCError('pull request `{}` does not exist'.format(
+            pullrequestid))
     return pull_request
 
 
 def build_commit_data(rhodecode_vcs_repo, commit, detail_level):
     commit2 = commit
     commit1 = commit.first_parent
 
     parsed_diff = []
     if detail_level == 'extended':
         for f_path in commit.added_paths:
             parsed_diff.append(_get_commit_dict(filename=f_path, op='A'))
         for f_path in commit.changed_paths:
             parsed_diff.append(_get_commit_dict(filename=f_path, op='M'))
         for f_path in commit.removed_paths:
             parsed_diff.append(_get_commit_dict(filename=f_path, op='D'))
 
     elif detail_level == 'full':
         from rhodecode.lib import diffs
 
         _diff = rhodecode_vcs_repo.get_diff(commit1, commit2,)
         diff_processor = diffs.DiffProcessor(_diff, diff_format='newdiff', show_full_diff=True)
 
         for dp in diff_processor.prepare():
             del dp['stats']['ops']
             _stats = dp['stats']
             parsed_diff.append(_get_commit_dict(
                 filename=dp['filename'], op=dp['operation'],
                 new_revision=dp['new_revision'],
                 old_revision=dp['old_revision'],
                 raw_diff=dp['raw_diff'], stats=_stats))
 
     return parsed_diff
 
 
 def get_commit_or_error(ref, repo):
     try:
         ref_type, _, ref_hash = ref.split(':')
     except ValueError:
         raise JSONRPCError(
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref))
     try:
         # TODO: dan: refactor this to use repo.scm_instance().get_commit()
         # once get_commit supports ref_types
         return get_commit_from_ref_name(repo, ref_hash)
     except RepositoryError:
-        raise JSONRPCError('Ref `{ref}` does not exist'.format(ref=ref))
+        raise JSONRPCError(f'Ref `{ref}` does not exist')
 
 
 def _get_ref_hash(repo, type_, name):
     vcs_repo = repo.scm_instance()
     if type_ in ['branch'] and vcs_repo.alias in ('hg', 'git'):
         return vcs_repo.branches[name]
     elif type_ in ['bookmark', 'book'] and vcs_repo.alias == 'hg':
         return vcs_repo.bookmarks[name]
     else:
         raise ValueError()
 
 
 def resolve_ref_or_error(ref, repo, allowed_ref_types=None):
     allowed_ref_types = allowed_ref_types or ['bookmark', 'book', 'tag', 'branch']
 
     def _parse_ref(type_, name, hash_=None):
         return type_, name, hash_
 
     try:
         ref_type, ref_name, ref_hash = _parse_ref(*ref.split(':'))
     except TypeError:
         raise JSONRPCError(
             'Ref `{ref}` given in a wrong format. Please check the API'
             ' documentation for more details'.format(ref=ref))
 
     if ref_type not in allowed_ref_types:
         raise JSONRPCError(
             'Ref `{ref}` type is not allowed. '
             'Only:{allowed_refs} are possible.'.format(
                 ref=ref, allowed_refs=allowed_ref_types))
 
     try:
         ref_hash = ref_hash or _get_ref_hash(repo, ref_type, ref_name)
     except (KeyError, ValueError):
         raise JSONRPCError(
             'The specified value:{type}:`{name}` does not exist, or is not allowed.'.format(
                 type=ref_type, name=ref_name))
 
     return ':'.join([ref_type, ref_name, ref_hash])
 
 
 def _get_commit_dict(
         filename, op, new_revision=None, old_revision=None,
         raw_diff=None, stats=None):
     if stats is None:
         stats = {
             "added": None,
             "binary": None,
             "deleted": None
         }
     return {
         "filename": safe_str(filename),
         "op": op,
 
         # extra details
         "new_revision": new_revision,
         "old_revision": old_revision,
 
         "raw_diff": raw_diff,
         "stats": stats
     }

rhodecode/api/views/__init__.py

@@ -1,19 +1,17 b''
-
-
 # Copyright (C) 2015-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/

rhodecode/api/views/deprecated_api.py

@@ -1,102 +1,100 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 """
 NOTE:
     Place for deprecated APIs here, if a call needs to be deprecated, please
     put it here, and point to a new version
 """
 import logging
 
 from rhodecode.api import jsonrpc_method, jsonrpc_deprecated_method
 from rhodecode.api.utils import Optional, OAttr
 
 
 log = logging.getLogger(__name__)
 
 
 # permission check inside
 @jsonrpc_method()
 @jsonrpc_deprecated_method(
     use_method='comment_commit', deprecated_at_version='3.4.0')
 def changeset_comment(request, apiuser, repoid, revision, message,
                       userid=Optional(OAttr('apiuser')),
                       status=Optional(None)):
     """
     Set a changeset comment, and optionally change the status of the
     changeset.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param revision: Specify the revision for which to set a comment.
     :type revision: str
     :param message: The comment text.
     :type message: str
     :param userid: Set the user name of the comment creator.
     :type userid: Optional(str or int)
     :param status: Set the comment status. The following are valid options:
         * not_reviewed
         * approved
         * rejected
         * under_review
     :type status: str
 
     Example error output:
 
     .. code-block:: javascript
 
         {
             "id" : <id_given_in_input>,
             "result" : {
                 "msg": "Commented on commit `<revision>` for repository `<repoid>`",
                 "status_change": null or <status>,
                 "success": true
             },
             "error" : null
         }
 
     """
     from .repo_api import comment_commit
 
     return comment_commit(request=request,
         apiuser=apiuser, repoid=repoid, commit_id=revision,
         message=message, userid=userid, status=status)
 
 
 @jsonrpc_method()
 @jsonrpc_deprecated_method(
     use_method='get_ip', deprecated_at_version='4.0.0')
 def show_ip(request, apiuser, userid=Optional(OAttr('apiuser'))):
     from .server_api import get_ip
     return get_ip(request=request, apiuser=apiuser, userid=userid)
 
 
 @jsonrpc_method()
 @jsonrpc_deprecated_method(
     use_method='get_user_locks', deprecated_at_version='4.0.0')
 def get_locks(request, apiuser, userid=Optional(OAttr('apiuser'))):
     from .user_api import get_user_locks
     return get_user_locks(request=request, apiuser=apiuser, userid=userid)
  No newline at end of file

rhodecode/api/views/gist_api.py

@@ -1,257 +1,255 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import logging
 import time
 
 from rhodecode.api import jsonrpc_method, JSONRPCError
 from rhodecode.api.exc import JSONRPCValidationError
 from rhodecode.api.utils import (
     Optional, OAttr, get_gist_or_error, get_user_or_error,
     has_superadmin_permission)
 from rhodecode.model.db import Session, or_
 from rhodecode.model.gist import Gist, GistModel
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_gist(request, apiuser, gistid, content=Optional(False)):
     """
     Get the specified gist, based on the gist ID.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param gistid: Set the id of the private or public gist
     :type gistid: str
     :param content: Return the gist content. Default is false.
     :type content: Optional(bool)
     """
 
     gist = get_gist_or_error(gistid)
     content = Optional.extract(content)
 
     if not has_superadmin_permission(apiuser):
         if gist.gist_owner != apiuser.user_id:
-            raise JSONRPCError('gist `%s` does not exist' % (gistid,))
+            raise JSONRPCError('gist `{}` does not exist'.format(gistid))
     data = gist.get_api_data()
 
     if content:
         from rhodecode.model.gist import GistModel
         rev, gist_files = GistModel().get_gist_files(gistid)
-        data['content'] = dict([(x.path, x.str_content) for x in gist_files])
+        data['content'] = {x.path: x.str_content for x in gist_files}
     return data
 
 
 @jsonrpc_method()
 def get_gists(request, apiuser, userid=Optional(OAttr('apiuser'))):
     """
     Get all gists for given user. If userid is empty returned gists
     are for user who called the api
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param userid: user to get gists for
     :type userid: Optional(str or int)
     """
 
     if not has_superadmin_permission(apiuser):
         # make sure normal user does not pass someone else userid,
         # he is not allowed to do that
         if not isinstance(userid, Optional) and userid != apiuser.user_id:
             raise JSONRPCError(
                 'userid is not the same as your user'
             )
 
     if isinstance(userid, Optional):
         user_id = apiuser.user_id
     else:
         user_id = get_user_or_error(userid).user_id
 
     gists = []
     _gists = Gist().query() \
         .filter(or_(
         Gist.gist_expires == -1, Gist.gist_expires >= time.time())) \
         .filter(Gist.gist_owner == user_id) \
         .order_by(Gist.created_on.desc())
     for gist in _gists:
         gists.append(gist.get_api_data())
     return gists
 
 
 @jsonrpc_method()
 def create_gist(
         request, apiuser, files, gistid=Optional(None),
         owner=Optional(OAttr('apiuser')),
         gist_type=Optional(Gist.GIST_PUBLIC), lifetime=Optional(-1),
         acl_level=Optional(Gist.ACL_LEVEL_PUBLIC),
         description=Optional('')):
     """
     Creates a new Gist.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param files: files to be added to the gist. The data structure has
         to match the following example::
 
           {'filename1': {'content':'...'}, 'filename2': {'content':'...'}}
 
     :type files: dict
     :param gistid: Set a custom id for the gist
     :type gistid: Optional(str)
     :param owner: Set the 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 acl_level: acl level for this gist, can be
         ``acl_public`` or ``acl_private`` If the value is set to
         ``acl_private`` only logged in users are able to access this gist.
         If not set it defaults to ``acl_public``.
     :type acl_level: Optional(str)
     :param description: gist description
     :type description: Optional(str)
 
     Example  output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "created new gist",
         "gist": {}
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to create gist"
       }
 
     """
     from rhodecode.model import validation_schema
     from rhodecode.model.validation_schema.schemas import gist_schema
 
     if isinstance(owner, Optional):
         owner = apiuser.user_id
 
     owner = get_user_or_error(owner)
 
     lifetime = Optional.extract(lifetime)
     schema = gist_schema.GistSchema().bind(
         # bind the given values if it's allowed, however the deferred
         # validator will still validate it according to other rules
         lifetime_options=[lifetime])
 
     try:
         nodes = gist_schema.nodes_to_sequence(
             files, colander_node=schema.get('nodes'))
 
         schema_data = schema.deserialize(dict(
             gistid=Optional.extract(gistid),
             description=Optional.extract(description),
             gist_type=Optional.extract(gist_type),
             lifetime=lifetime,
             gist_acl_level=Optional.extract(acl_level),
             nodes=nodes
         ))
 
         # convert to safer format with just KEYs so we sure no duplicates
         schema_data['nodes'] = gist_schema.sequence_to_nodes(
             schema_data['nodes'], colander_node=schema.get('nodes'))
 
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     try:
         gist = GistModel().create(
             owner=owner,
             gist_id=schema_data['gistid'],
             description=schema_data['description'],
             gist_mapping=schema_data['nodes'],
             gist_type=schema_data['gist_type'],
             lifetime=schema_data['lifetime'],
             gist_acl_level=schema_data['gist_acl_level'])
         Session().commit()
         return {
             'msg': 'created new gist',
             'gist': gist.get_api_data()
         }
     except Exception:
         log.exception('Error occurred during creation of gist')
         raise JSONRPCError('failed to create gist')
 
 
 @jsonrpc_method()
 def delete_gist(request, apiuser, gistid):
     """
     Deletes existing gist
 
     :param apiuser: filled automatically from apikey
     :type apiuser: AuthUser
     :param gistid: id of gist to delete
     :type gistid: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "deleted gist ID: <gist_id>",
         "gist": null
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to delete gist ID:<gist_id>"
       }
 
     """
 
     gist = get_gist_or_error(gistid)
     if not has_superadmin_permission(apiuser):
         if gist.gist_owner != apiuser.user_id:
-            raise JSONRPCError('gist `%s` does not exist' % (gistid,))
+            raise JSONRPCError('gist `{}` does not exist'.format(gistid))
 
     try:
         GistModel().delete(gist)
         Session().commit()
         return {
-            'msg': 'deleted gist ID:%s' % (gist.gist_access_id,),
+            'msg': 'deleted gist ID:{}'.format(gist.gist_access_id),
             'gist': None
         }
     except Exception:
         log.exception('Error occured during gist deletion')
         raise JSONRPCError('failed to delete gist ID:%s'
                            % (gist.gist_access_id,))
  No newline at end of file

rhodecode/api/views/pull_request_api.py

@@ -1,1113 +1,1111 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import logging
 
 from rhodecode.api import jsonrpc_method, JSONRPCError, JSONRPCValidationError
 from rhodecode.api.utils import (
     has_superadmin_permission, Optional, OAttr, get_repo_or_error,
     get_pull_request_or_error, get_commit_or_error, get_user_or_error,
     validate_repo_permissions, resolve_ref_or_error, validate_set_owner_permissions)
 from rhodecode.lib import channelstream
 from rhodecode.lib.auth import (HasRepoPermissionAnyApi)
 from rhodecode.lib.base import vcs_operation_context
 from rhodecode.lib.utils2 import str2bool
 from rhodecode.lib.vcs.backends.base import unicode_to_reference
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import (
     Session, ChangesetStatus, ChangesetComment, PullRequest, PullRequestReviewers)
 from rhodecode.model.pull_request import PullRequestModel, MergeCheck
 from rhodecode.model.settings import SettingsModel
 from rhodecode.model.validation_schema import Invalid
 from rhodecode.model.validation_schema.schemas.reviewer_schema import ReviewerListSchema
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_pull_request(request, apiuser, pullrequestid, repoid=Optional(None),
                      merge_state=Optional(False)):
     """
     Get a pull request based on the given ID.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Optional, repository name or repository ID from where
         the pull request was opened.
     :type repoid: str or int
     :param pullrequestid: ID of the requested pull request.
     :type pullrequestid: int
     :param merge_state: Optional calculate merge state for each repository.
         This could result in longer time to fetch the data
     :type merge_state: bool
 
     Example output:
 
     .. code-block:: bash
 
       "id": <id_given_in_input>,
       "result":
         {
             "pull_request_id":   "<pull_request_id>",
             "url":               "<url>",
             "title":             "<title>",
             "description":       "<description>",
             "status" :           "<status>",
             "created_on":        "<date_time_created>",
             "updated_on":        "<date_time_updated>",
             "versions":          "<number_or_versions_of_pr>",
             "commit_ids":        [
                                      ...
                                      "<commit_id>",
                                      "<commit_id>",
                                      ...
                                  ],
             "review_status":    "<review_status>",
             "mergeable":         {
                                      "status":  "<bool>",
                                      "message": "<message>",
                                  },
             "source":            {
                                      "clone_url":     "<clone_url>",
                                      "repository":    "<repository_name>",
                                      "reference":
                                      {
                                          "name":      "<name>",
                                          "type":      "<type>",
                                          "commit_id": "<commit_id>",
                                      }
                                  },
             "target":            {
                                      "clone_url":   "<clone_url>",
                                      "repository":    "<repository_name>",
                                      "reference":
                                      {
                                          "name":      "<name>",
                                          "type":      "<type>",
                                          "commit_id": "<commit_id>",
                                      }
                                  },
             "merge":             {
                                      "clone_url":   "<clone_url>",
                                      "reference":
                                      {
                                          "name":      "<name>",
                                          "type":      "<type>",
                                          "commit_id": "<commit_id>",
                                      }
                                  },
            "author":             <user_obj>,
            "reviewers":          [
                                      ...
                                      {
                                         "user":          "<user_obj>",
                                         "review_status": "<review_status>",
                                      }
                                      ...
                                  ]
         },
        "error": null
     """
 
     pull_request = get_pull_request_or_error(pullrequestid)
     if Optional.extract(repoid):
         repo = get_repo_or_error(repoid)
     else:
         repo = pull_request.target_repo
 
     if not PullRequestModel().check_user_read(pull_request, apiuser, api=True):
         raise JSONRPCError('repository `%s` or pull request `%s` '
                            'does not exist' % (repoid, pullrequestid))
 
     # NOTE(marcink): only calculate and return merge state if the pr state is 'created'
     # otherwise we can lock the repo on calculation of merge state while update/merge
     # is happening.
     pr_created = pull_request.pull_request_state == pull_request.STATE_CREATED
     merge_state = Optional.extract(merge_state, binary=True) and pr_created
     data = pull_request.get_api_data(with_merge_state=merge_state)
     return data
 
 
 @jsonrpc_method()
 def get_pull_requests(request, apiuser, repoid, status=Optional('new'),
                       merge_state=Optional(False)):
     """
     Get all pull requests from the repository specified in `repoid`.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Optional repository name or repository ID.
     :type repoid: str or int
     :param status: Only return pull requests with the specified status.
         Valid options are.
         * ``new`` (default)
         * ``open``
         * ``closed``
     :type status: str
     :param merge_state: Optional calculate merge state for each repository.
         This could result in longer time to fetch the data
     :type merge_state: bool
 
     Example output:
 
     .. code-block:: bash
 
       "id": <id_given_in_input>,
       "result":
         [
             ...
             {
                 "pull_request_id":   "<pull_request_id>",
                 "url":               "<url>",
                 "title" :            "<title>",
                 "description":       "<description>",
                 "status":            "<status>",
                 "created_on":        "<date_time_created>",
                 "updated_on":        "<date_time_updated>",
                 "commit_ids":        [
                                          ...
                                          "<commit_id>",
                                          "<commit_id>",
                                          ...
                                      ],
                 "review_status":    "<review_status>",
                 "mergeable":         {
                                         "status":      "<bool>",
                                         "message:      "<message>",
                                      },
                 "source":            {
                                          "clone_url":     "<clone_url>",
                                          "reference":
                                          {
                                              "name":      "<name>",
                                              "type":      "<type>",
                                              "commit_id": "<commit_id>",
                                          }
                                      },
                 "target":            {
                                          "clone_url":   "<clone_url>",
                                          "reference":
                                          {
                                              "name":      "<name>",
                                              "type":      "<type>",
                                              "commit_id": "<commit_id>",
                                          }
                                      },
                 "merge":             {
                                          "clone_url":   "<clone_url>",
                                          "reference":
                                          {
                                              "name":      "<name>",
                                              "type":      "<type>",
                                              "commit_id": "<commit_id>",
                                          }
                                      },
                "author":             <user_obj>,
                "reviewers":          [
                                          ...
                                          {
                                             "user":          "<user_obj>",
                                             "review_status": "<review_status>",
                                          }
                                          ...
                                      ]
             }
             ...
         ],
       "error": null
 
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = (
             'repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     status = Optional.extract(status)
     merge_state = Optional.extract(merge_state, binary=True)
     pull_requests = PullRequestModel().get_all(repo, statuses=[status],
                                                order_by='id', order_dir='desc')
     data = [pr.get_api_data(with_merge_state=merge_state) for pr in pull_requests]
     return data
 
 
 @jsonrpc_method()
 def merge_pull_request(
         request, apiuser, pullrequestid, repoid=Optional(None),
         userid=Optional(OAttr('apiuser'))):
     """
     Merge the pull request specified by `pullrequestid` into its target
     repository.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Optional, repository name or repository ID of the
         target repository to which the |pr| is to be merged.
     :type repoid: str or int
     :param pullrequestid: ID of the pull request which shall be merged.
     :type pullrequestid: int
     :param userid: Merge the pull request as this user.
     :type userid: Optional(str or int)
 
     Example output:
 
     .. code-block:: bash
 
         "id": <id_given_in_input>,
         "result": {
             "executed":               "<bool>",
             "failure_reason":         "<int>",
             "merge_status_message":   "<str>",
             "merge_commit_id":        "<merge_commit_id>",
             "possible":               "<bool>",
             "merge_ref":        {
                                     "commit_id": "<commit_id>",
                                     "type":      "<type>",
                                     "name":      "<name>"
                                 }
         },
         "error": null
     """
     pull_request = get_pull_request_or_error(pullrequestid)
     if Optional.extract(repoid):
         repo = get_repo_or_error(repoid)
     else:
         repo = pull_request.target_repo
     auth_user = apiuser
 
     if not isinstance(userid, Optional):
         is_repo_admin = HasRepoPermissionAnyApi('repository.admin')(
             user=apiuser, repo_name=repo.repo_name)
         if has_superadmin_permission(apiuser) or is_repo_admin:
             apiuser = get_user_or_error(userid)
             auth_user = apiuser.AuthUser()
         else:
             raise JSONRPCError('userid is not the same as your user')
 
     if pull_request.pull_request_state != PullRequest.STATE_CREATED:
         raise JSONRPCError(
             'Operation forbidden because pull request is in state {}, '
             'only state {} is allowed.'.format(
                 pull_request.pull_request_state, PullRequest.STATE_CREATED))
 
     with pull_request.set_state(PullRequest.STATE_UPDATING):
         check = MergeCheck.validate(pull_request, auth_user=auth_user,
                                     translator=request.translate)
     merge_possible = not check.failed
 
     if not merge_possible:
         error_messages = []
         for err_type, error_msg in check.errors:
             error_msg = request.translate(error_msg)
             error_messages.append(error_msg)
 
         reasons = ','.join(error_messages)
         raise JSONRPCError(
-            'merge not possible for following reasons: {}'.format(reasons))
+            f'merge not possible for following reasons: {reasons}')
 
     target_repo = pull_request.target_repo
     extras = vcs_operation_context(
         request.environ, repo_name=target_repo.repo_name,
         username=auth_user.username, action='push',
         scm=target_repo.repo_type)
     with pull_request.set_state(PullRequest.STATE_UPDATING):
         merge_response = PullRequestModel().merge_repo(
             pull_request, apiuser, extras=extras)
     if merge_response.executed:
         PullRequestModel().close_pull_request(pull_request.pull_request_id, auth_user)
 
         Session().commit()
 
     # In previous versions the merge response directly contained the merge
     # commit id. It is now contained in the merge reference object. To be
     # backwards compatible we have to extract it again.
     merge_response = merge_response.asdict()
     merge_response['merge_commit_id'] = merge_response['merge_ref'].commit_id
 
     return merge_response
 
 
 @jsonrpc_method()
 def get_pull_request_comments(
         request, apiuser, pullrequestid, repoid=Optional(None)):
     """
     Get all comments of pull request specified with the `pullrequestid`
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Optional repository name or repository ID.
     :type repoid: str or int
     :param pullrequestid: The pull request ID.
     :type pullrequestid: int
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : [
             {
               "comment_author": {
                 "active": true,
                 "full_name_or_username": "Tom Gore",
                 "username": "admin"
               },
               "comment_created_on": "2017-01-02T18:43:45.533",
               "comment_f_path": null,
               "comment_id": 25,
               "comment_lineno": null,
               "comment_status": {
                 "status": "under_review",
                 "status_lbl": "Under Review"
               },
               "comment_text": "Example text",
               "comment_type": null,
               "comment_last_version: 0,
               "pull_request_version": null,
               "comment_commit_id": None,
               "comment_pull_request_id": <pull_request_id>
             }
         ],
         error :  null
     """
 
     pull_request = get_pull_request_or_error(pullrequestid)
     if Optional.extract(repoid):
         repo = get_repo_or_error(repoid)
     else:
         repo = pull_request.target_repo
 
     if not PullRequestModel().check_user_read(
             pull_request, apiuser, api=True):
         raise JSONRPCError('repository `%s` or pull request `%s` '
                            'does not exist' % (repoid, pullrequestid))
 
     (pull_request_latest,
      pull_request_at_ver,
      pull_request_display_obj,
      at_version) = PullRequestModel().get_pr_version(
         pull_request.pull_request_id, version=None)
 
     versions = pull_request_display_obj.versions()
     ver_map = {
         ver.pull_request_version_id: cnt
         for cnt, ver in enumerate(versions, 1)
     }
 
     # GENERAL COMMENTS with versions #
     q = CommentsModel()._all_general_comments_of_pull_request(pull_request)
     q = q.order_by(ChangesetComment.comment_id.asc())
     general_comments = q.all()
 
     # INLINE COMMENTS with versions  #
     q = CommentsModel()._all_inline_comments_of_pull_request(pull_request)
     q = q.order_by(ChangesetComment.comment_id.asc())
     inline_comments = q.all()
 
     data = []
     for comment in inline_comments + general_comments:
         full_data = comment.get_api_data()
         pr_version_id = None
         if comment.pull_request_version_id:
             pr_version_id = 'v{}'.format(
                 ver_map[comment.pull_request_version_id])
 
         # sanitize some entries
 
         full_data['pull_request_version'] = pr_version_id
         full_data['comment_author'] = {
             'username': full_data['comment_author'].username,
             'full_name_or_username': full_data['comment_author'].full_name_or_username,
             'active': full_data['comment_author'].active,
         }
 
         if full_data['comment_status']:
             full_data['comment_status'] = {
                 'status': full_data['comment_status'][0].status,
                 'status_lbl': full_data['comment_status'][0].status_lbl,
             }
         else:
             full_data['comment_status'] = {}
 
         data.append(full_data)
     return data
 
 
 @jsonrpc_method()
 def comment_pull_request(
         request, apiuser, pullrequestid, repoid=Optional(None),
         message=Optional(None), commit_id=Optional(None), status=Optional(None),
         comment_type=Optional(ChangesetComment.COMMENT_TYPE_NOTE),
         resolves_comment_id=Optional(None), extra_recipients=Optional([]),
         userid=Optional(OAttr('apiuser')), send_email=Optional(True)):
     """
     Comment on the pull request specified with the `pullrequestid`,
     in the |repo| specified by the `repoid`, and optionally change the
     review status.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Optional repository name or repository ID.
     :type repoid: str or int
     :param pullrequestid: The pull request ID.
     :type pullrequestid: int
     :param commit_id: Specify the commit_id for which to set a comment. If
         given commit_id is different than latest in the PR status
         change won't be performed.
     :type commit_id: str
     :param message: The text content of the comment.
     :type message: str
     :param status: (**Optional**) Set the approval status of the pull
         request. One of: 'not_reviewed', 'approved', 'rejected',
         'under_review'
     :type status: str
     :param comment_type: Comment type, one of: 'note', 'todo'
     :type comment_type: Optional(str), default: 'note'
     :param resolves_comment_id: id of comment which this one will resolve
     :type resolves_comment_id: Optional(int)
     :param extra_recipients: list of user ids or usernames to add
         notifications for this comment. Acts like a CC for notification
     :type extra_recipients: Optional(list)
     :param userid: Comment on the pull request as this user
     :type userid: Optional(str or int)
     :param send_email: Define if this comment should also send email notification
     :type send_email: Optional(bool)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : {
             "pull_request_id":  "<Integer>",
             "comment_id":       "<Integer>",
             "status": {"given": <given_status>,
                        "was_changed": <bool status_was_actually_changed> },
         },
         error :  null
     """
     _ = request.translate
 
     pull_request = get_pull_request_or_error(pullrequestid)
     if Optional.extract(repoid):
         repo = get_repo_or_error(repoid)
     else:
         repo = pull_request.target_repo
 
     db_repo_name = repo.repo_name
     auth_user = apiuser
     if not isinstance(userid, Optional):
         is_repo_admin = HasRepoPermissionAnyApi('repository.admin')(
             user=apiuser, repo_name=db_repo_name)
         if has_superadmin_permission(apiuser) or is_repo_admin:
             apiuser = get_user_or_error(userid)
             auth_user = apiuser.AuthUser()
         else:
             raise JSONRPCError('userid is not the same as your user')
 
     if pull_request.is_closed():
         raise JSONRPCError(f'pull request `{pullrequestid}` comment failed, pull request is closed')
 
     if not PullRequestModel().check_user_read(
             pull_request, apiuser, api=True):
-        raise JSONRPCError('repository `%s` does not exist' % (repoid,))
+        raise JSONRPCError('repository `{}` does not exist'.format(repoid))
     message = Optional.extract(message)
     status = Optional.extract(status)
     commit_id = Optional.extract(commit_id)
     comment_type = Optional.extract(comment_type)
     resolves_comment_id = Optional.extract(resolves_comment_id)
     extra_recipients = Optional.extract(extra_recipients)
     send_email = Optional.extract(send_email, binary=True)
 
     if not message and not status:
         raise JSONRPCError(
             'Both message and status parameters are missing. '
             'At least one is required.')
 
     if status and status not in (st[0] for st in ChangesetStatus.STATUSES):
         raise JSONRPCError(f'Unknown comment status: `{status}`')
 
     if commit_id and commit_id not in pull_request.revisions:
         raise JSONRPCError(f'Invalid commit_id `{commit_id}` for this pull request.')
 
     allowed_to_change_status = PullRequestModel().check_user_change_status(
         pull_request, apiuser)
 
     # if commit_id is passed re-validated if user is allowed to change status
     # based on the latest commit_id from the PR
     if commit_id:
         commit_idx = pull_request.revisions.index(commit_id)
         if commit_idx != 0:
             log.warning('Resetting allowed_to_change_status = False because commit is NOT the latest in pull-request')
             allowed_to_change_status = False
 
     if resolves_comment_id:
         comment = ChangesetComment.get(resolves_comment_id)
         if not comment:
             raise JSONRPCError(f'Invalid resolves_comment_id `{resolves_comment_id}` for this pull request.')
         if comment.comment_type != ChangesetComment.COMMENT_TYPE_TODO:
             raise JSONRPCError(f'Comment `{resolves_comment_id}` is wrong type for setting status to resolved.')
 
     text = message
     status_label = ChangesetStatus.get_status_lbl(status)
     if status and allowed_to_change_status:
         st_message = ('Status change %(transition_icon)s %(status)s'
                       % {'transition_icon': '>', 'status': status_label})
         text = message or st_message
 
     rc_config = SettingsModel().get_all_settings()
     renderer = rc_config.get('rhodecode_markup_renderer', 'rst')
 
     status_change = status and allowed_to_change_status
     comment = CommentsModel().create(
         text=text,
         repo=pull_request.target_repo.repo_id,
         user=apiuser.user_id,
         pull_request=pull_request.pull_request_id,
         f_path=None,
         line_no=None,
         status_change=(status_label if status_change else None),
         status_change_type=(status if status_change else None),
         closing_pr=False,
         renderer=renderer,
         comment_type=comment_type,
         resolves_comment_id=resolves_comment_id,
         auth_user=auth_user,
         extra_recipients=extra_recipients,
         send_email=send_email
     )
 
     is_inline = comment.is_inline
 
     if allowed_to_change_status and status:
         old_calculated_status = pull_request.calculated_review_status()
         ChangesetStatusModel().set_status(
             pull_request.target_repo.repo_id,
             status,
             apiuser.user_id,
             comment,
             pull_request=pull_request.pull_request_id
         )
         Session().flush()
 
     Session().commit()
 
     PullRequestModel().trigger_pull_request_hook(
         pull_request, apiuser, 'comment',
         data={'comment': comment})
 
     if allowed_to_change_status and status:
         # we now calculate the status of pull request, and based on that
         # calculation we set the commits status
         calculated_status = pull_request.calculated_review_status()
         if old_calculated_status != calculated_status:
             PullRequestModel().trigger_pull_request_hook(
                 pull_request, apiuser, 'review_status_change',
                 data={'status': calculated_status})
 
     data = {
         'pull_request_id': pull_request.pull_request_id,
         'comment_id': comment.comment_id if comment else None,
         'status': {'given': status, 'was_changed': status_change},
     }
 
     comment_broadcast_channel = channelstream.comment_channel(
         db_repo_name, pull_request_obj=pull_request)
 
     comment_data = data
     comment_type = 'inline' if is_inline else 'general'
     channelstream.comment_channelstream_push(
         request, comment_broadcast_channel, apiuser,
         _('posted a new {} comment').format(comment_type),
         comment_data=comment_data)
 
     return data
 
 
 def _reviewers_validation(obj_list):
     schema = ReviewerListSchema()
     try:
         reviewer_objects = schema.deserialize(obj_list)
     except Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     # validate users
     for reviewer_object in reviewer_objects:
         user = get_user_or_error(reviewer_object['username'])
         reviewer_object['user_id'] = user.user_id
     return reviewer_objects
 
 
 @jsonrpc_method()
 def create_pull_request(
         request, apiuser, source_repo, target_repo, source_ref, target_ref,
         owner=Optional(OAttr('apiuser')), title=Optional(''), description=Optional(''),
         description_renderer=Optional(''),
         reviewers=Optional(None), observers=Optional(None)):
     """
     Creates a new pull request.
 
     Accepts refs in the following formats:
 
         * branch:<branch_name>:<sha>
         * branch:<branch_name>
         * bookmark:<bookmark_name>:<sha> (Mercurial only)
         * bookmark:<bookmark_name> (Mercurial only)
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param source_repo: Set the source repository name.
     :type source_repo: str
     :param target_repo: Set the target repository name.
     :type target_repo: str
     :param source_ref: Set the source ref name.
     :type source_ref: str
     :param target_ref: Set the target ref name.
     :type target_ref: str
     :param owner: user_id or username
     :type owner: Optional(str)
     :param title: Optionally Set the pull request title, it's generated otherwise
     :type title: str
     :param description: Set the pull request description.
     :type description: Optional(str)
     :type description_renderer: Optional(str)
     :param description_renderer: Set pull request renderer for the description.
         It should be 'rst', 'markdown' or 'plain'. If not give default
         system renderer will be used
     :param reviewers: Set the new pull request reviewers list.
         Reviewer defined by review rules will be added automatically to the
         defined list.
     :type reviewers: Optional(list)
         Accepts username strings or objects of the format:
 
             [{'username': 'nick', 'reasons': ['original author'], 'mandatory': <bool>}]
     :param observers: Set the new pull request observers list.
         Reviewer defined by review rules will be added automatically to the
         defined list. This feature is only available in RhodeCode EE
     :type observers: Optional(list)
         Accepts username strings or objects of the format:
 
             [{'username': 'nick', 'reasons': ['original author']}]
     """
 
     source_db_repo = get_repo_or_error(source_repo)
     target_db_repo = get_repo_or_error(target_repo)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, source_repo, source_db_repo, _perms)
 
     owner = validate_set_owner_permissions(apiuser, owner)
 
     full_source_ref = resolve_ref_or_error(source_ref, source_db_repo)
     full_target_ref = resolve_ref_or_error(target_ref, target_db_repo)
 
     get_commit_or_error(full_source_ref, source_db_repo)
     get_commit_or_error(full_target_ref, target_db_repo)
 
     reviewer_objects = Optional.extract(reviewers) or []
     observer_objects = Optional.extract(observers) or []
 
     # serialize and validate passed in given reviewers
     if reviewer_objects:
         reviewer_objects = _reviewers_validation(reviewer_objects)
 
     if observer_objects:
         observer_objects = _reviewers_validation(reviewer_objects)
 
     get_default_reviewers_data, validate_default_reviewers, validate_observers = \
         PullRequestModel().get_reviewer_functions()
 
     source_ref_obj = unicode_to_reference(full_source_ref)
     target_ref_obj = unicode_to_reference(full_target_ref)
 
     # recalculate reviewers logic, to make sure we can validate this
     default_reviewers_data = get_default_reviewers_data(
         owner,
         source_db_repo,
         source_ref_obj,
         target_db_repo,
         target_ref_obj,
     )
 
     # now MERGE our given with the calculated from the default rules
     just_reviewers = [
         x for x in default_reviewers_data['reviewers']
         if x['role'] == PullRequestReviewers.ROLE_REVIEWER]
     reviewer_objects = just_reviewers + reviewer_objects
 
     try:
         reviewers = validate_default_reviewers(
             reviewer_objects, default_reviewers_data)
     except ValueError as e:
-        raise JSONRPCError('Reviewers Validation: {}'.format(e))
+        raise JSONRPCError(f'Reviewers Validation: {e}')
 
     # now MERGE our given with the calculated from the default rules
     just_observers = [
         x for x in default_reviewers_data['reviewers']
         if x['role'] == PullRequestReviewers.ROLE_OBSERVER]
     observer_objects = just_observers + observer_objects
 
     try:
         observers = validate_observers(
             observer_objects, default_reviewers_data)
     except ValueError as e:
-        raise JSONRPCError('Observer Validation: {}'.format(e))
+        raise JSONRPCError(f'Observer Validation: {e}')
 
     title = Optional.extract(title)
     if not title:
         title_source_ref = source_ref_obj.name
         title = PullRequestModel().generate_pullrequest_title(
             source=source_repo,
             source_ref=title_source_ref,
             target=target_repo
         )
 
     diff_info = default_reviewers_data['diff_info']
     common_ancestor_id = diff_info['ancestor']
     # NOTE(marcink): reversed is consistent with how we open it in the WEB interface
     commits = [commit['commit_id'] for commit in reversed(diff_info['commits'])]
 
     if not common_ancestor_id:
         raise JSONRPCError('no common ancestor found between specified references')
 
     if not commits:
         raise JSONRPCError('no commits found for merge between specified references')
 
     # recalculate target ref based on ancestor
     full_target_ref = ':'.join((target_ref_obj.type, target_ref_obj.name, common_ancestor_id))
 
     # fetch renderer, if set fallback to plain in case of PR
     rc_config = SettingsModel().get_all_settings()
     default_system_renderer = rc_config.get('rhodecode_markup_renderer', 'plain')
     description = Optional.extract(description)
     description_renderer = Optional.extract(description_renderer) or default_system_renderer
 
     pull_request = PullRequestModel().create(
         created_by=owner.user_id,
         source_repo=source_repo,
         source_ref=full_source_ref,
         target_repo=target_repo,
         target_ref=full_target_ref,
         common_ancestor_id=common_ancestor_id,
         revisions=commits,
         reviewers=reviewers,
         observers=observers,
         title=title,
         description=description,
         description_renderer=description_renderer,
         reviewer_data=default_reviewers_data,
         auth_user=apiuser
     )
 
     Session().commit()
     data = {
-        'msg': 'Created new pull request `{}`'.format(title),
+        'msg': f'Created new pull request `{title}`',
         'pull_request_id': pull_request.pull_request_id,
     }
     return data
 
 
 @jsonrpc_method()
 def update_pull_request(
         request, apiuser, pullrequestid, repoid=Optional(None),
         title=Optional(''), description=Optional(''), description_renderer=Optional(''),
         reviewers=Optional(None), observers=Optional(None), update_commits=Optional(None)):
     """
     Updates a pull request.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Optional repository name or repository ID.
     :type repoid: str or int
     :param pullrequestid: The pull request ID.
     :type pullrequestid: int
     :param title: Set the pull request title.
     :type title: str
     :param description: Update pull request description.
     :type description: Optional(str)
     :type description_renderer: Optional(str)
     :param description_renderer: Update pull request renderer for the description.
         It should be 'rst', 'markdown' or 'plain'
     :param reviewers: Update pull request reviewers list with new value.
     :type reviewers: Optional(list)
         Accepts username strings or objects of the format:
 
             [{'username': 'nick', 'reasons': ['original author'], 'mandatory': <bool>}]
     :param observers: Update pull request observers list with new value.
     :type observers: Optional(list)
         Accepts username strings or objects of the format:
 
             [{'username': 'nick', 'reasons': ['should be aware about this PR']}]
     :param update_commits: Trigger update of commits for this pull request
     :type: update_commits: Optional(bool)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : {
             "msg": "Updated pull request `63`",
             "pull_request": <pull_request_object>,
             "updated_reviewers": {
               "added": [
                 "username"
               ],
               "removed": []
             },
             "updated_observers": {
               "added": [
                 "username"
               ],
               "removed": []
             },
             "updated_commits": {
               "added": [
                 "<sha1_hash>"
               ],
               "common": [
                 "<sha1_hash>",
                 "<sha1_hash>",
               ],
               "removed": []
             }
         }
         error :  null
     """
 
     pull_request = get_pull_request_or_error(pullrequestid)
     if Optional.extract(repoid):
         repo = get_repo_or_error(repoid)
     else:
         repo = pull_request.target_repo
 
     if not PullRequestModel().check_user_update(
             pull_request, apiuser, api=True):
         raise JSONRPCError(
-            'pull request `%s` update failed, no permission to update.' % (
-                pullrequestid,))
+            'pull request `{}` update failed, no permission to update.'.format(
+                pullrequestid))
     if pull_request.is_closed():
         raise JSONRPCError(
-            'pull request `%s` update failed, pull request is closed' % (
-                pullrequestid,))
+            'pull request `{}` update failed, pull request is closed'.format(
+                pullrequestid))
 
     reviewer_objects = Optional.extract(reviewers) or []
     observer_objects = Optional.extract(observers) or []
 
     title = Optional.extract(title)
     description = Optional.extract(description)
     description_renderer = Optional.extract(description_renderer)
 
     # Update title/description
     title_changed = False
     if title or description:
         PullRequestModel().edit(
             pull_request,
             title or pull_request.title,
             description or pull_request.description,
             description_renderer or pull_request.description_renderer,
             apiuser)
         Session().commit()
         title_changed = True
 
     commit_changes = {"added": [], "common": [], "removed": []}
 
     # Update commits
     commits_changed = False
     if str2bool(Optional.extract(update_commits)):
 
         if pull_request.pull_request_state != PullRequest.STATE_CREATED:
             raise JSONRPCError(
                 'Operation forbidden because pull request is in state {}, '
                 'only state {} is allowed.'.format(
                     pull_request.pull_request_state, PullRequest.STATE_CREATED))
 
         with pull_request.set_state(PullRequest.STATE_UPDATING):
             if PullRequestModel().has_valid_update_type(pull_request):
                 db_user = apiuser.get_instance()
                 update_response = PullRequestModel().update_commits(
                     pull_request, db_user)
                 commit_changes = update_response.changes or commit_changes
             Session().commit()
             commits_changed = True
 
     # Update reviewers
     # serialize and validate passed in given reviewers
     if reviewer_objects:
         reviewer_objects = _reviewers_validation(reviewer_objects)
 
     if observer_objects:
         observer_objects = _reviewers_validation(reviewer_objects)
 
     # re-use stored rules
     default_reviewers_data = pull_request.reviewer_data
 
     __, validate_default_reviewers, validate_observers = \
         PullRequestModel().get_reviewer_functions()
 
     if reviewer_objects:
         try:
             reviewers = validate_default_reviewers(reviewer_objects, default_reviewers_data)
         except ValueError as e:
-            raise JSONRPCError('Reviewers Validation: {}'.format(e))
+            raise JSONRPCError(f'Reviewers Validation: {e}')
     else:
         reviewers = []
 
     if observer_objects:
         try:
             observers = validate_default_reviewers(reviewer_objects, default_reviewers_data)
         except ValueError as e:
-            raise JSONRPCError('Observer Validation: {}'.format(e))
+            raise JSONRPCError(f'Observer Validation: {e}')
     else:
         observers = []
 
     reviewers_changed = False
     reviewers_changes = {"added": [], "removed": []}
     if reviewers:
         old_calculated_status = pull_request.calculated_review_status()
         added_reviewers, removed_reviewers = \
             PullRequestModel().update_reviewers(pull_request, reviewers, apiuser.get_instance())
 
         reviewers_changes['added'] = sorted(
             [get_user_or_error(n).username for n in added_reviewers])
         reviewers_changes['removed'] = sorted(
             [get_user_or_error(n).username for n in removed_reviewers])
         Session().commit()
 
         # trigger status changed if change in reviewers changes the status
         calculated_status = pull_request.calculated_review_status()
         if old_calculated_status != calculated_status:
             PullRequestModel().trigger_pull_request_hook(
                 pull_request, apiuser, 'review_status_change',
                 data={'status': calculated_status})
         reviewers_changed = True
 
     observers_changed = False
     observers_changes = {"added": [], "removed": []}
     if observers:
         added_observers, removed_observers = \
             PullRequestModel().update_observers(pull_request, observers, apiuser.get_instance())
 
         observers_changes['added'] = sorted(
             [get_user_or_error(n).username for n in added_observers])
         observers_changes['removed'] = sorted(
             [get_user_or_error(n).username for n in removed_observers])
         Session().commit()
 
         reviewers_changed = True
 
     # push changed to channelstream
     if commits_changed or reviewers_changed or observers_changed:
         pr_broadcast_channel = channelstream.pr_channel(pull_request)
         msg = 'Pull request was updated.'
         channelstream.pr_update_channelstream_push(
             request, pr_broadcast_channel, apiuser, msg)
 
     data = {
-        'msg': 'Updated pull request `{}`'.format(pull_request.pull_request_id),
+        'msg': f'Updated pull request `{pull_request.pull_request_id}`',
         'pull_request': pull_request.get_api_data(),
         'updated_commits': commit_changes,
         'updated_reviewers': reviewers_changes,
         'updated_observers': observers_changes,
     }
 
     return data
 
 
 @jsonrpc_method()
 def close_pull_request(
         request, apiuser, pullrequestid, repoid=Optional(None),
         userid=Optional(OAttr('apiuser')), message=Optional('')):
     """
     Close the pull request specified by `pullrequestid`.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Repository name or repository ID to which the pull
         request belongs.
     :type repoid: str or int
     :param pullrequestid: ID of the pull request to be closed.
     :type pullrequestid: int
     :param userid: Close the pull request as this user.
     :type userid: Optional(str or int)
     :param message: Optional message to close the Pull Request with. If not
         specified it will be generated automatically.
     :type message: Optional(str)
 
     Example output:
 
     .. code-block:: bash
 
         "id": <id_given_in_input>,
         "result": {
             "pull_request_id":  "<int>",
             "close_status":     "<str:status_lbl>,
             "closed":           "<bool>"
         },
         "error": null
 
     """
     _ = request.translate
 
     pull_request = get_pull_request_or_error(pullrequestid)
     if Optional.extract(repoid):
         repo = get_repo_or_error(repoid)
     else:
         repo = pull_request.target_repo
 
     is_repo_admin = HasRepoPermissionAnyApi('repository.admin')(
                     user=apiuser, repo_name=repo.repo_name)
     if not isinstance(userid, Optional):
         if has_superadmin_permission(apiuser) or is_repo_admin:
             apiuser = get_user_or_error(userid)
         else:
             raise JSONRPCError('userid is not the same as your user')
 
     if pull_request.is_closed():
         raise JSONRPCError(
-            'pull request `%s` is already closed' % (pullrequestid,))
+            'pull request `{}` is already closed'.format(pullrequestid))
 
     # only owner or admin or person with write permissions
     allowed_to_close = PullRequestModel().check_user_update(
             pull_request, apiuser, api=True)
 
     if not allowed_to_close:
         raise JSONRPCError(
-            'pull request `%s` close failed, no permission to close.' % (
-                pullrequestid,))
+            'pull request `{}` close failed, no permission to close.'.format(
+                pullrequestid))
 
     # message we're using to close the PR, else it's automatically generated
     message = Optional.extract(message)
 
     # finally close the PR, with proper message comment
     comment, status = PullRequestModel().close_pull_request_with_comment(
         pull_request, apiuser, repo, message=message, auth_user=apiuser)
     status_lbl = ChangesetStatus.get_status_lbl(status)
 
     Session().commit()
 
     data = {
         'pull_request_id': pull_request.pull_request_id,
         'close_status': status_lbl,
         'closed': True,
     }
     return data

rhodecode/api/views/repo_api.py

@@ -1,2536 +1,2534 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 import time
 
 import rhodecode
 from rhodecode.api import (
     jsonrpc_method, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
 from rhodecode.api.utils import (
     has_superadmin_permission, Optional, OAttr, get_repo_or_error,
     get_user_group_or_error, get_user_or_error, validate_repo_permissions,
     get_perm_or_error, parse_args, get_origin, build_commit_data,
     validate_set_owner_permissions)
 from rhodecode.lib import audit_logger, rc_cache, channelstream
 from rhodecode.lib import repo_maintenance
 from rhodecode.lib.auth import (
     HasPermissionAnyApi, HasUserGroupPermissionAnyApi,
     HasRepoPermissionAnyApi)
 from rhodecode.lib.celerylib.utils import get_task_id
 from rhodecode.lib.utils2 import (
     str2bool, time_to_datetime, safe_str, safe_int)
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.exceptions import (
     StatusChangeOnClosedPullRequestError, CommentVersionMismatch)
 from rhodecode.lib.vcs import RepositoryError
 from rhodecode.lib.vcs.exceptions import NodeDoesNotExistError
 from rhodecode.model.changeset_status import ChangesetStatusModel
 from rhodecode.model.comment import CommentsModel
 from rhodecode.model.db import (
     Session, ChangesetStatus, RepositoryField, Repository, RepoGroup,
     ChangesetComment)
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.pull_request import PullRequestModel
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.scm import ScmModel, RepoList
 from rhodecode.model.settings import SettingsModel, VcsSettingsModel
 from rhodecode.model import validation_schema
 from rhodecode.model.validation_schema.schemas import repo_schema
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_repo(request, apiuser, repoid, cache=Optional(True)):
     """
     Gets an existing repository by its name or repository_id.
 
     The members section so the output returns users groups or users
     associated with that repository.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository id.
     :type repoid: str or int
     :param cache: use the cached value for last changeset
     :type: cache: Optional(bool)
 
     Example output:
 
     .. code-block:: bash
 
         {
           "error": null,
           "id": <repo_id>,
           "result": {
             "clone_uri": null,
             "created_on": "timestamp",
             "description": "repo description",
             "enable_downloads": false,
             "enable_locking": false,
             "enable_statistics": false,
             "followers": [
               {
                 "active": true,
                 "admin": false,
                 "api_key": "****************************************",
                 "api_keys": [
                   "****************************************"
                 ],
                 "email": "[e-mail]",
                 "emails": [
                   "[e-mail]"
                 ],
                 "extern_name": "rhodecode",
                 "extern_type": "rhodecode",
                 "firstname": "username",
                 "ip_addresses": [],
                 "language": null,
                 "last_login": "2015-09-16T17:16:35.854",
                 "lastname": "surname",
                 "user_id": <user_id>,
                 "username": "name"
               }
             ],
             "fork_of": "parent-repo",
             "landing_rev": [
               "rev",
               "tip"
             ],
             "last_changeset": {
               "author": "User <[e-mail]>",
               "branch": "default",
               "date": "timestamp",
               "message": "last commit message",
               "parents": [
                 {
                   "raw_id": "commit-id"
                 }
               ],
               "raw_id": "commit-id",
               "revision": <revision number>,
               "short_id": "short id"
             },
             "lock_reason": null,
             "locked_by": null,
             "locked_date": null,
             "owner": "owner-name",
             "permissions": [
               {
                 "name": "super-admin-name",
                 "origin": "super-admin",
                 "permission": "repository.admin",
                 "type": "user"
               },
               {
                 "name": "owner-name",
                 "origin": "owner",
                 "permission": "repository.admin",
                 "type": "user"
               },
               {
                 "name": "user-group-name",
                 "origin": "permission",
                 "permission": "repository.write",
                 "type": "user_group"
               }
             ],
             "private": true,
             "repo_id": 676,
             "repo_name": "user-group/repo-name",
             "repo_type": "hg"
           }
         }
     """
 
     repo = get_repo_or_error(repoid)
     cache = Optional.extract(cache)
 
     include_secrets = False
     if has_superadmin_permission(apiuser):
         include_secrets = True
     else:
         # check if we have at least read permission for this repo !
         _perms = (
             'repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     permissions = []
     for _user in repo.permissions():
         user_data = {
             'name': _user.username,
             'permission': _user.permission,
             'origin': get_origin(_user),
             'type': "user",
         }
         permissions.append(user_data)
 
     for _user_group in repo.permission_user_groups():
         user_group_data = {
             'name': _user_group.users_group_name,
             'permission': _user_group.permission,
             'origin': get_origin(_user_group),
             'type': "user_group",
         }
         permissions.append(user_group_data)
 
     following_users = [
         user.user.get_api_data(include_secrets=include_secrets)
         for user in repo.followers]
 
     if not cache:
         repo.update_commit_cache()
     data = repo.get_api_data(include_secrets=include_secrets)
     data['permissions'] = permissions
     data['followers'] = following_users
 
     return data
 
 
 @jsonrpc_method()
 def get_repos(request, apiuser, root=Optional(None), traverse=Optional(True)):
     """
     Lists all existing repositories.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param root: specify root repository group to fetch repositories.
         filters the returned repositories to be members of given root group.
     :type root: Optional(None)
     :param traverse: traverse given root into subrepositories. With this flag
         set to False, it will only return top-level repositories from `root`.
         if root is empty it will return just top-level repositories.
     :type traverse: Optional(True)
 
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: [
                   {
                     "repo_id" :          "<repo_id>",
                     "repo_name" :        "<reponame>"
                     "repo_type" :        "<repo_type>",
                     "clone_uri" :        "<clone_uri>",
                     "private": :         "<bool>",
                     "created_on" :       "<datetimecreated>",
                     "description" :      "<description>",
                     "landing_rev":       "<landing_rev>",
                     "owner":             "<repo_owner>",
                     "fork_of":           "<name_of_fork_parent>",
                     "enable_downloads":  "<bool>",
                     "enable_locking":    "<bool>",
                     "enable_statistics": "<bool>",
                   },
                   ...
                 ]
         error:  null
     """
 
     include_secrets = has_superadmin_permission(apiuser)
     _perms = ('repository.read', 'repository.write', 'repository.admin',)
     extras = {'user': apiuser}
 
     root = Optional.extract(root)
     traverse = Optional.extract(traverse, binary=True)
 
     if root:
         # verify parent existance, if it's empty return an error
         parent = RepoGroup.get_by_group_name(root)
         if not parent:
             raise JSONRPCError(
-                'Root repository group `{}` does not exist'.format(root))
+                f'Root repository group `{root}` does not exist')
 
         if traverse:
             repos = RepoModel().get_repos_for_root(root=root, traverse=traverse)
         else:
             repos = RepoModel().get_repos_for_root(root=parent)
     else:
         if traverse:
             repos = RepoModel().get_all()
         else:
             # return just top-level
             repos = RepoModel().get_repos_for_root(root=None)
 
     repo_list = RepoList(repos, perm_set=_perms, extra_kwargs=extras)
     return [repo.get_api_data(include_secrets=include_secrets)
             for repo in repo_list]
 
 
 @jsonrpc_method()
 def get_repo_changeset(request, apiuser, repoid, revision,
                        details=Optional('basic')):
     """
     Returns information about a changeset.
 
     Additionally parameters define the amount of details returned by
     this function.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository id
     :type repoid: str or int
     :param revision: revision for which listing should be done
     :type revision: str
     :param details: details can be 'basic|extended|full' full gives diff
         info details like the diff itself, and number of changed files etc.
     :type details: Optional(str)
 
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     changes_details = Optional.extract(details)
     _changes_details_types = ['basic', 'extended', 'full']
     if changes_details not in _changes_details_types:
         raise JSONRPCError(
             'ret_type must be one of %s' % (
                 ','.join(_changes_details_types)))
 
     vcs_repo = repo.scm_instance()
     pre_load = ['author', 'branch', 'date', 'message', 'parents',
                 'status', '_commit', '_file_paths']
 
     try:
         commit = repo.get_commit(commit_id=revision, pre_load=pre_load)
     except TypeError as e:
         raise JSONRPCError(safe_str(e))
     _cs_json = commit.__json__()
     _cs_json['diff'] = build_commit_data(vcs_repo, commit, changes_details)
     if changes_details == 'full':
         _cs_json['refs'] = commit._get_refs()
     return _cs_json
 
 
 @jsonrpc_method()
 def get_repo_changesets(request, apiuser, repoid, start_rev, limit,
                         details=Optional('basic')):
     """
     Returns a set of commits limited by the number starting
     from the `start_rev` option.
 
     Additional parameters define the amount of details returned by this
     function.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param start_rev: The starting revision from where to get changesets.
     :type start_rev: str
     :param limit: Limit the number of commits to this amount
     :type limit: str or int
     :param details: Set the level of detail returned. Valid option are:
         ``basic``, ``extended`` and ``full``.
     :type details: Optional(str)
 
     .. note::
 
        Setting the parameter `details` to the value ``full`` is extensive
        and returns details like the diff itself, and the number
        of changed files.
 
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     changes_details = Optional.extract(details)
     _changes_details_types = ['basic', 'extended', 'full']
     if changes_details not in _changes_details_types:
         raise JSONRPCError(
             'ret_type must be one of %s' % (
                 ','.join(_changes_details_types)))
 
     limit = int(limit)
     pre_load = ['author', 'branch', 'date', 'message', 'parents',
                 'status', '_commit', '_file_paths']
 
     vcs_repo = repo.scm_instance()
     # SVN needs a special case to distinguish its index and commit id
     if vcs_repo and vcs_repo.alias == 'svn' and (start_rev == '0'):
         start_rev = vcs_repo.commit_ids[0]
 
     try:
         commits = vcs_repo.get_commits(
             start_id=start_rev, pre_load=pre_load, translate_tags=False)
     except TypeError as e:
         raise JSONRPCError(safe_str(e))
     except Exception:
         log.exception('Fetching of commits failed')
         raise JSONRPCError('Error occurred during commit fetching')
 
     ret = []
     for cnt, commit in enumerate(commits):
         if cnt >= limit != -1:
             break
         _cs_json = commit.__json__()
         _cs_json['diff'] = build_commit_data(vcs_repo, commit, changes_details)
         if changes_details == 'full':
             _cs_json['refs'] = {
                 'branches': [commit.branch],
                 'bookmarks': getattr(commit, 'bookmarks', []),
                 'tags': commit.tags
             }
         ret.append(_cs_json)
     return ret
 
 
 @jsonrpc_method()
 def get_repo_nodes(request, apiuser, repoid, revision, root_path,
                    ret_type=Optional('all'), details=Optional('basic'),
                    max_file_bytes=Optional(None)):
     """
     Returns a list of nodes and 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 only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param revision: The revision for which listing should be done.
     :type revision: str
     :param root_path: The path from which to start displaying.
     :type root_path: str
     :param ret_type: Set the return type. Valid options are
         ``all`` (default), ``files`` and ``dirs``.
     :type ret_type: Optional(str)
     :param details: Returns extended information about nodes, such as
         md5, binary, and or content.
         The valid options are ``basic`` and ``full``.
     :type details: Optional(str)
     :param max_file_bytes: Only return file content under this file size bytes
     :type details: Optional(int)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: [
                     {
                       "binary": false,
                       "content": "File line",
                       "extension": "md",
                       "lines": 2,
                       "md5": "059fa5d29b19c0657e384749480f6422",
                       "mimetype": "text/x-minidsrc",
                       "name": "file.md",
                       "size": 580,
                       "type": "file"
                     },
                   ...
                 ]
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     ret_type = Optional.extract(ret_type)
     details = Optional.extract(details)
     max_file_bytes = Optional.extract(max_file_bytes)
 
     _extended_types = ['basic', 'full']
     if details not in _extended_types:
         ret_types = ','.join(_extended_types)
         raise JSONRPCError(f'ret_type must be one of {ret_types}')
 
     extended_info = False
     content = False
     if details == 'basic':
         extended_info = True
 
     if details == 'full':
         extended_info = content = True
 
     _map = {}
     try:
         # check if repo is not empty by any chance, skip quicker if it is.
         _scm = repo.scm_instance()
         if _scm.is_empty():
             return []
 
         _d, _f = ScmModel().get_nodes(
             repo, revision, root_path, flat=False,
             extended_info=extended_info, content=content,
             max_file_bytes=max_file_bytes)
 
         _map = {
             'all': _d + _f,
             'files': _f,
             'dirs': _d,
         }
 
         return _map[ret_type]
     except KeyError:
         keys = ','.join(sorted(_map.keys()))
         raise JSONRPCError(f'ret_type must be one of {keys}')
     except Exception:
         log.exception("Exception occurred while trying to get repo nodes")
         raise JSONRPCError(f'failed to get repo: `{repo.repo_name}` nodes')
 
 
 @jsonrpc_method()
 def get_repo_file(request, apiuser, repoid, commit_id, file_path,
                   max_file_bytes=Optional(0), details=Optional('basic'),
                   cache=Optional(True)):
     """
     Returns a single file from repository at given revision.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param commit_id: The revision for which listing should be done.
     :type commit_id: str
     :param file_path: The path from which to start displaying.
     :type file_path: str
     :param details: Returns different set of information about nodes.
         The valid options are ``minimal`` ``basic`` and ``full``.
     :type details: Optional(str)
     :param max_file_bytes: Only return file content under this file size bytes
     :type max_file_bytes: Optional(int)
     :param cache: Use internal caches for fetching files. If disabled fetching
         files is slower but more memory efficient
     :type cache: Optional(bool)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
             "binary": false,
             "extension": "py",
             "lines": 35,
             "content": "....",
             "md5": "76318336366b0f17ee249e11b0c99c41",
             "mimetype": "text/x-python",
             "name": "python.py",
             "size": 817,
             "type": "file",
         }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     cache = Optional.extract(cache, binary=True)
     details = Optional.extract(details)
     max_file_bytes = Optional.extract(max_file_bytes)
 
     _extended_types = ['minimal', 'minimal+search', 'basic', 'full']
     if details not in _extended_types:
         ret_types = ','.join(_extended_types)
         raise JSONRPCError(f'ret_type must be one of %s, got {ret_types}', details)
     extended_info = False
     content = False
 
     if details == 'minimal':
         extended_info = False
 
     elif details == 'basic':
         extended_info = True
 
     elif details == 'full':
         extended_info = content = True
 
     file_path = safe_str(file_path)
     try:
         # check if repo is not empty by any chance, skip quicker if it is.
         _scm = repo.scm_instance()
         if _scm.is_empty():
             return None
 
         node = ScmModel().get_node(
             repo, commit_id, file_path, extended_info=extended_info,
             content=content, max_file_bytes=max_file_bytes, cache=cache)
 
     except NodeDoesNotExistError:
         raise JSONRPCError(
             f'There is no file in repo: `{repo.repo_name}` at path `{file_path}` for commit: `{commit_id}`')
     except Exception:
         log.exception("Exception occurred while trying to get repo %s file",
                       repo.repo_name)
         raise JSONRPCError(f'failed to get repo: `{repo.repo_name}` file at path {file_path}')
 
     return node
 
 
 @jsonrpc_method()
 def get_repo_fts_tree(request, apiuser, repoid, commit_id, root_path):
     """
     Returns a list of tree nodes for path at given revision. This api is built
     strictly for usage in full text search building, and shouldn't be consumed
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     repo_id = repo.repo_id
     cache_seconds = rhodecode.ConfigGet().get_int('rc_cache.cache_repo.expiration_time')
     cache_on = cache_seconds > 0
 
-    cache_namespace_uid = 'repo.{}'.format(repo_id)
+    cache_namespace_uid = f'repo.{repo_id}'
     rc_cache.get_or_create_region('cache_repo', cache_namespace_uid)
 
     def compute_fts_tree(cache_ver, repo_id, commit_id, root_path):
         return ScmModel().get_fts_data(repo_id, commit_id, root_path)
 
     try:
         # check if repo is not empty by any chance, skip quicker if it is.
         _scm = repo.scm_instance()
         if not _scm or _scm.is_empty():
             return []
     except RepositoryError:
         log.exception("Exception occurred while trying to get repo nodes")
         raise JSONRPCError(f'failed to get repo: `{repo.repo_name}` nodes')
 
     try:
         # we need to resolve commit_id to a FULL sha for cache to work correctly.
         # sending 'master' is a pointer that needs to be translated to current commit.
         commit_id = _scm.get_commit(commit_id=commit_id).raw_id
         log.debug(
             'Computing FTS REPO TREE for repo_id %s commit_id `%s` '
             'with caching: %s[TTL: %ss]' % (
                 repo_id, commit_id, cache_on, cache_seconds or 0))
 
         tree_files = compute_fts_tree(rc_cache.FILE_TREE_CACHE_VER, repo_id, commit_id, root_path)
 
         return tree_files
 
     except Exception:
         log.exception("Exception occurred while trying to get repo nodes")
         raise JSONRPCError('failed to get repo: `%s` nodes' % repo.repo_name)
 
 
 @jsonrpc_method()
 def get_repo_refs(request, apiuser, repoid):
     """
     Returns a dictionary of current references. It returns
     bookmarks, branches, closed_branches, and tags for given repository
 
     It's possible to specify ret_type to show only `files` or `dirs`.
 
     This command can only be run using an |authtoken| with admin rights,
     or users with at least read rights to |repos|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         "result": {
             "bookmarks": {
               "dev": "5611d30200f4040ba2ab4f3d64e5b06408a02188",
               "master": "367f590445081d8ec8c2ea0456e73ae1f1c3d6cf"
             },
             "branches": {
               "default": "5611d30200f4040ba2ab4f3d64e5b06408a02188",
               "stable": "367f590445081d8ec8c2ea0456e73ae1f1c3d6cf"
             },
             "branches_closed": {},
             "tags": {
               "tip": "5611d30200f4040ba2ab4f3d64e5b06408a02188",
               "v4.4.0": "1232313f9e6adac5ce5399c2a891dc1e72b79022",
               "v4.4.1": "cbb9f1d329ae5768379cdec55a62ebdd546c4e27",
               "v4.4.2": "24ffe44a27fcd1c5b6936144e176b9f6dd2f3a17",
             }
         }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write', 'repository.read',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         # check if repo is not empty by any chance, skip quicker if it is.
         vcs_instance = repo.scm_instance()
         refs = vcs_instance.refs()
         return refs
     except Exception:
         log.exception("Exception occurred while trying to get repo refs")
         raise JSONRPCError(
             'failed to get repo: `%s` references' % repo.repo_name
         )
 
 
 @jsonrpc_method()
 def create_repo(
         request, apiuser, repo_name, repo_type,
         owner=Optional(OAttr('apiuser')),
         description=Optional(''),
         private=Optional(False),
         clone_uri=Optional(None),
         push_uri=Optional(None),
         landing_rev=Optional(None),
         enable_statistics=Optional(False),
         enable_locking=Optional(False),
         enable_downloads=Optional(False),
         copy_permissions=Optional(False)):
     """
     Creates a repository.
 
     * If the repository name contains "/", repository will be created inside
       a repository group or nested repository groups
 
       For example "foo/bar/repo1" will create |repo| called "repo1" inside
       group "foo/bar". You have to have permissions to access and write to
       the last repository group ("bar" in this example)
 
     This command can only be run using an |authtoken| with at least
     permissions to create repositories, or write permissions to
     parent repository groups.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repo_name: Set the repository name.
     :type repo_name: str
     :param repo_type: Set the repository type; 'hg','git', or 'svn'.
     :type repo_type: str
     :param owner: user_id or username
     :type owner: Optional(str)
     :param description: Set the repository description.
     :type description: Optional(str)
     :param private: set repository as private
     :type private: bool
     :param clone_uri: set clone_uri
     :type clone_uri: str
     :param push_uri: set push_uri
     :type push_uri: str
     :param landing_rev: <rev_type>:<rev>, e.g branch:default, book:dev, rev:abcd
     :type landing_rev: str
     :param enable_locking:
     :type enable_locking: bool
     :param enable_downloads:
     :type enable_downloads: bool
     :param enable_statistics:
     :type enable_statistics: bool
     :param copy_permissions: Copy permission from group in which the
         repository is being created.
     :type copy_permissions: bool
 
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg": "Created new repository `<reponame>`",
                   "success": true,
                   "task": "<celery task id or None if done sync>"
                 }
         error:  null
 
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
          'failed to create repository `<repo_name>`'
       }
 
     """
 
     owner = validate_set_owner_permissions(apiuser, owner)
 
     description = Optional.extract(description)
     copy_permissions = Optional.extract(copy_permissions)
     clone_uri = Optional.extract(clone_uri)
     push_uri = Optional.extract(push_uri)
 
     defs = SettingsModel().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):
         repo_type = defs.get('repo_type')
     if isinstance(enable_statistics, Optional):
         enable_statistics = defs.get('repo_enable_statistics')
     if isinstance(enable_locking, Optional):
         enable_locking = defs.get('repo_enable_locking')
     if isinstance(enable_downloads, Optional):
         enable_downloads = defs.get('repo_enable_downloads')
 
     landing_ref, _label = ScmModel.backend_landing_ref(repo_type)
     ref_choices, _labels = ScmModel().get_repo_landing_revs(request.translate)
     ref_choices = list(set(ref_choices + [landing_ref]))
 
     landing_commit_ref = Optional.extract(landing_rev) or landing_ref
 
     schema = repo_schema.RepoSchema().bind(
         repo_type_options=rhodecode.BACKENDS.keys(),
         repo_ref_options=ref_choices,
         repo_type=repo_type,
         # user caller
         user=apiuser)
 
     try:
         schema_data = schema.deserialize(dict(
             repo_name=repo_name,
             repo_type=repo_type,
             repo_owner=owner.username,
             repo_description=description,
             repo_landing_commit_ref=landing_commit_ref,
             repo_clone_uri=clone_uri,
             repo_push_uri=push_uri,
             repo_private=private,
             repo_copy_permissions=copy_permissions,
             repo_enable_statistics=enable_statistics,
             repo_enable_downloads=enable_downloads,
             repo_enable_locking=enable_locking))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     try:
         data = {
             'owner': owner,
             'repo_name': schema_data['repo_group']['repo_name_without_group'],
             'repo_name_full': schema_data['repo_name'],
             'repo_group': schema_data['repo_group']['repo_group_id'],
             'repo_type': schema_data['repo_type'],
             'repo_description': schema_data['repo_description'],
             'repo_private': schema_data['repo_private'],
             'clone_uri': schema_data['repo_clone_uri'],
             'push_uri': schema_data['repo_push_uri'],
             'repo_landing_rev': schema_data['repo_landing_commit_ref'],
             'enable_statistics': schema_data['repo_enable_statistics'],
             'enable_locking': schema_data['repo_enable_locking'],
             'enable_downloads': schema_data['repo_enable_downloads'],
             'repo_copy_permissions': schema_data['repo_copy_permissions'],
         }
 
         task = RepoModel().create(form_data=data, cur_user=owner.user_id)
         task_id = get_task_id(task)
         # no commit, it's done in RepoModel, or async via celery
         return {
-            'msg': "Created new repository `%s`" % (schema_data['repo_name'],),
+            'msg': "Created new repository `{}`".format(schema_data['repo_name']),
             'success': True,  # cannot return the repo data here since fork
             # can be done async
             'task': task_id
         }
     except Exception:
         log.exception(
-            u"Exception while trying to create the repository %s",
+            "Exception while trying to create the repository %s",
             schema_data['repo_name'])
         raise JSONRPCError(
-            'failed to create repository `%s`' % (schema_data['repo_name'],))
+            'failed to create repository `{}`'.format(schema_data['repo_name']))
 
 
 @jsonrpc_method()
 def add_field_to_repo(request, apiuser, repoid, key, label=Optional(''),
                       description=Optional('')):
     """
     Adds an extra field to a repository.
 
     This command can only be run using an |authtoken| with at least
     write permissions to the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository id.
     :type repoid: str or int
     :param key: Create a unique field key for this repository.
     :type key: str
     :param label:
     :type label: Optional(str)
     :param description:
     :type description: Optional(str)
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     label = Optional.extract(label) or key
     description = Optional.extract(description)
 
     field = RepositoryField.get_by_key_name(key, repo)
     if field:
         raise JSONRPCError('Field with key '
                            '`%s` exists for repo `%s`' % (key, repoid))
 
     try:
         RepoModel().add_repo_field(repo, key, field_label=label,
                                    field_desc=description)
         Session().commit()
         return {
-            'msg': "Added new repository field `%s`" % (key,),
+            'msg': "Added new repository field `{}`".format(key),
             'success': True,
         }
     except Exception:
         log.exception("Exception occurred while trying to add field to repo")
         raise JSONRPCError(
-            'failed to create new field for repository `%s`' % (repoid,))
+            'failed to create new field for repository `{}`'.format(repoid))
 
 
 @jsonrpc_method()
 def remove_field_from_repo(request, apiuser, repoid, key):
     """
     Removes an extra field from a repository.
 
     This command can only be run using an |authtoken| with at least
     write permissions to the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param key: Set the unique field key for this repository.
     :type key: str
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     field = RepositoryField.get_by_key_name(key, repo)
     if not field:
         raise JSONRPCError('Field with key `%s` does not '
                            'exists for repo `%s`' % (key, repoid))
 
     try:
         RepoModel().delete_repo_field(repo, field_key=key)
         Session().commit()
         return {
-            'msg': "Deleted repository field `%s`" % (key,),
+            'msg': "Deleted repository field `{}`".format(key),
             'success': True,
         }
     except Exception:
         log.exception(
             "Exception occurred while trying to delete field from repo")
         raise JSONRPCError(
-            'failed to delete field for repository `%s`' % (repoid,))
+            'failed to delete field for repository `{}`'.format(repoid))
 
 
 @jsonrpc_method()
 def update_repo(
         request, apiuser, repoid, repo_name=Optional(None),
         owner=Optional(OAttr('apiuser')), description=Optional(''),
         private=Optional(False),
         clone_uri=Optional(None), push_uri=Optional(None),
         landing_rev=Optional(None), fork_of=Optional(None),
         enable_statistics=Optional(False),
         enable_locking=Optional(False),
         enable_downloads=Optional(False), fields=Optional('')):
-    """
+    r"""
     Updates a repository with the given information.
 
     This command can only be run using an |authtoken| with at least
     admin permissions to the |repo|.
 
     * If the repository name contains "/", repository will be updated
       accordingly with a repository group or nested repository groups
 
       For example repoid=repo-test name="foo/bar/repo-test" will update |repo|
       called "repo-test" and place it inside group "foo/bar".
       You have to have permissions to access and write to the last repository
       group ("bar" in this example)
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: repository name or repository ID.
     :type repoid: str or int
     :param repo_name: Update the |repo| name, including the
         repository group it's in.
     :type repo_name: str
     :param owner: Set the |repo| owner.
     :type owner: str
     :param fork_of: Set the |repo| as fork of another |repo|.
     :type fork_of: str
     :param description: Update the |repo| description.
     :type description: str
     :param private: Set the |repo| as private. (True | False)
     :type private: bool
     :param clone_uri: Update the |repo| clone URI.
     :type clone_uri: str
     :param landing_rev: Set the |repo| landing revision. e.g branch:default, book:dev, rev:abcd
     :type landing_rev: str
     :param enable_statistics: Enable statistics on the |repo|, (True | False).
     :type enable_statistics: bool
     :param enable_locking: Enable |repo| locking.
     :type enable_locking: bool
     :param enable_downloads: Enable downloads from the |repo|, (True | False).
     :type enable_downloads: bool
     :param fields: Add extra fields to the |repo|. Use the following
         example format: ``field_key=field_val,field_key2=fieldval2``.
         Escape ', ' with \,
     :type fields: str
     """
 
     repo = get_repo_or_error(repoid)
 
     include_secrets = False
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
     else:
         include_secrets = True
 
     updates = dict(
         repo_name=repo_name
         if not isinstance(repo_name, Optional) else repo.repo_name,
 
         fork_id=fork_of
         if not isinstance(fork_of, Optional) else repo.fork.repo_name if repo.fork else None,
 
         user=owner
         if not isinstance(owner, Optional) else repo.user.username,
 
         repo_description=description
         if not isinstance(description, Optional) else repo.description,
 
         repo_private=private
         if not isinstance(private, Optional) else repo.private,
 
         clone_uri=clone_uri
         if not isinstance(clone_uri, Optional) else repo.clone_uri,
 
         push_uri=push_uri
         if not isinstance(push_uri, Optional) else repo.push_uri,
 
         repo_landing_rev=landing_rev
         if not isinstance(landing_rev, Optional) else repo._landing_revision,
 
         repo_enable_statistics=enable_statistics
         if not isinstance(enable_statistics, Optional) else repo.enable_statistics,
 
         repo_enable_locking=enable_locking
         if not isinstance(enable_locking, Optional) else repo.enable_locking,
 
         repo_enable_downloads=enable_downloads
         if not isinstance(enable_downloads, Optional) else repo.enable_downloads)
 
     landing_ref, _label = ScmModel.backend_landing_ref(repo.repo_type)
     ref_choices, _labels = ScmModel().get_repo_landing_revs(
         request.translate, repo=repo)
     ref_choices = list(set(ref_choices + [landing_ref]))
 
     old_values = repo.get_api_data()
     repo_type = repo.repo_type
     schema = repo_schema.RepoSchema().bind(
         repo_type_options=rhodecode.BACKENDS.keys(),
         repo_ref_options=ref_choices,
         repo_type=repo_type,
         # user caller
         user=apiuser,
         old_values=old_values)
     try:
         schema_data = schema.deserialize(dict(
             # we save old value, users cannot change type
             repo_type=repo_type,
 
             repo_name=updates['repo_name'],
             repo_owner=updates['user'],
             repo_description=updates['repo_description'],
             repo_clone_uri=updates['clone_uri'],
             repo_push_uri=updates['push_uri'],
             repo_fork_of=updates['fork_id'],
             repo_private=updates['repo_private'],
             repo_landing_commit_ref=updates['repo_landing_rev'],
             repo_enable_statistics=updates['repo_enable_statistics'],
             repo_enable_downloads=updates['repo_enable_downloads'],
             repo_enable_locking=updates['repo_enable_locking']))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     # save validated data back into the updates dict
     validated_updates = dict(
         repo_name=schema_data['repo_group']['repo_name_without_group'],
         repo_group=schema_data['repo_group']['repo_group_id'],
 
         user=schema_data['repo_owner'],
         repo_description=schema_data['repo_description'],
         repo_private=schema_data['repo_private'],
         clone_uri=schema_data['repo_clone_uri'],
         push_uri=schema_data['repo_push_uri'],
         repo_landing_rev=schema_data['repo_landing_commit_ref'],
         repo_enable_statistics=schema_data['repo_enable_statistics'],
         repo_enable_locking=schema_data['repo_enable_locking'],
         repo_enable_downloads=schema_data['repo_enable_downloads'],
     )
 
     if schema_data['repo_fork_of']:
         fork_repo = get_repo_or_error(schema_data['repo_fork_of'])
         validated_updates['fork_id'] = fork_repo.repo_id
 
     # extra fields
     fields = parse_args(Optional.extract(fields), key_prefix='ex_')
     if fields:
         validated_updates.update(fields)
 
     try:
         RepoModel().update(repo, **validated_updates)
         audit_logger.store_api(
             'repo.edit', action_data={'old_data': old_values},
             user=apiuser, repo=repo)
         Session().commit()
         return {
-            'msg': 'updated repo ID:%s %s' % (repo.repo_id, repo.repo_name),
+            'msg': 'updated repo ID:{} {}'.format(repo.repo_id, repo.repo_name),
             'repository': repo.get_api_data(include_secrets=include_secrets)
         }
     except Exception:
         log.exception(
-            u"Exception while trying to update the repository %s",
+            "Exception while trying to update the repository %s",
             repoid)
         raise JSONRPCError('failed to update repo `%s`' % repoid)
 
 
 @jsonrpc_method()
 def fork_repo(request, apiuser, repoid, fork_name,
               owner=Optional(OAttr('apiuser')),
               description=Optional(''),
               private=Optional(False),
               clone_uri=Optional(None),
               landing_rev=Optional(None),
               copy_permissions=Optional(False)):
     """
     Creates a fork of the specified |repo|.
 
     * If the fork_name contains "/", fork will be created inside
       a repository group or nested repository groups
 
       For example "foo/bar/fork-repo" will create fork called "fork-repo"
       inside group "foo/bar". You have to have permissions to access and
       write to the last repository group ("bar" in this example)
 
     This command can only be run using an |authtoken| with minimum
     read permissions of the forked repo, create fork permissions for an user.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set repository name or repository ID.
     :type repoid: str or int
     :param fork_name: Set the fork name, including it's repository group membership.
     :type fork_name: str
     :param owner: Set the fork owner.
     :type owner: str
     :param description: Set the fork description.
     :type description: str
     :param copy_permissions: Copy permissions from parent |repo|. The
         default is False.
     :type copy_permissions: bool
     :param private: Make the fork private. The default is False.
     :type private: bool
     :param landing_rev: Set the landing revision. E.g branch:default, book:dev, rev:abcd
 
     Example output:
 
     .. code-block:: bash
 
         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>"
                   }
 
     Example error output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg": "Created fork of `<reponame>` as `<forkname>`",
                   "success": true,
                   "task": "<celery task id or None if done sync>"
                 }
         error:  null
 
     """
 
     repo = get_repo_or_error(repoid)
     repo_name = repo.repo_name
 
     if not has_superadmin_permission(apiuser):
         # check if we have at least read permission for
         # this repo that we fork !
         _perms = ('repository.admin', 'repository.write', 'repository.read')
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
         # check if the regular user has at least fork permissions as well
         if not HasPermissionAnyApi(PermissionModel.FORKING_ENABLED)(user=apiuser):
             raise JSONRPCForbidden()
 
     # check if user can set owner parameter
     owner = validate_set_owner_permissions(apiuser, owner)
 
     description = Optional.extract(description)
     copy_permissions = Optional.extract(copy_permissions)
     clone_uri = Optional.extract(clone_uri)
 
     landing_ref, _label = ScmModel.backend_landing_ref(repo.repo_type)
     ref_choices, _labels = ScmModel().get_repo_landing_revs(request.translate)
     ref_choices = list(set(ref_choices + [landing_ref]))
     landing_commit_ref = Optional.extract(landing_rev) or landing_ref
 
     private = Optional.extract(private)
 
     schema = repo_schema.RepoSchema().bind(
         repo_type_options=rhodecode.BACKENDS.keys(),
         repo_ref_options=ref_choices,
         repo_type=repo.repo_type,
         # user caller
         user=apiuser)
 
     try:
         schema_data = schema.deserialize(dict(
             repo_name=fork_name,
             repo_type=repo.repo_type,
             repo_owner=owner.username,
             repo_description=description,
             repo_landing_commit_ref=landing_commit_ref,
             repo_clone_uri=clone_uri,
             repo_private=private,
             repo_copy_permissions=copy_permissions))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     try:
         data = {
             'fork_parent_id': repo.repo_id,
 
             'repo_name': schema_data['repo_group']['repo_name_without_group'],
             'repo_name_full': schema_data['repo_name'],
             'repo_group': schema_data['repo_group']['repo_group_id'],
             'repo_type': schema_data['repo_type'],
             'description': schema_data['repo_description'],
             'private': schema_data['repo_private'],
             'copy_permissions': schema_data['repo_copy_permissions'],
             'landing_rev': schema_data['repo_landing_commit_ref'],
         }
 
         task = RepoModel().create_fork(data, cur_user=owner.user_id)
         # no commit, it's done in RepoModel, or async via celery
         task_id = get_task_id(task)
 
         return {
-            'msg': 'Created fork of `%s` as `%s`' % (
+            'msg': 'Created fork of `{}` as `{}`'.format(
                 repo.repo_name, schema_data['repo_name']),
             'success': True,  # cannot return the repo data here since fork
             # can be done async
             'task': task_id
         }
     except Exception:
         log.exception(
-            u"Exception while trying to create fork %s",
+            "Exception while trying to create fork %s",
             schema_data['repo_name'])
         raise JSONRPCError(
-            'failed to fork repository `%s` as `%s`' % (
+            'failed to fork repository `{}` as `{}`'.format(
                 repo_name, schema_data['repo_name']))
 
 
 @jsonrpc_method()
 def delete_repo(request, apiuser, repoid, forks=Optional('')):
     """
     Deletes a repository.
 
     * When the `forks` parameter is set it's possible to detach or delete
       forks of deleted repository.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param forks: Set to `detach` or `delete` forks from the |repo|.
     :type forks: Optional(str)
 
     Example error output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg": "Deleted repository `<reponame>`",
                   "success": true
                 }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     repo_name = repo.repo_name
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         handle_forks = Optional.extract(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,)
             )
         old_data = repo.get_api_data()
         RepoModel().delete(repo, forks=forks)
 
         repo = audit_logger.RepoWrap(repo_id=None,
                                      repo_name=repo.repo_name)
 
         audit_logger.store_api(
             'repo.delete', action_data={'old_data': old_data},
             user=apiuser, repo=repo)
 
         ScmModel().mark_for_invalidation(repo_name, delete=True)
         Session().commit()
         return {
-            'msg': 'Deleted repository `%s`%s' % (repo_name, _forks_msg),
+            'msg': 'Deleted repository `{}`{}'.format(repo_name, _forks_msg),
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying to delete repo")
         raise JSONRPCError(
-            'failed to delete repository `%s`' % (repo_name,)
+            'failed to delete repository `{}`'.format(repo_name)
         )
 
 
 #TODO: marcink, change name ?
 @jsonrpc_method()
 def invalidate_cache(request, apiuser, repoid, delete_keys=Optional(False)):
     """
     Invalidates the cache for the specified repository.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Sets the repository name or repository ID.
     :type repoid: str or int
     :param delete_keys: This deletes the invalidated keys instead of
         just flagging them.
     :type delete_keys: Optional(``True`` | ``False``)
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'msg': Cache for repository `<repository name>` was invalidated,
         'repository': <repository name>
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error : {
          'Error occurred during cache invalidation action'
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin', 'repository.write',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     delete = Optional.extract(delete_keys)
     try:
         ScmModel().mark_for_invalidation(repo.repo_name, delete=delete)
         return {
-            'msg': 'Cache for repository `%s` was invalidated' % (repoid,),
+            'msg': 'Cache for repository `{}` was invalidated'.format(repoid),
             'repository': repo.repo_name
         }
     except Exception:
         log.exception(
             "Exception occurred while trying to invalidate repo cache")
         raise JSONRPCError(
             'Error occurred during cache invalidation action'
         )
 
 
 #TODO: marcink, change name ?
 @jsonrpc_method()
 def lock(request, apiuser, repoid, locked=Optional(None),
          userid=Optional(OAttr('apiuser'))):
     """
     Sets the lock state of the specified |repo| by the given user.
     From more information, see :ref:`repo-locking`.
 
     * If the ``userid`` option is not set, the repository is locked to the
       user who called the method.
     * If the ``locked`` parameter is not set, the current lock state of the
       repository is displayed.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Sets the repository name or repository ID.
     :type repoid: str or int
     :param locked: Sets the lock state.
     :type locked: Optional(``True`` | ``False``)
     :param userid: Set the repository lock to this user.
     :type userid: Optional(str or int)
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'repo': '<reponame>',
         'locked': <bool: lock state>,
         'locked_since': <int: lock timestamp>,
         'locked_by': <username of person who made the lock>,
         'lock_reason': <str: reason for locking>,
         'lock_state_changed': <bool: True if lock state has been changed in this request>,
         'msg': 'Repo `<reponame>` locked by `<username>` on <timestamp>.'
         or
         'msg': 'Repo `<repository name>` not locked.'
         or
         'msg': 'User `<user name>` set lock state for repo `<repository name>` to `<new lock state>`'
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         'Error occurred locking repository `<reponame>`'
       }
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         # check if we have at least write permission for this repo !
         _perms = ('repository.admin', 'repository.write',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
         # make sure normal user does not pass someone else userid,
         # he is not allowed to do that
         if not isinstance(userid, Optional) and userid != apiuser.user_id:
             raise JSONRPCError('userid is not the same as your user')
 
     if isinstance(userid, Optional):
         userid = apiuser.user_id
 
     user = get_user_or_error(userid)
 
     if isinstance(locked, Optional):
         lockobj = repo.locked
 
         if lockobj[0] is None:
             _d = {
                 'repo': repo.repo_name,
                 'locked': False,
                 'locked_since': None,
                 'locked_by': None,
                 'lock_reason': None,
                 'lock_state_changed': False,
                 'msg': 'Repo `%s` not locked.' % repo.repo_name
             }
             return _d
         else:
             _user_id, _time, _reason = lockobj
             lock_user = get_user_or_error(userid)
             _d = {
                 'repo': repo.repo_name,
                 'locked': True,
                 'locked_since': _time,
                 'locked_by': lock_user.username,
                 'lock_reason': _reason,
                 'lock_state_changed': False,
                 'msg': ('Repo `%s` locked by `%s` on `%s`.'
                         % (repo.repo_name, lock_user.username,
                            json.dumps(time_to_datetime(_time))))
             }
             return _d
 
     # force locked state through a flag
     else:
         locked = str2bool(locked)
         lock_reason = Repository.LOCK_API
         try:
             if locked:
                 lock_time = time.time()
                 Repository.lock(repo, user.user_id, lock_time, lock_reason)
             else:
                 lock_time = None
                 Repository.unlock(repo)
             _d = {
                 'repo': repo.repo_name,
                 'locked': locked,
                 'locked_since': lock_time,
                 'locked_by': user.username,
                 'lock_reason': lock_reason,
                 'lock_state_changed': True,
                 'msg': ('User `%s` set lock state for repo `%s` to `%s`'
                         % (user.username, repo.repo_name, locked))
             }
             return _d
         except Exception:
             log.exception(
                 "Exception occurred while trying to lock repository")
             raise JSONRPCError(
                 'Error occurred locking repository `%s`' % repo.repo_name
             )
 
 
 @jsonrpc_method()
 def comment_commit(
         request, apiuser, repoid, commit_id, message, status=Optional(None),
         comment_type=Optional(ChangesetComment.COMMENT_TYPE_NOTE),
         resolves_comment_id=Optional(None), extra_recipients=Optional([]),
         userid=Optional(OAttr('apiuser')), send_email=Optional(True)):
     """
     Set a commit comment, and optionally change the status of the commit.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param commit_id: Specify the commit_id for which to set a comment.
     :type commit_id: str
     :param message: The comment text.
     :type message: str
     :param status: (**Optional**) status of commit, one of: 'not_reviewed',
         'approved', 'rejected', 'under_review'
     :type status: str
     :param comment_type: Comment type, one of: 'note', 'todo'
     :type comment_type: Optional(str), default: 'note'
     :param resolves_comment_id: id of comment which this one will resolve
     :type resolves_comment_id: Optional(int)
     :param extra_recipients: list of user ids or usernames to add
         notifications for this comment. Acts like a CC for notification
     :type extra_recipients: Optional(list)
     :param userid: Set the user name of the comment creator.
     :type userid: Optional(str or int)
     :param send_email: Define if this comment should also send email notification
     :type send_email: Optional(bool)
 
     Example error output:
 
     .. code-block:: bash
 
         {
             "id" : <id_given_in_input>,
             "result" : {
                 "msg": "Commented on commit `<commit_id>` for repository `<repoid>`",
                 "status_change": null or <status>,
                 "success": true
             },
             "error" :  null
         }
 
     """
     _ = request.translate
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.read', 'repository.write', 'repository.admin')
         validate_repo_permissions(apiuser, repoid, repo, _perms)
     db_repo_name = repo.repo_name
 
     try:
         commit = repo.scm_instance().get_commit(commit_id=commit_id)
         commit_id = commit.raw_id
     except Exception as e:
         log.exception('Failed to fetch commit')
         raise JSONRPCError(safe_str(e))
 
     if isinstance(userid, Optional):
         userid = apiuser.user_id
 
     user = get_user_or_error(userid)
     status = Optional.extract(status)
     comment_type = Optional.extract(comment_type)
     resolves_comment_id = Optional.extract(resolves_comment_id)
     extra_recipients = Optional.extract(extra_recipients)
     send_email = Optional.extract(send_email, binary=True)
 
     allowed_statuses = [x[0] for x in ChangesetStatus.STATUSES]
     if status and status not in allowed_statuses:
         raise JSONRPCError('Bad status, must be on '
                            'of %s got %s' % (allowed_statuses, status,))
 
     if resolves_comment_id:
         comment = ChangesetComment.get(resolves_comment_id)
         if not comment:
             raise JSONRPCError(
                 'Invalid resolves_comment_id `%s` for this commit.'
                 % resolves_comment_id)
         if comment.comment_type != ChangesetComment.COMMENT_TYPE_TODO:
             raise JSONRPCError(
                 'Comment `%s` is wrong type for setting status to resolved.'
                 % resolves_comment_id)
 
     try:
         rc_config = SettingsModel().get_all_settings()
         renderer = rc_config.get('rhodecode_markup_renderer', 'rst')
         status_change_label = ChangesetStatus.get_status_lbl(status)
         comment = CommentsModel().create(
             message, repo, user, commit_id=commit_id,
             status_change=status_change_label,
             status_change_type=status,
             renderer=renderer,
             comment_type=comment_type,
             resolves_comment_id=resolves_comment_id,
             auth_user=apiuser,
             extra_recipients=extra_recipients,
             send_email=send_email
         )
         is_inline = comment.is_inline
 
         if status:
             # also do a status change
             try:
                 ChangesetStatusModel().set_status(
                     repo, status, user, comment, revision=commit_id,
                     dont_allow_on_closed_pull_request=True
                 )
             except StatusChangeOnClosedPullRequestError:
                 log.exception(
                     "Exception occurred while trying to change repo commit status")
                 msg = ('Changing status on a commit associated with '
                        'a closed pull request is not allowed')
                 raise JSONRPCError(msg)
 
         CommentsModel().trigger_commit_comment_hook(
             repo, apiuser, 'create',
             data={'comment': comment, 'commit': commit})
 
         Session().commit()
 
         comment_broadcast_channel = channelstream.comment_channel(
             db_repo_name, commit_obj=commit)
 
         comment_data = {'comment': comment, 'comment_id': comment.comment_id}
         comment_type = 'inline' if is_inline else 'general'
         channelstream.comment_channelstream_push(
             request, comment_broadcast_channel, apiuser,
             _('posted a new {} comment').format(comment_type),
             comment_data=comment_data)
 
         return {
             'msg': (
-                'Commented on commit `%s` for repository `%s`' % (
+                'Commented on commit `{}` for repository `{}`'.format(
                     comment.revision, repo.repo_name)),
             'status_change': status,
             'success': True,
         }
     except JSONRPCError:
         # catch any inside errors, and re-raise them to prevent from
         # below global catch to silence them
         raise
     except Exception:
         log.exception("Exception occurred while trying to comment on commit")
         raise JSONRPCError(
-            'failed to set comment on repository `%s`' % (repo.repo_name,)
+            'failed to set comment on repository `{}`'.format(repo.repo_name)
         )
 
 
 @jsonrpc_method()
 def get_repo_comments(request, apiuser, repoid,
                       commit_id=Optional(None), comment_type=Optional(None),
                       userid=Optional(None)):
     """
     Get all comments for a repository
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param commit_id: Optionally filter the comments by the commit_id
     :type commit_id: Optional(str), default: None
     :param comment_type: Optionally filter the comments by the comment_type
         one of: 'note', 'todo'
     :type comment_type: Optional(str), default: None
     :param userid: Optionally filter the comments by the author of comment
     :type userid: Optional(str or int), Default: None
 
     Example error output:
 
     .. code-block:: bash
 
         {
             "id" : <id_given_in_input>,
             "result" : [
                 {
                   "comment_author": <USER_DETAILS>,
                   "comment_created_on": "2017-02-01T14:38:16.309",
                   "comment_f_path": "file.txt",
                   "comment_id": 282,
                   "comment_lineno": "n1",
                   "comment_resolved_by": null,
                   "comment_status": [],
                   "comment_text": "This file needs a header",
                   "comment_type": "todo",
                   "comment_last_version: 0
                 }
             ],
             "error" :  null
         }
 
     """
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.read', 'repository.write', 'repository.admin')
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     commit_id = Optional.extract(commit_id)
 
     userid = Optional.extract(userid)
     if userid:
         user = get_user_or_error(userid)
     else:
         user = None
 
     comment_type = Optional.extract(comment_type)
     if comment_type and comment_type not in ChangesetComment.COMMENT_TYPES:
         raise JSONRPCError(
                 'comment_type must be one of `{}` got {}'.format(
                     ChangesetComment.COMMENT_TYPES, comment_type)
             )
 
     comments = CommentsModel().get_repository_comments(
         repo=repo, comment_type=comment_type, user=user, commit_id=commit_id)
     return comments
 
 
 @jsonrpc_method()
 def get_comment(request, apiuser, comment_id):
     """
     Get single comment from repository or pull_request
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param comment_id: comment id found in the URL of comment
     :type comment_id: str or int
 
     Example error output:
 
     .. code-block:: bash
 
         {
             "id" : <id_given_in_input>,
             "result" : {
                   "comment_author": <USER_DETAILS>,
                   "comment_created_on": "2017-02-01T14:38:16.309",
                   "comment_f_path": "file.txt",
                   "comment_id": 282,
                   "comment_lineno": "n1",
                   "comment_resolved_by": null,
                   "comment_status": [],
                   "comment_text": "This file needs a header",
                   "comment_type": "todo",
                   "comment_last_version: 0
             },
             "error" :  null
         }
 
     """
 
     comment = ChangesetComment.get(comment_id)
     if not comment:
-        raise JSONRPCError('comment `%s` does not exist' % (comment_id,))
+        raise JSONRPCError('comment `{}` does not exist'.format(comment_id))
 
     perms = ('repository.read', 'repository.write', 'repository.admin')
     has_comment_perm = HasRepoPermissionAnyApi(*perms)\
         (user=apiuser, repo_name=comment.repo.repo_name)
 
     if not has_comment_perm:
-        raise JSONRPCError('comment `%s` does not exist' % (comment_id,))
+        raise JSONRPCError('comment `{}` does not exist'.format(comment_id))
 
     return comment
 
 
 @jsonrpc_method()
 def edit_comment(request, apiuser, message, comment_id, version,
                  userid=Optional(OAttr('apiuser'))):
     """
     Edit comment on the pull request or commit,
     specified by the `comment_id` and version. Initially version should be 0
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param comment_id: Specify the comment_id for editing
     :type comment_id: int
     :param version: version of the comment that will be created, starts from 0
     :type version: int
     :param message: The text content of the comment.
     :type message: str
     :param userid: Comment on the pull request as this user
     :type userid: Optional(str or int)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : {
             "comment": "<comment data>",
             "version": "<Integer>",
         },
         error :  null
     """
 
     auth_user = apiuser
     comment = ChangesetComment.get(comment_id)
     if not comment:
-        raise JSONRPCError('comment `%s` does not exist' % (comment_id,))
+        raise JSONRPCError('comment `{}` does not exist'.format(comment_id))
 
     is_super_admin = has_superadmin_permission(apiuser)
     is_repo_admin = HasRepoPermissionAnyApi('repository.admin')\
         (user=apiuser, repo_name=comment.repo.repo_name)
 
     if not isinstance(userid, Optional):
         if is_super_admin or is_repo_admin:
             apiuser = get_user_or_error(userid)
             auth_user = apiuser.AuthUser()
         else:
             raise JSONRPCError('userid is not the same as your user')
 
     comment_author = comment.author.user_id == auth_user.user_id
 
     if comment.immutable:
         raise JSONRPCError("Immutable comment cannot be edited")
 
     if not (is_super_admin or is_repo_admin or comment_author):
         raise JSONRPCError("you don't have access to edit this comment")
 
     try:
         comment_history = CommentsModel().edit(
             comment_id=comment_id,
             text=message,
             auth_user=auth_user,
             version=version,
         )
         Session().commit()
     except CommentVersionMismatch:
         raise JSONRPCError(
-            'comment ({}) version ({}) mismatch'.format(comment_id, version)
+            f'comment ({comment_id}) version ({version}) mismatch'
         )
     if not comment_history and not message:
         raise JSONRPCError(
-            "comment ({}) can't be changed with empty string".format(comment_id)
+            f"comment ({comment_id}) can't be changed with empty string"
         )
 
     if comment.pull_request:
         pull_request = comment.pull_request
         PullRequestModel().trigger_pull_request_hook(
             pull_request, apiuser, 'comment_edit',
             data={'comment': comment})
     else:
         db_repo = comment.repo
         commit_id = comment.revision
         commit = db_repo.get_commit(commit_id)
         CommentsModel().trigger_commit_comment_hook(
             db_repo, apiuser, 'edit',
             data={'comment': comment, 'commit': commit})
 
     data = {
         'comment': comment,
         'version': comment_history.version if comment_history else None,
     }
     return data
 
 
 # TODO(marcink): write this with all required logic for deleting a comments in PR or commits
 # @jsonrpc_method()
 # def delete_comment(request, apiuser, comment_id):
 #     auth_user = apiuser
 #
 #     comment = ChangesetComment.get(comment_id)
 #     if not comment:
 #         raise JSONRPCError('comment `%s` does not exist' % (comment_id,))
 #
 #     is_super_admin = has_superadmin_permission(apiuser)
 #     is_repo_admin = HasRepoPermissionAnyApi('repository.admin')\
 #         (user=apiuser, repo_name=comment.repo.repo_name)
 #
 #     comment_author = comment.author.user_id == auth_user.user_id
 #     if not (comment.immutable is False and (is_super_admin or is_repo_admin) or comment_author):
 #         raise JSONRPCError("you don't have access to edit this comment")
 
 @jsonrpc_method()
 def grant_user_permission(request, apiuser, repoid, userid, perm):
     """
     Grant permissions for the specified user on the given repository,
     or update existing permissions if found.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param userid: Set the user name.
     :type userid: str
     :param perm: Set the user permissions, using the following format
         ``(repository.(none|read|write|admin))``
     :type perm: str
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Granted perm: `<perm>` for user: `<username>` in repo: `<reponame>`",
                   "success": true
                 }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     user = get_user_or_error(userid)
     perm = get_perm_or_error(perm)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     perm_additions = [[user.user_id, perm.permission_name, "user"]]
     try:
         changes = RepoModel().update_permissions(
             repo=repo, perm_additions=perm_additions, cur_user=apiuser)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo.edit.permissions', action_data=action_data, user=apiuser, repo=repo)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
-            'msg': 'Granted perm: `%s` for user: `%s` in repo: `%s`' % (
+            'msg': 'Granted perm: `{}` for user: `{}` in repo: `{}`'.format(
                 perm.permission_name, user.username, repo.repo_name
             ),
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying edit permissions for repo")
         raise JSONRPCError(
-            'failed to edit permission for user: `%s` in repo: `%s`' % (
+            'failed to edit permission for user: `{}` in repo: `{}`'.format(
                 userid, repoid
             )
         )
 
 
 @jsonrpc_method()
 def revoke_user_permission(request, apiuser, repoid, userid):
     """
     Revoke permission for a user on the specified repository.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param userid: Set the user name of revoked user.
     :type userid: str or int
 
     Example error output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Revoked perm for user: `<username>` in repo: `<reponame>`",
                   "success": true
                 }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     user = get_user_or_error(userid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     perm_deletions = [[user.user_id, None, "user"]]
     try:
         changes = RepoModel().update_permissions(
             repo=repo, perm_deletions=perm_deletions, cur_user=user)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo.edit.permissions', action_data=action_data, user=apiuser, repo=repo)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
-            'msg': 'Revoked perm for user: `%s` in repo: `%s`' % (
+            'msg': 'Revoked perm for user: `{}` in repo: `{}`'.format(
                 user.username, repo.repo_name
             ),
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying revoke permissions to repo")
         raise JSONRPCError(
-            'failed to edit permission for user: `%s` in repo: `%s`' % (
+            'failed to edit permission for user: `{}` in repo: `{}`'.format(
                 userid, repoid
             )
         )
 
 
 @jsonrpc_method()
 def grant_user_group_permission(request, apiuser, repoid, usergroupid, perm):
     """
     Grant permission for a user group on the specified repository,
     or update existing permissions.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param usergroupid: Specify the ID of the user group.
     :type usergroupid: str or int
     :param perm: Set the user group permissions using the following
         format: (repository.(none|read|write|admin))
     :type perm: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg" : "Granted perm: `<perm>` for group: `<usersgroupname>` in repo: `<reponame>`",
         "success": true
 
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to edit permission for user group: `<usergroup>` in repo `<repo>`'
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     perm = get_perm_or_error(perm)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     user_group = get_user_group_or_error(usergroupid)
     if not has_superadmin_permission(apiuser):
         # check if we have at least read permission for this user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
 
     perm_additions = [[user_group.users_group_id, perm.permission_name, "user_group"]]
     try:
         changes = RepoModel().update_permissions(
             repo=repo, perm_additions=perm_additions, cur_user=apiuser)
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo.edit.permissions', action_data=action_data, user=apiuser, repo=repo)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             'msg': 'Granted perm: `%s` for user group: `%s` in '
                    'repo: `%s`' % (
                        perm.permission_name, user_group.users_group_name,
                        repo.repo_name
                    ),
             'success': True
         }
     except Exception:
         log.exception(
             "Exception occurred while trying change permission on repo")
         raise JSONRPCError(
             'failed to edit permission for user group: `%s` in '
             'repo: `%s`' % (
                 usergroupid, repo.repo_name
             )
         )
 
 
 @jsonrpc_method()
 def revoke_user_group_permission(request, apiuser, repoid, usergroupid):
     """
     Revoke the permissions of a user group on a given repository.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo|.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: Set the repository name or repository ID.
     :type repoid: str or int
     :param usergroupid: Specify the user group ID.
     :type usergroupid: str or int
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Revoked perm for group: `<usersgroupname>` in repo: `<reponame>`",
                   "success": true
                 }
         error:  null
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     user_group = get_user_group_or_error(usergroupid)
     if not has_superadmin_permission(apiuser):
         # check if we have at least read permission for this user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
 
     perm_deletions = [[user_group.users_group_id, None, "user_group"]]
     try:
         changes = RepoModel().update_permissions(
             repo=repo, perm_deletions=perm_deletions, cur_user=apiuser)
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo.edit.permissions', action_data=action_data, user=apiuser, repo=repo)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
-            'msg': 'Revoked perm for user group: `%s` in repo: `%s`' % (
+            'msg': 'Revoked perm for user group: `{}` in repo: `{}`'.format(
                 user_group.users_group_name, repo.repo_name
             ),
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying revoke "
                       "user group permission on repo")
         raise JSONRPCError(
             'failed to edit permission for user group: `%s` in '
             'repo: `%s`' % (
                 user_group.users_group_name, repo.repo_name
             )
         )
 
 
 @jsonrpc_method()
 def pull(request, apiuser, repoid, remote_uri=Optional(None)):
     """
     Triggers a pull on the given repository from a remote location. You
     can use this to keep remote repositories up-to-date.
 
     This command can only be run using an |authtoken| with admin
     rights to the specified repository. For more information,
     see :ref:`config-token-ref`.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param remote_uri: Optional remote URI to pass in for pull
     :type remote_uri: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "Pulled from url `<remote_url>` on repo `<repository name>`"
         "repository": "<repository name>"
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "Unable to push changes from `<remote_url>`"
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     remote_uri = Optional.extract(remote_uri)
     remote_uri_display = remote_uri or repo.clone_uri_hidden
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         ScmModel().pull_changes(
             repo.repo_name, apiuser.username, remote_uri=remote_uri)
         return {
-            'msg': 'Pulled from url `%s` on repo `%s`' % (
+            'msg': 'Pulled from url `{}` on repo `{}`'.format(
                 remote_uri_display, repo.repo_name),
             'repository': repo.repo_name
         }
     except Exception:
         log.exception("Exception occurred while trying to "
                       "pull changes from remote location")
         raise JSONRPCError(
             'Unable to pull changes from `%s`' % remote_uri_display
         )
 
 
 @jsonrpc_method()
 def strip(request, apiuser, repoid, revision, branch):
     """
     Strips the given revision from the specified repository.
 
     * This will remove the revision and all of its decendants.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
     :param revision: The revision you wish to strip.
     :type revision: str
     :param branch: The branch from which to strip the revision.
     :type branch: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "'Stripped commit <commit_hash> from repo `<repository name>`'"
         "repository": "<repository name>"
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "Unable to strip commit <commit_hash> from repo `<repository name>`"
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         ScmModel().strip(repo, revision, branch)
         audit_logger.store_api(
             'repo.commit.strip', action_data={'commit_id': revision},
             repo=repo,
             user=apiuser, commit=True)
 
         return {
-            'msg': 'Stripped commit %s from repo `%s`' % (
+            'msg': 'Stripped commit {} from repo `{}`'.format(
                 revision, repo.repo_name),
             'repository': repo.repo_name
         }
     except Exception:
         log.exception("Exception while trying to strip")
         raise JSONRPCError(
-            'Unable to strip commit %s from repo `%s`' % (
+            'Unable to strip commit {} from repo `{}`'.format(
                 revision, repo.repo_name)
         )
 
 
 @jsonrpc_method()
 def get_repo_settings(request, apiuser, repoid, key=Optional(None)):
     """
     Returns all settings for a repository. If key is given it only returns the
     setting identified by the key or null.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository id.
     :type repoid: str or int
     :param key: Key of the setting to return.
     :type: key: Optional(str)
 
     Example output:
 
     .. code-block:: bash
 
         {
             "error": null,
             "id": 237,
             "result": {
                 "extensions_largefiles": true,
                 "extensions_evolve": true,
                 "hooks_changegroup_push_logger": true,
                 "hooks_changegroup_repo_size": false,
                 "hooks_outgoing_pull_logger": true,
                 "phases_publish": "True",
                 "rhodecode_hg_use_rebase_for_merging": true,
                 "rhodecode_pr_merge_enabled": true,
                 "rhodecode_use_outdated_comments": true
             }
         }
     """
 
     # Restrict access to this api method to super-admins, and repo admins only.
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         settings_model = VcsSettingsModel(repo=repo)
         settings = settings_model.get_global_settings()
         settings.update(settings_model.get_repo_settings())
 
         # If only a single setting is requested fetch it from all settings.
         key = Optional.extract(key)
         if key is not None:
             settings = settings.get(key, None)
     except Exception:
-        msg = 'Failed to fetch settings for repository `{}`'.format(repoid)
+        msg = f'Failed to fetch settings for repository `{repoid}`'
         log.exception(msg)
         raise JSONRPCError(msg)
 
     return settings
 
 
 @jsonrpc_method()
 def set_repo_settings(request, apiuser, repoid, settings):
     """
     Update repository settings. Returns true on success.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository id.
     :type repoid: str or int
     :param settings: The new settings for the repository.
     :type: settings: dict
 
     Example output:
 
     .. code-block:: bash
 
         {
             "error": null,
             "id": 237,
             "result": true
         }
     """
     # Restrict access to this api method to super-admins, and repo admins only.
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     if type(settings) is not dict:
         raise JSONRPCError('Settings have to be a JSON Object.')
 
     try:
         settings_model = VcsSettingsModel(repo=repoid)
 
         # Merge global, repo and incoming settings.
         new_settings = settings_model.get_global_settings()
         new_settings.update(settings_model.get_repo_settings())
         new_settings.update(settings)
 
         # Update the settings.
         inherit_global_settings = new_settings.get(
             'inherit_global_settings', False)
         settings_model.create_or_update_repo_settings(
             new_settings, inherit_global_settings=inherit_global_settings)
         Session().commit()
     except Exception:
-        msg = 'Failed to update settings for repository `{}`'.format(repoid)
+        msg = f'Failed to update settings for repository `{repoid}`'
         log.exception(msg)
         raise JSONRPCError(msg)
 
     # Indicate success.
     return True
 
 
 @jsonrpc_method()
 def maintenance(request, apiuser, repoid):
     """
     Triggers a maintenance on the given repository.
 
     This command can only be run using an |authtoken| with admin
     rights to the specified repository. For more information,
     see :ref:`config-token-ref`.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repoid: The repository name or repository ID.
     :type repoid: str or int
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "executed maintenance command",
         "executed_actions": [
            <action_message>, <action_message2>...
         ],
         "repository": "<repository name>"
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "Unable to execute maintenance on `<reponame>`"
       }
 
     """
 
     repo = get_repo_or_error(repoid)
     if not has_superadmin_permission(apiuser):
         _perms = ('repository.admin',)
         validate_repo_permissions(apiuser, repoid, repo, _perms)
 
     try:
         maintenance = repo_maintenance.RepoMaintenance()
         executed_actions = maintenance.execute(repo)
 
         return {
             'msg': 'executed maintenance command',
             'executed_actions': executed_actions,
             'repository': repo.repo_name
         }
     except Exception:
         log.exception("Exception occurred while trying to run maintenance")
         raise JSONRPCError(
             'Unable to execute maintenance on `%s`' % repo.repo_name)

rhodecode/api/views/repo_group_api.py

@@ -1,762 +1,760 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import logging
 
 from rhodecode.api import JSONRPCValidationError
 from rhodecode.api import jsonrpc_method, JSONRPCError
 from rhodecode.api.utils import (
     has_superadmin_permission, Optional, OAttr, get_user_or_error,
     get_repo_group_or_error, get_perm_or_error, get_user_group_or_error,
     get_origin, validate_repo_group_permissions, validate_set_owner_permissions)
 from rhodecode.lib import audit_logger
 from rhodecode.lib.auth import (
     HasRepoGroupPermissionAnyApi, HasUserGroupPermissionAnyApi)
 from rhodecode.model.db import Session
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.scm import RepoGroupList
 from rhodecode.model import validation_schema
 from rhodecode.model.validation_schema.schemas import repo_group_schema
 
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_repo_group(request, apiuser, repogroupid):
     """
     Return the specified |repo| group, along with permissions,
     and repositories inside the group
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repogroupid: Specify the name of ID of the repository group.
     :type repogroupid: str or int
 
 
     Example output:
 
     .. code-block:: bash
 
         {
           "error": null,
           "id": repo-group-id,
           "result": {
             "group_description": "repo group description",
             "group_id": 14,
             "group_name": "group name",
             "permissions": [
               {
                 "name": "super-admin-username",
                 "origin": "super-admin",
                 "permission": "group.admin",
                 "type": "user"
               },
               {
                 "name": "owner-name",
                 "origin": "owner",
                 "permission": "group.admin",
                 "type": "user"
               },
               {
                 "name": "user-group-name",
                 "origin": "permission",
                 "permission": "group.write",
                 "type": "user_group"
               }
             ],
             "owner": "owner-name",
             "parent_group": null,
             "repositories": [ repo-list ]
           }
         }
     """
 
     repo_group = get_repo_group_or_error(repogroupid)
     if not has_superadmin_permission(apiuser):
         # check if we have at least read permission for this repo group !
         _perms = ('group.admin', 'group.write', 'group.read',)
         if not HasRepoGroupPermissionAnyApi(*_perms)(
                 user=apiuser, group_name=repo_group.group_name):
             raise JSONRPCError(
-                'repository group `%s` does not exist' % (repogroupid,))
+                'repository group `{}` does not exist'.format(repogroupid))
 
     permissions = []
     for _user in repo_group.permissions():
         user_data = {
             'name': _user.username,
             'permission': _user.permission,
             'origin': get_origin(_user),
             'type': "user",
         }
         permissions.append(user_data)
 
     for _user_group in repo_group.permission_user_groups():
         user_group_data = {
             'name': _user_group.users_group_name,
             'permission': _user_group.permission,
             'origin': get_origin(_user_group),
             'type': "user_group",
         }
         permissions.append(user_group_data)
 
     data = repo_group.get_api_data()
     data["permissions"] = permissions
     return data
 
 
 @jsonrpc_method()
 def get_repo_groups(request, apiuser):
     """
     Returns all repository groups.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     """
 
     result = []
     _perms = ('group.read', 'group.write', 'group.admin',)
     extras = {'user': apiuser}
     for repo_group in RepoGroupList(RepoGroupModel().get_all(),
                                     perm_set=_perms, extra_kwargs=extras):
         result.append(repo_group.get_api_data())
     return result
 
 
 @jsonrpc_method()
 def create_repo_group(
         request, apiuser, group_name,
         owner=Optional(OAttr('apiuser')),
         description=Optional(''),
         copy_permissions=Optional(False)):
     """
     Creates a repository group.
 
     * If the repository group name contains "/", repository group will be
       created inside a repository group or nested repository groups
 
       For example "foo/bar/group1" will create repository group called "group1"
       inside group "foo/bar". You have to have permissions to access and
       write to the last repository group ("bar" in this example)
 
     This command can only be run using an |authtoken| with at least
     permissions to create repository groups, or admin permissions to
     parent repository groups.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param group_name: Set the repository group name.
     :type group_name: str
     :param description: Set the |repo| group description.
     :type description: str
     :param owner: Set the |repo| group owner.
     :type owner: str
     :param copy_permissions:
     :type copy_permissions:
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
           "msg": "Created new repo group `<repo_group_name>`"
           "repo_group": <repogroup_object>
       }
       error :  null
 
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         failed to create repo group `<repogroupid>`
       }
 
     """
 
     owner = validate_set_owner_permissions(apiuser, owner)
 
     description = Optional.extract(description)
     copy_permissions = Optional.extract(copy_permissions)
 
     schema = repo_group_schema.RepoGroupSchema().bind(
         # user caller
         user=apiuser)
 
     try:
         schema_data = schema.deserialize(dict(
             repo_group_name=group_name,
             repo_group_owner=owner.username,
             repo_group_description=description,
             repo_group_copy_permissions=copy_permissions,
         ))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     validated_group_name = schema_data['repo_group_name']
 
     try:
         repo_group = RepoGroupModel().create(
             owner=owner,
             group_name=validated_group_name,
             group_description=schema_data['repo_group_description'],
             copy_permissions=schema_data['repo_group_copy_permissions'])
         Session().flush()
 
         repo_group_data = repo_group.get_api_data()
         audit_logger.store_api(
             'repo_group.create', action_data={'data': repo_group_data},
             user=apiuser)
 
         Session().commit()
 
         PermissionModel().trigger_permission_flush()
 
         return {
             'msg': 'Created new repo group `%s`' % validated_group_name,
             'repo_group': repo_group.get_api_data()
         }
     except Exception:
         log.exception("Exception occurred while trying create repo group")
         raise JSONRPCError(
-            'failed to create repo group `%s`' % (validated_group_name,))
+            'failed to create repo group `{}`'.format(validated_group_name))
 
 
 @jsonrpc_method()
 def update_repo_group(
         request, apiuser, repogroupid, group_name=Optional(''),
         description=Optional(''), owner=Optional(OAttr('apiuser')),
         enable_locking=Optional(False)):
     """
     Updates repository group with the details given.
 
     This command can only be run using an |authtoken| with admin
     permissions.
 
     * If the group_name name contains "/", repository group will be updated
       accordingly with a repository group or nested repository groups
 
       For example repogroupid=group-test group_name="foo/bar/group-test"
       will update repository group called "group-test" and place it
       inside group "foo/bar".
       You have to have permissions to access and write to the last repository
       group ("bar" in this example)
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repogroupid: Set the ID of repository group.
     :type repogroupid: str or int
     :param group_name: Set the name of the |repo| group.
     :type group_name: str
     :param description: Set a description for the group.
     :type description: str
     :param owner: Set the |repo| group owner.
     :type owner: str
     :param enable_locking: Enable |repo| locking. The default is false.
     :type enable_locking: bool
     """
 
     repo_group = get_repo_group_or_error(repogroupid)
 
     if not has_superadmin_permission(apiuser):
         validate_repo_group_permissions(
             apiuser, repogroupid, repo_group, ('group.admin',))
 
     updates = dict(
         group_name=group_name
         if not isinstance(group_name, Optional) else repo_group.group_name,
 
         group_description=description
         if not isinstance(description, Optional) else repo_group.group_description,
 
         user=owner
         if not isinstance(owner, Optional) else repo_group.user.username,
 
         enable_locking=enable_locking
         if not isinstance(enable_locking, Optional) else repo_group.enable_locking
     )
 
     schema = repo_group_schema.RepoGroupSchema().bind(
         # user caller
         user=apiuser,
         old_values=repo_group.get_api_data())
 
     try:
         schema_data = schema.deserialize(dict(
             repo_group_name=updates['group_name'],
             repo_group_owner=updates['user'],
             repo_group_description=updates['group_description'],
             repo_group_enable_locking=updates['enable_locking'],
         ))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     validated_updates = dict(
         group_name=schema_data['repo_group']['repo_group_name_without_group'],
         group_parent_id=schema_data['repo_group']['repo_group_id'],
         user=schema_data['repo_group_owner'],
         group_description=schema_data['repo_group_description'],
         enable_locking=schema_data['repo_group_enable_locking'],
     )
 
     old_data = repo_group.get_api_data()
     try:
         RepoGroupModel().update(repo_group, validated_updates)
         audit_logger.store_api(
             'repo_group.edit', action_data={'old_data': old_data},
             user=apiuser)
 
         Session().commit()
         return {
-            'msg': 'updated repository group ID:%s %s' % (
+            'msg': 'updated repository group ID:{} {}'.format(
                 repo_group.group_id, repo_group.group_name),
             'repo_group': repo_group.get_api_data()
         }
     except Exception:
         log.exception(
-            u"Exception occurred while trying update repo group %s",
+            "Exception occurred while trying update repo group %s",
             repogroupid)
         raise JSONRPCError('failed to update repository group `%s`'
                            % (repogroupid,))
 
 
 @jsonrpc_method()
 def delete_repo_group(request, apiuser, repogroupid):
     """
     Deletes a |repo| group.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repogroupid: Set the name or ID of repository group to be
         deleted.
     :type repogroupid: str or int
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'msg': 'deleted repo group ID:<repogroupid> <repogroupname>'
         'repo_group': null
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to delete repo group ID:<repogroupid> <repogroupname>"
       }
 
     """
 
     repo_group = get_repo_group_or_error(repogroupid)
     if not has_superadmin_permission(apiuser):
         validate_repo_group_permissions(
             apiuser, repogroupid, repo_group, ('group.admin',))
 
     old_data = repo_group.get_api_data()
     try:
         RepoGroupModel().delete(repo_group)
         audit_logger.store_api(
             'repo_group.delete', action_data={'old_data': old_data},
             user=apiuser)
         Session().commit()
         return {
             'msg': 'deleted repo group ID:%s %s' %
                    (repo_group.group_id, repo_group.group_name),
             'repo_group': None
         }
     except Exception:
         log.exception("Exception occurred while trying to delete repo group")
         raise JSONRPCError('failed to delete repo group ID:%s %s' %
                            (repo_group.group_id, repo_group.group_name))
 
 
 @jsonrpc_method()
 def grant_user_permission_to_repo_group(
         request, apiuser, repogroupid, userid, perm,
         apply_to_children=Optional('none')):
     """
     Grant permission for a user on the given repository group, or update
     existing permissions if found.
 
     This command can only be run using an |authtoken| with admin
     permissions.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repogroupid: Set the name or ID of repository group.
     :type repogroupid: str or int
     :param userid: Set the user name.
     :type userid: str
     :param perm: (group.(none|read|write|admin))
     :type perm: str
     :param apply_to_children: 'none', 'repos', 'groups', 'all'
     :type apply_to_children: str
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
                   "success": true
                 }
         error:  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
       }
 
     """
 
     repo_group = get_repo_group_or_error(repogroupid)
 
     if not has_superadmin_permission(apiuser):
         validate_repo_group_permissions(
             apiuser, repogroupid, repo_group, ('group.admin',))
 
     user = get_user_or_error(userid)
     perm = get_perm_or_error(perm, prefix='group.')
     apply_to_children = Optional.extract(apply_to_children)
 
     perm_additions = [[user.user_id, perm.permission_name, "user"]]
     try:
         changes = RepoGroupModel().update_permissions(
             repo_group=repo_group, perm_additions=perm_additions,
             recursive=apply_to_children, cur_user=apiuser)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo_group.edit.permissions', action_data=action_data,
             user=apiuser)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             '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.exception("Exception occurred while trying to grant "
                       "user permissions to repo group")
         raise JSONRPCError(
             'failed to edit permission for user: '
             '`%s` in repo group: `%s`' % (userid, repo_group.name))
 
 
 @jsonrpc_method()
 def revoke_user_permission_from_repo_group(
         request, apiuser, repogroupid, userid,
         apply_to_children=Optional('none')):
     """
     Revoke permission for a user in a given repository group.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo| group.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repogroupid: Set the name or ID of the repository group.
     :type repogroupid: str or int
     :param userid: Set the user name to revoke.
     :type userid: str
     :param apply_to_children: 'none', 'repos', 'groups', 'all'
     :type apply_to_children: str
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Revoked perm (recursive:<apply_to_children>) for user: `<username>` in repo group: `<repo_group_name>`",
                   "success": true
                 }
         error:  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to edit permission for user: `<userid>` in repo group: `<repo_group_name>`"
       }
 
     """
 
     repo_group = get_repo_group_or_error(repogroupid)
 
     if not has_superadmin_permission(apiuser):
         validate_repo_group_permissions(
             apiuser, repogroupid, repo_group, ('group.admin',))
 
     user = get_user_or_error(userid)
     apply_to_children = Optional.extract(apply_to_children)
 
     perm_deletions = [[user.user_id, None, "user"]]
     try:
         changes = RepoGroupModel().update_permissions(
             repo_group=repo_group, perm_deletions=perm_deletions,
             recursive=apply_to_children, cur_user=apiuser)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo_group.edit.permissions', action_data=action_data,
             user=apiuser)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             '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.exception("Exception occurred while trying revoke user "
                       "permission from repo group")
         raise JSONRPCError(
             'failed to edit permission for user: '
             '`%s` in repo group: `%s`' % (userid, repo_group.name))
 
 
 @jsonrpc_method()
 def grant_user_group_permission_to_repo_group(
         request, apiuser, repogroupid, usergroupid, perm,
         apply_to_children=Optional('none'), ):
     """
     Grant permission for a user group on given repository group, or update
     existing permissions if found.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo| group.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param repogroupid: Set the 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
     :param apply_to_children: 'none', 'repos', 'groups', 'all'
     :type apply_to_children: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg" : "Granted perm: `<perm>` (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
         "success": true
 
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
       }
 
     """
 
     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 has_superadmin_permission(apiuser):
         validate_repo_group_permissions(
             apiuser, repogroupid, repo_group, ('group.admin',))
 
         # check if we have at least read permission for this user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
 
     apply_to_children = Optional.extract(apply_to_children)
 
     perm_additions = [[user_group.users_group_id, perm.permission_name, "user_group"]]
     try:
         changes = RepoGroupModel().update_permissions(
             repo_group=repo_group, perm_additions=perm_additions,
             recursive=apply_to_children, cur_user=apiuser)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo_group.edit.permissions', action_data=action_data,
             user=apiuser)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             '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
                    ),
             'success': True
         }
     except Exception:
         log.exception("Exception occurred while trying to grant user "
                       "group permissions to repo group")
         raise JSONRPCError(
             'failed to edit permission for user group: `%s` in '
             'repo group: `%s`' % (
                 usergroupid, repo_group.name
             )
         )
 
 
 @jsonrpc_method()
 def revoke_user_group_permission_from_repo_group(
         request, apiuser, repogroupid, usergroupid,
         apply_to_children=Optional('none')):
     """
     Revoke permission for user group on given repository.
 
     This command can only be run using an |authtoken| with admin
     permissions on the |repo| group.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :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
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg" : "Revoked perm (recursive:<apply_to_children>) for user group: `<usersgroupname>` in repo group: `<repo_group_name>`",
                   "success": true
                 }
         error:  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to edit permission for user group: `<usergroup>` in repo group: `<repo_group_name>`"
       }
 
 
     """
 
     repo_group = get_repo_group_or_error(repogroupid)
     user_group = get_user_group_or_error(usergroupid)
     if not has_superadmin_permission(apiuser):
         validate_repo_group_permissions(
             apiuser, repogroupid, repo_group, ('group.admin',))
 
         # check if we have at least read permission for this user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
 
     apply_to_children = Optional.extract(apply_to_children)
 
     perm_deletions = [[user_group.users_group_id, None, "user_group"]]
     try:
         changes = RepoGroupModel().update_permissions(
             repo_group=repo_group, perm_deletions=perm_deletions,
             recursive=apply_to_children, cur_user=apiuser)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'repo_group.edit.permissions', action_data=action_data,
             user=apiuser)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             '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
         }
     except Exception:
         log.exception("Exception occurred while trying revoke user group "
                       "permissions from repo group")
         raise JSONRPCError(
             'failed to edit permission for user group: '
             '`%s` in repo group: `%s`' % (
                 user_group.users_group_name, repo_group.name
             )
         )

rhodecode/api/views/search_api.py

@@ -1,152 +1,150 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import logging
 
 from rhodecode.api import jsonrpc_method
 from rhodecode.api.exc import JSONRPCValidationError, JSONRPCForbidden
 from rhodecode.api.utils import Optional, has_superadmin_permission
 from rhodecode.lib.index import searcher_from_config
 from rhodecode.lib.user_log_filter import user_log_filter
 from rhodecode.model import validation_schema
 from rhodecode.model.db import joinedload, UserLog
 from rhodecode.model.validation_schema.schemas import search_schema
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def search(request, apiuser, search_query, search_type, page_limit=Optional(10),
            page=Optional(1), search_sort=Optional('desc:date'),
            repo_name=Optional(None), repo_group_name=Optional(None)):
     """
     Fetch Full Text Search results using API.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param search_query: Search query.
     :type search_query: str
     :param search_type: Search type. The following are valid options:
         * commit
         * content
         * path
     :type search_type: str
     :param page_limit: Page item limit, from 1 to 500. Default 10 items.
     :type page_limit: Optional(int)
     :param page: Page number. Default first page.
     :type page: Optional(int)
     :param search_sort: Search sort order.Must start with asc: or desc: Default desc:date.
         The following are valid options:
         * asc|desc:message.raw
         * asc|desc:date
         * asc|desc:author.email.raw
         * asc|desc:message.raw
         * newfirst (old legacy equal to desc:date)
         * oldfirst (old legacy equal to asc:date)
 
     :type search_sort: Optional(str)
     :param repo_name: Filter by one repo. Default is all.
     :type repo_name: Optional(str)
     :param repo_group_name: Filter by one repo group. Default is all.
     :type repo_group_name: Optional(str)
     """
 
     data = {'execution_time': ''}
     repo_name = Optional.extract(repo_name)
     repo_group_name = Optional.extract(repo_group_name)
 
     schema = search_schema.SearchParamsSchema()
 
     try:
         search_params = schema.deserialize(
             dict(search_query=search_query,
                  search_type=search_type,
                  search_sort=Optional.extract(search_sort),
                  page_limit=Optional.extract(page_limit),
                  requested_page=Optional.extract(page))
         )
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     search_query = search_params.get('search_query')
     search_type = search_params.get('search_type')
     search_sort = search_params.get('search_sort')
 
     if search_params.get('search_query'):
         page_limit = search_params['page_limit']
         requested_page = search_params['requested_page']
 
         searcher = searcher_from_config(request.registry.settings)
 
         try:
             search_result = searcher.search(
                 search_query, search_type, apiuser, repo_name, repo_group_name,
                 requested_page=requested_page, page_limit=page_limit, sort=search_sort)
 
             data.update(dict(
                 results=list(search_result['results']), page=requested_page,
                 item_count=search_result['count'],
                 items_per_page=page_limit))
         finally:
             searcher.cleanup()
 
         if not search_result['error']:
-            data['execution_time'] = '%s results (%.4f seconds)' % (
+            data['execution_time'] = '{} results ({:.4f} seconds)'.format(
                 search_result['count'],
                 search_result['runtime'])
         else:
             node = schema['search_query']
             raise JSONRPCValidationError(
                 colander_exc=validation_schema.Invalid(node, search_result['error']))
 
     return data
 
 
 @jsonrpc_method()
 def get_audit_logs(request, apiuser, query):
     """
     return full audit logs based on the query.
 
     Please see `example query in admin > settings > audit logs` for examples
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param query: filter query, example: action:repo.artifact.add date:[20200401 TO 20200601]"
     :type query: str
     """
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     filter_term = query
     ret = []
 
     # show all user actions
     user_log = UserLog.query() \
         .options(joinedload(UserLog.user)) \
         .options(joinedload(UserLog.repository)) \
         .order_by(UserLog.action_date.desc())
 
     audit_log = user_log_filter(user_log, filter_term)
 
     for entry in audit_log:
         ret.append(entry)
     return ret

rhodecode/api/views/server_api.py

@@ -1,420 +1,418 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 import itertools
 import base64
 
 from rhodecode.api import (
     jsonrpc_method, JSONRPCError, JSONRPCForbidden, find_methods)
 
 from rhodecode.api.utils import (
     Optional, OAttr, has_superadmin_permission, get_user_or_error)
 from rhodecode.lib.utils import repo2db_mapper
 from rhodecode.lib import system_info
 from rhodecode.lib import user_sessions
 from rhodecode.lib import exc_tracking
 from rhodecode.lib.ext_json import json
 from rhodecode.lib.utils2 import safe_int
 from rhodecode.model.db import UserIpMap
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.settings import VcsSettingsModel
 from rhodecode.apps.file_store import utils
 from rhodecode.apps.file_store.exceptions import FileNotAllowedException, \
     FileOverSizeException
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_server_info(request, apiuser):
     """
     Returns the |RCE| server information.
 
     This includes the running version of |RCE| and all installed
     packages. This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'modules': [<module name>,...]
         'py_version': <python version>,
         'platform': <platform type>,
         'rhodecode_version': <rhodecode version>
       }
       error :  null
     """
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     server_info = ScmModel().get_server_info(request.environ)
     # rhodecode-index requires those
 
     server_info['index_storage'] = server_info['search']['value']['location']
     server_info['storage'] = server_info['storage']['value']['path']
 
     return server_info
 
 
 @jsonrpc_method()
 def get_repo_store(request, apiuser):
     """
     Returns the |RCE| repository storage information.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'modules': [<module name>,...]
         'py_version': <python version>,
         'platform': <platform type>,
         'rhodecode_version': <rhodecode version>
       }
       error :  null
     """
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     path = VcsSettingsModel().get_repos_location()
     return {"path": path}
 
 
 @jsonrpc_method()
 def get_ip(request, apiuser, userid=Optional(OAttr('apiuser'))):
     """
     Displays the IP Address as seen from the |RCE| server.
 
     * This command displays the IP Address, as well as all the defined IP
       addresses for the specified user. If the ``userid`` is not set, the
       data returned is for the user calling the method.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from |authtoken|.
     :type apiuser: AuthUser
     :param userid: Sets the userid for which associated IP Address data
         is returned.
     :type userid: Optional(str or int)
 
     Example output:
 
     .. code-block:: bash
 
         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 not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     userid = Optional.extract(userid, evaluate_locals=locals())
     userid = getattr(userid, 'user_id', userid)
 
     user = get_user_or_error(userid)
     ips = UserIpMap.query().filter(UserIpMap.user == user).all()
     return {
         'server_ip_addr': request.rpc_ip_addr,
         'user_ips': ips
     }
 
 
 @jsonrpc_method()
 def rescan_repos(request, apiuser, remove_obsolete=Optional(False)):
     """
     Triggers a rescan of the specified repositories.
 
     * If the ``remove_obsolete`` option is set, it also deletes repositories
       that are found in the database but not on the file system, so called
       "clean zombies".
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param remove_obsolete: Deletes repositories from the database that
         are not found on the filesystem.
     :type remove_obsolete: Optional(``True`` | ``False``)
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         'added': [<added repository name>,...]
         'removed': [<removed repository name>,...]
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         'Error occurred during rescan repositories action'
       }
 
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     try:
         rm_obsolete = Optional.extract(remove_obsolete)
         added, removed = repo2db_mapper(ScmModel().repo_scan(),
                                         remove_obsolete=rm_obsolete)
         return {'added': added, 'removed': removed}
     except Exception:
         log.exception('Failed to run repo rescann')
         raise JSONRPCError(
             'Error occurred during rescan repositories action'
         )
 
 
 @jsonrpc_method()
 def cleanup_sessions(request, apiuser, older_then=Optional(60)):
     """
     Triggers a session cleanup action.
 
     If the ``older_then`` option is set, only sessions that hasn't been
     accessed in the given number of days will be removed.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param older_then: Deletes session that hasn't been accessed
         in given number of days.
     :type older_then: Optional(int)
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result: {
         "backend": "<type of backend>",
         "sessions_removed": <number_of_removed_sessions>
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         'Error occurred during session cleanup'
       }
 
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     older_then = safe_int(Optional.extract(older_then)) or 60
     older_than_seconds = 60 * 60 * 24 * older_then
 
     config = system_info.rhodecode_config().get_value()['value']['config']
     session_model = user_sessions.get_session_handler(
         config.get('beaker.session.type', 'memory'))(config)
 
     backend = session_model.SESSION_TYPE
     try:
         cleaned = session_model.clean_sessions(
             older_than_seconds=older_than_seconds)
         return {'sessions_removed': cleaned, 'backend': backend}
     except user_sessions.CleanupCommand as msg:
         return {'cleanup_command': msg.message, 'backend': backend}
     except Exception as e:
         log.exception('Failed session cleanup')
         raise JSONRPCError(
             'Error occurred during session cleanup'
         )
 
 
 @jsonrpc_method()
 def get_method(request, apiuser, pattern=Optional('*')):
     """
     Returns list of all available API methods. By default match pattern
     os "*" but any other pattern can be specified. eg *comment* will return
     all methods with comment inside them. If just single method is matched
     returned data will also include method specification
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param pattern: pattern to match method names against
     :type pattern: Optional("*")
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       "result": [
         "changeset_comment",
         "comment_pull_request",
         "comment_commit"
       ]
       error :  null
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       "result": [
         "comment_commit",
         {
           "apiuser": "<RequiredType>",
           "comment_type": "<Optional:u'note'>",
           "commit_id": "<RequiredType>",
           "message": "<RequiredType>",
           "repoid": "<RequiredType>",
           "request": "<RequiredType>",
           "resolves_comment_id": "<Optional:None>",
           "status": "<Optional:None>",
           "userid": "<Optional:<OptionalAttr:apiuser>>"
         }
       ]
       error :  null
     """
     from rhodecode.config.patches import inspect_getargspec
     inspect = inspect_getargspec()
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     pattern = Optional.extract(pattern)
 
     matches = find_methods(request.registry.jsonrpc_methods, pattern)
 
     args_desc = []
     matches_keys = list(matches.keys())
     if len(matches_keys) == 1:
         func = matches[matches_keys[0]]
 
         argspec = inspect.getargspec(func)
         arglist = argspec[0]
         defaults = list(map(repr, argspec[3] or []))
 
         default_empty = '<RequiredType>'
 
         # kw arguments required by this method
         func_kwargs = dict(itertools.zip_longest(
             reversed(arglist), reversed(defaults), fillvalue=default_empty))
         args_desc.append(func_kwargs)
 
     return matches_keys + args_desc
 
 
 @jsonrpc_method()
 def store_exception(request, apiuser, exc_data_json, prefix=Optional('rhodecode')):
     """
     Stores sent exception inside the built-in exception tracker in |RCE| server.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
 
     :param exc_data_json: JSON data with exception e.g
         {"exc_traceback": "Value `1` is not allowed", "exc_type_name": "ValueError"}
     :type exc_data_json: JSON data
 
     :param prefix: prefix for error type, e.g 'rhodecode', 'vcsserver', 'rhodecode-tools'
     :type prefix: Optional("rhodecode")
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       "result": {
         "exc_id": 139718459226384,
         "exc_url": "http://localhost:8080/_admin/settings/exceptions/139718459226384"
       }
       error :  null
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     prefix = Optional.extract(prefix)
     exc_id = exc_tracking.generate_id()
 
     try:
         exc_data = json.loads(exc_data_json)
     except Exception:
         log.error('Failed to parse JSON: %r', exc_data_json)
         raise JSONRPCError('Failed to parse JSON data from exc_data_json field. '
                            'Please make sure it contains a valid JSON.')
 
     try:
         exc_traceback = exc_data['exc_traceback']
         exc_type_name = exc_data['exc_type_name']
     except KeyError as err:
         raise JSONRPCError('Missing exc_traceback, or exc_type_name '
                            'in exc_data_json field. Missing: {}'.format(err))
 
     exc_tracking._store_exception(
         exc_id=exc_id, exc_traceback=exc_traceback,
         exc_type_name=exc_type_name, prefix=prefix)
 
     exc_url = request.route_url(
         'admin_settings_exception_tracker_show', exception_id=exc_id)
     return {'exc_id': exc_id, 'exc_url': exc_url}

rhodecode/api/views/testing_api.py

@@ -1,93 +1,91 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 
 import datetime
 import decimal
 import logging
 import time
 
 from rhodecode.api import jsonrpc_method, jsonrpc_deprecated_method, JSONRPCError, JSONRPCForbidden
 
 from rhodecode.api.utils import Optional, OAttr
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def test(request, apiuser, args):
     return args
 
 
 @jsonrpc_method()
 def test_ok(request, apiuser):
     return {
         'who': f'hello {apiuser}',
         'obj': {
             'time': time.time(),
             'dt': datetime.datetime.now(),
             'decimal': decimal.Decimal('0.123')
         }
     }
 
 
 @jsonrpc_method()
 def test_error(request, apiuser):
     raise JSONRPCError('error happened')
 
 
 @jsonrpc_method()
 def test_exception(request, apiuser):
     raise Exception('something unhandled')
 
 
 @jsonrpc_method()
 def test_params(request, apiuser, params):
     return {
         'who': f'hello {apiuser}',
         'params': params
     }
 
 
 @jsonrpc_method()
 def test_params_opt(
         request, apiuser, params, opt1=False, opt2=Optional(True),
         opt3=Optional(OAttr('apiuser'))):
     opt2 = Optional.extract(opt2)
     opt3 = Optional.extract(opt3, evaluate_locals=locals())
     return {
         'who': f'hello {apiuser}',
         'params': params,
         'opts': [
             opt1, opt2, opt3
         ]
     }
 
 
 @jsonrpc_method()
 @jsonrpc_deprecated_method(
     use_method='test_ok', deprecated_at_version='4.0.0')
 def test_deprecated_method(request, apiuser):
     return 'value'
 
 
 @jsonrpc_method()
 def test_forbidden_method(request, apiuser):
     raise JSONRPCForbidden()

rhodecode/api/views/user_api.py

@@ -1,572 +1,570 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 
 from rhodecode.api import (
     jsonrpc_method, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
 from rhodecode.api.utils import (
     Optional, OAttr, has_superadmin_permission, get_user_or_error, store_update)
 from rhodecode.lib import audit_logger
 from rhodecode.lib.auth import AuthUser, PasswordGenerator
 from rhodecode.lib.exceptions import DefaultUserException
 from rhodecode.lib.utils2 import safe_int, str2bool
 from rhodecode.model.db import Session, User, Repository
 from rhodecode.model.user import UserModel
 from rhodecode.model import validation_schema
 from rhodecode.model.validation_schema.schemas import user_schema
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_user(request, apiuser, userid=Optional(OAttr('apiuser'))):
     """
     Returns the information associated with a username or userid.
 
     * If the ``userid`` is not set, this command returns the information
       for the ``userid`` calling the method.
 
     .. note::
 
        Normal users may only run this command against their ``userid``. For
        full privileges you must run this command using an |authtoken| with
        admin rights.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param userid: Sets the userid for which data will be returned.
     :type userid: Optional(str or int)
 
     Example output:
 
     .. code-block:: bash
 
         {
           "error": null,
           "id": <id>,
           "result": {
             "active": true,
             "admin": false,
             "api_keys": [ list of keys ],
             "auth_tokens": [ list of tokens with details ],
             "email": "[e-mail]",
             "emails": [
               "[e-mail]"
             ],
             "extern_name": "rhodecode",
             "extern_type": "rhodecode",
             "firstname": "username",
             "description": "user description",
             "ip_addresses": [],
             "language": null,
             "last_login": "Timestamp",
             "last_activity": "Timestamp",
             "lastname": "surnae",
             "permissions": <deprecated>,
             "permissions_summary": {
               "global": [
                 "hg.inherit_default_perms.true",
                 "usergroup.read",
                 "hg.repogroup.create.false",
                 "hg.create.none",
                 "hg.password_reset.enabled",
                 "hg.extern_activate.manual",
                 "hg.create.write_on_repogroup.false",
                 "hg.usergroup.create.false",
                 "group.none",
                 "repository.none",
                 "hg.register.none",
                 "hg.fork.repository"
               ],
               "repositories": { "username/example": "repository.write"},
               "repositories_groups": { "user-group/repo": "group.none" },
               "user_groups": { "user_group_name": "usergroup.read" }
             }
             "user_id": 32,
             "username": "username"
           }
         }
     """
 
     if not has_superadmin_permission(apiuser):
         # make sure normal user does not pass someone else userid,
         # he is not allowed to do that
         if not isinstance(userid, Optional) and userid != apiuser.user_id:
             raise JSONRPCError('userid is not the same as your user')
 
     userid = Optional.extract(userid, evaluate_locals=locals())
     userid = getattr(userid, 'user_id', userid)
 
     user = get_user_or_error(userid)
     data = user.get_api_data(include_secrets=True)
     permissions = AuthUser(user_id=user.user_id).permissions
     data['permissions'] = permissions  # TODO(marcink): should be deprecated
     data['permissions_summary'] = permissions
     return data
 
 
 @jsonrpc_method()
 def get_users(request, apiuser):
     """
     Lists all users in the |RCE| user database.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: [<user_object>, ...]
         error:  null
     """
 
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     result = []
     users_list = User.query().order_by(User.username) \
         .filter(User.username != User.DEFAULT_USER) \
         .all()
     for user in users_list:
         result.append(user.get_api_data(include_secrets=True))
     return result
 
 
 @jsonrpc_method()
 def create_user(request, apiuser, username, email, password=Optional(''),
                 firstname=Optional(''), lastname=Optional(''), description=Optional(''),
                 active=Optional(True), admin=Optional(False),
                 extern_name=Optional('rhodecode'),
                 extern_type=Optional('rhodecode'),
                 force_password_change=Optional(False),
                 create_personal_repo_group=Optional(None)):
     """
     Creates a new user and returns the new user object.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param username: Set the new username.
     :type username: str or int
     :param email: Set the user email address.
     :type email: str
     :param password: Set the new user password.
     :type password: Optional(str)
     :param firstname: Set the new user firstname.
     :type firstname: Optional(str)
     :param lastname: Set the new user surname.
     :type lastname: Optional(str)
     :param description: Set user description, or short bio. Metatags are allowed.
     :type description: Optional(str)
     :param active: Set the user as active.
     :type active: Optional(``True`` | ``False``)
     :param admin: Give the new user admin rights.
     :type admin: Optional(``True`` | ``False``)
     :param extern_name: Set the authentication plugin name.
         Using LDAP this is filled with LDAP UID.
     :type extern_name: Optional(str)
     :param extern_type: Set the new user authentication plugin.
     :type extern_type: Optional(str)
     :param force_password_change: Force the new user to change password
         on next login.
     :type force_password_change: Optional(``True`` | ``False``)
     :param create_personal_repo_group: Create personal repo group for this user
     :type create_personal_repo_group: Optional(``True`` | ``False``)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
             "msg" : "created new user `<username>`",
             "user": <user_obj>
         }
         error:  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "user `<username>` already exist"
         or
         "email `<email>` already exist"
         or
         "failed to create user `<username>`"
       }
 
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     if UserModel().get_by_username(username):
-        raise JSONRPCError("user `%s` already exist" % (username,))
+        raise JSONRPCError("user `{}` already exist".format(username))
 
     if UserModel().get_by_email(email, case_insensitive=True):
-        raise JSONRPCError("email `%s` already exist" % (email,))
+        raise JSONRPCError("email `{}` already exist".format(email))
 
     # generate random password if we actually given the
     # extern_name and it's not rhodecode
     if (not isinstance(extern_name, Optional) and
                 Optional.extract(extern_name) != 'rhodecode'):
         # generate temporary password if user is external
         password = PasswordGenerator().gen_password(length=16)
     create_repo_group = Optional.extract(create_personal_repo_group)
     if isinstance(create_repo_group, str):
         create_repo_group = str2bool(create_repo_group)
 
     username = Optional.extract(username)
     password = Optional.extract(password)
     email = Optional.extract(email)
     first_name = Optional.extract(firstname)
     last_name = Optional.extract(lastname)
     description = Optional.extract(description)
     active = Optional.extract(active)
     admin = Optional.extract(admin)
     extern_type = Optional.extract(extern_type)
     extern_name = Optional.extract(extern_name)
 
     schema = user_schema.UserSchema().bind(
         # user caller
         user=apiuser)
     try:
         schema_data = schema.deserialize(dict(
             username=username,
             email=email,
             password=password,
             first_name=first_name,
             last_name=last_name,
             active=active,
             admin=admin,
             description=description,
             extern_type=extern_type,
             extern_name=extern_name,
             ))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     try:
         user = UserModel().create_or_update(
             username=schema_data['username'],
             password=schema_data['password'],
             email=schema_data['email'],
             firstname=schema_data['first_name'],
             lastname=schema_data['last_name'],
             description=schema_data['description'],
             active=schema_data['active'],
             admin=schema_data['admin'],
             extern_type=schema_data['extern_type'],
             extern_name=schema_data['extern_name'],
             force_password_change=Optional.extract(force_password_change),
             create_repo_group=create_repo_group
         )
         Session().flush()
         creation_data = user.get_api_data()
         audit_logger.store_api(
             'user.create', action_data={'data': creation_data},
             user=apiuser)
 
         Session().commit()
         return {
             'msg': 'created new user `%s`' % username,
             'user': user.get_api_data(include_secrets=True)
         }
     except Exception:
         log.exception('Error occurred during creation of user')
-        raise JSONRPCError('failed to create user `%s`' % (username,))
+        raise JSONRPCError('failed to create user `{}`'.format(username))
 
 
 @jsonrpc_method()
 def update_user(request, apiuser, userid, username=Optional(None),
                 email=Optional(None), password=Optional(None),
                 firstname=Optional(None), lastname=Optional(None),
                 description=Optional(None), active=Optional(None), admin=Optional(None),
                 extern_type=Optional(None), extern_name=Optional(None), ):
     """
     Updates the details for the specified user, if that user exists.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from |authtoken|.
     :type apiuser: AuthUser
     :param userid: Set the ``userid`` to update.
     :type userid: str or int
     :param username: Set the new username.
     :type username: str or int
     :param email: Set the new email.
     :type email: str
     :param password: Set the new password.
     :type password: Optional(str)
     :param firstname: Set the new first name.
     :type firstname: Optional(str)
     :param lastname: Set the new surname.
     :type lastname: Optional(str)
     :param description: Set user description, or short bio. Metatags are allowed.
     :type description: Optional(str)
     :param active: Set the new user as active.
     :type active: Optional(``True`` | ``False``)
     :param admin: Give the user admin rights.
     :type admin: Optional(``True`` | ``False``)
     :param extern_name: Set the authentication plugin user name.
         Using LDAP this is filled with LDAP UID.
     :type extern_name: Optional(str)
     :param extern_type: Set the authentication plugin type.
     :type extern_type: Optional(str)
 
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
             "msg" : "updated user ID:<userid> <username>",
             "user": <user_object>,
         }
         error:  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to update user `<username>`"
       }
 
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     user = get_user_or_error(userid)
     old_data = user.get_api_data()
     # 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, description, 'description')
         store_update(updates, active, 'active')
         store_update(updates, admin, 'admin')
         store_update(updates, extern_name, 'extern_name')
         store_update(updates, extern_type, 'extern_type')
 
         user = UserModel().update_user(user, **updates)
         audit_logger.store_api(
             'user.edit', action_data={'old_data': old_data},
             user=apiuser)
         Session().commit()
         return {
-            'msg': 'updated user ID:%s %s' % (user.user_id, user.username),
+            'msg': 'updated user ID:{} {}'.format(user.user_id, user.username),
             'user': user.get_api_data(include_secrets=True)
         }
     except DefaultUserException:
         log.exception("Default user edit exception")
         raise JSONRPCError('editing default user is forbidden')
     except Exception:
         log.exception("Error occurred during update of user")
-        raise JSONRPCError('failed to update user `%s`' % (userid,))
+        raise JSONRPCError('failed to update user `{}`'.format(userid))
 
 
 @jsonrpc_method()
 def delete_user(request, apiuser, userid):
     """
     Deletes the specified user from the |RCE| user database.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     .. important::
 
        Ensure all open pull requests and open code review
        requests to this user are close.
 
        Also ensure all repositories, or repository groups owned by this
        user are reassigned before deletion.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param userid: Set the user to delete.
     :type userid: str or int
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
             "msg" : "deleted user ID:<userid> <username>",
             "user": null
         }
         error:  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to delete user ID:<userid> <username>"
       }
 
     """
     if not has_superadmin_permission(apiuser):
         raise JSONRPCForbidden()
 
     user = get_user_or_error(userid)
     old_data = user.get_api_data()
     try:
         UserModel().delete(userid)
         audit_logger.store_api(
             'user.delete', action_data={'old_data': old_data},
             user=apiuser)
 
         Session().commit()
         return {
-            'msg': 'deleted user ID:%s %s' % (user.user_id, user.username),
+            'msg': 'deleted user ID:{} {}'.format(user.user_id, user.username),
             'user': None
         }
     except Exception:
         log.exception("Error occurred during deleting of user")
         raise JSONRPCError(
-            'failed to delete user ID:%s %s' % (user.user_id, user.username))
+            'failed to delete user ID:{} {}'.format(user.user_id, user.username))
 
 
 @jsonrpc_method()
 def get_user_locks(request, apiuser, userid=Optional(OAttr('apiuser'))):
     """
     Displays all repositories locked by the specified user.
 
     * If this command is run by a non-admin user, it returns
       a list of |repos| locked by that user.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param userid: Sets the userid whose list of locked |repos| will be
         displayed.
     :type userid: Optional(str or int)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : {
             [repo_object, repo_object,...]
         }
         error :  null
     """
 
     include_secrets = False
     if not has_superadmin_permission(apiuser):
         # make sure normal user does not pass someone else userid,
         # he is not allowed to do that
         if not isinstance(userid, Optional) and userid != apiuser.user_id:
             raise JSONRPCError('userid is not the same as your user')
     else:
         include_secrets = True
 
     userid = Optional.extract(userid, evaluate_locals=locals())
     userid = getattr(userid, 'user_id', userid)
     user = get_user_or_error(userid)
 
     ret = []
 
     # show all locks
     for r in Repository.getAll():
         _user_id, _time, _reason = r.locked
         if _user_id and _time:
             _api_data = r.get_api_data(include_secrets=include_secrets)
             # if we use user filter just show the locks for this user
             if safe_int(_user_id) == user.user_id:
                 ret.append(_api_data)
 
     return ret
 
 
 @jsonrpc_method()
 def get_user_audit_logs(request, apiuser, userid=Optional(OAttr('apiuser'))):
     """
     Fetches all action logs made by the specified user.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param userid: Sets the userid whose list of locked |repos| will be
         displayed.
     :type userid: Optional(str or int)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : {
             [action, action,...]
         }
         error :  null
     """
 
     if not has_superadmin_permission(apiuser):
         # make sure normal user does not pass someone else userid,
         # he is not allowed to do that
         if not isinstance(userid, Optional) and userid != apiuser.user_id:
             raise JSONRPCError('userid is not the same as your user')
 
     userid = Optional.extract(userid, evaluate_locals=locals())
     userid = getattr(userid, 'user_id', userid)
     user = get_user_or_error(userid)
 
     ret = []
 
     # show all user actions
     for entry in UserModel().get_user_log(user, filter_term=None):
         ret.append(entry)
     return ret

rhodecode/api/views/user_group_api.py

@@ -1,907 +1,905 b''
-
-
 # Copyright (C) 2011-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import logging
 
 from rhodecode.api import (
     jsonrpc_method, JSONRPCError, JSONRPCForbidden, JSONRPCValidationError)
 from rhodecode.api.utils import (
     Optional, OAttr, store_update, has_superadmin_permission, get_origin,
     get_user_or_error, get_user_group_or_error, get_perm_or_error)
 from rhodecode.lib import audit_logger
 from rhodecode.lib.auth import HasUserGroupPermissionAnyApi, HasPermissionAnyApi
 from rhodecode.lib.exceptions import UserGroupAssignedException
 from rhodecode.model.db import Session
 from rhodecode.model.permission import PermissionModel
 from rhodecode.model.scm import UserGroupList
 from rhodecode.model.user_group import UserGroupModel
 from rhodecode.model import validation_schema
 from rhodecode.model.validation_schema.schemas import user_group_schema
 
 log = logging.getLogger(__name__)
 
 
 @jsonrpc_method()
 def get_user_group(request, apiuser, usergroupid):
     """
     Returns the data of an existing user group.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param usergroupid: Set the user group from which to return data.
     :type usergroupid: str or int
 
     Example error output:
 
     .. code-block:: bash
 
         {
           "error": null,
           "id": <id>,
           "result": {
             "active": true,
             "group_description": "group description",
             "group_name": "group name",
             "permissions": [
               {
                 "name": "owner-name",
                 "origin": "owner",
                 "permission": "usergroup.admin",
                 "type": "user"
               },
               {
               {
                 "name": "user name",
                 "origin": "permission",
                 "permission": "usergroup.admin",
                 "type": "user"
               },
               {
                 "name": "user group name",
                 "origin": "permission",
                 "permission": "usergroup.write",
                 "type": "user_group"
               }
             ],
             "permissions_summary": {
               "repositories": {
                 "aa-root-level-repo-1": "repository.admin"
               },
               "repositories_groups": {}
             },
             "owner": "owner name",
             "users": [],
             "users_group_id": 2
           }
         }
 
     """
 
     user_group = get_user_group_or_error(usergroupid)
     if not has_superadmin_permission(apiuser):
         # check if we have at least read permission for this user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
-            raise JSONRPCError('user group `%s` does not exist' % (
-                usergroupid,))
+            raise JSONRPCError('user group `{}` does not exist'.format(
+                usergroupid))
 
     permissions = []
     for _user in user_group.permissions():
         user_data = {
             'name': _user.username,
             'permission': _user.permission,
             'origin': get_origin(_user),
             'type': "user",
         }
         permissions.append(user_data)
 
     for _user_group in user_group.permission_user_groups():
         user_group_data = {
             'name': _user_group.users_group_name,
             'permission': _user_group.permission,
             'origin': get_origin(_user_group),
             'type': "user_group",
         }
         permissions.append(user_group_data)
 
     data = user_group.get_api_data()
     data["permissions"] = permissions
     data["permissions_summary"] = UserGroupModel().get_perms_summary(
         user_group.users_group_id)
     return data
 
 
 @jsonrpc_method()
 def get_user_groups(request, apiuser):
     """
     Lists all the existing user groups within RhodeCode.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
 
     Example error output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result : [<user_group_obj>,...]
         error : null
     """
 
     include_secrets = has_superadmin_permission(apiuser)
 
     result = []
     _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
     extras = {'user': apiuser}
     for user_group in UserGroupList(UserGroupModel().get_all(),
                                     perm_set=_perms, extra_kwargs=extras):
         result.append(
             user_group.get_api_data(include_secrets=include_secrets))
     return result
 
 
 @jsonrpc_method()
 def create_user_group(
         request, apiuser, group_name, description=Optional(''),
         owner=Optional(OAttr('apiuser')), active=Optional(True),
         sync=Optional(None)):
     """
     Creates a new user group.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param group_name: Set the name of the new user group.
     :type group_name: str
     :param description: Give a description of the new user group.
     :type description: str
     :param owner: Set the owner of the new user group.
         If not set, the owner is the |authtoken| user.
     :type owner: Optional(str or int)
     :param active: Set this group as active.
     :type active: Optional(``True`` | ``False``)
     :param sync: Set enabled or disabled the automatically sync from
         external authentication types like ldap. If User Group will be named like
         one from e.g ldap and sync flag is enabled members will be synced automatically.
         Sync type when enabled via API is set to `manual_api`
     :type sync: Optional(``True`` | ``False``)
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "msg": "created new user group `<groupname>`",
                   "user_group": <user_group_object>
                 }
         error:  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "user group `<group name>` already exist"
         or
         "failed to create group `<group name>`"
       }
 
     """
 
     if not has_superadmin_permission(apiuser):
         if not HasPermissionAnyApi('hg.usergroup.create.true')(user=apiuser):
             raise JSONRPCForbidden()
 
     if UserGroupModel().get_by_name(group_name):
-        raise JSONRPCError("user group `%s` already exist" % (group_name,))
+        raise JSONRPCError("user group `{}` already exist".format(group_name))
 
     if isinstance(owner, Optional):
         owner = apiuser.user_id
 
     owner = get_user_or_error(owner)
     active = Optional.extract(active)
     description = Optional.extract(description)
     sync = Optional.extract(sync)
 
     # set the sync option based on group_data
     group_data = None
     if sync:
         group_data = {
             'extern_type': 'manual_api',
             'extern_type_set_by': apiuser.username
         }
 
     schema = user_group_schema.UserGroupSchema().bind(
         # user caller
         user=apiuser)
     try:
         schema_data = schema.deserialize(dict(
             user_group_name=group_name,
             user_group_description=description,
             user_group_owner=owner.username,
             user_group_active=active,
             ))
     except validation_schema.Invalid as err:
         raise JSONRPCValidationError(colander_exc=err)
 
     try:
         user_group = UserGroupModel().create(
             name=schema_data['user_group_name'],
             description=schema_data['user_group_description'],
             owner=owner,
             active=schema_data['user_group_active'], group_data=group_data)
         Session().flush()
         creation_data = user_group.get_api_data()
         audit_logger.store_api(
             'user_group.create', action_data={'data': creation_data},
             user=apiuser)
         Session().commit()
 
         affected_user_ids = [apiuser.user_id, owner.user_id]
         PermissionModel().trigger_permission_flush(affected_user_ids)
 
         return {
             'msg': 'created new user group `%s`' % group_name,
             'user_group': creation_data
         }
     except Exception:
         log.exception("Error occurred during creation of user group")
-        raise JSONRPCError('failed to create group `%s`' % (group_name,))
+        raise JSONRPCError('failed to create group `{}`'.format(group_name))
 
 
 @jsonrpc_method()
 def update_user_group(request, apiuser, usergroupid, group_name=Optional(''),
                       description=Optional(''), owner=Optional(None),
                       active=Optional(True), sync=Optional(None)):
     """
     Updates the specified `user group` with the details provided.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param usergroupid: Set the id of the `user group` to update.
     :type usergroupid: str or int
     :param group_name: Set the new name the `user group`
     :type group_name: str
     :param description: Give a description for the `user group`
     :type description: str
     :param owner: Set the owner of the `user group`.
     :type owner: Optional(str or int)
     :param active: Set the group as active.
     :type active: Optional(``True`` | ``False``)
     :param sync: Set enabled or disabled the automatically sync from
         external authentication types like ldap. If User Group will be named like
         one from e.g ldap and sync flag is enabled members will be synced automatically.
         Sync type when enabled via API is set to `manual_api`
     :type sync: Optional(``True`` | ``False``)
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": 'updated user group ID:<user group id> <user group name>',
         "user_group": <user_group_object>
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : null
       error :  {
         "failed to update user group `<user group name>`"
       }
 
     """
 
     user_group = get_user_group_or_error(usergroupid)
     include_secrets = False
     if not has_superadmin_permission(apiuser):
         # check if we have admin permission for this user group !
         _perms = ('usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
     else:
         include_secrets = True
 
     if not isinstance(owner, Optional):
         owner = get_user_or_error(owner)
 
     old_data = user_group.get_api_data()
     updates = {}
     store_update(updates, group_name, 'users_group_name')
     store_update(updates, description, 'user_group_description')
     store_update(updates, owner, 'user')
     store_update(updates, active, 'users_group_active')
 
     sync = Optional.extract(sync)
     group_data = None
     if sync is True:
         group_data = {
             'extern_type': 'manual_api',
             'extern_type_set_by': apiuser.username
         }
     if sync is False:
         group_data = user_group.group_data
         if group_data and "extern_type" in group_data:
             del group_data["extern_type"]
 
     try:
         UserGroupModel().update(user_group, updates, group_data=group_data)
         audit_logger.store_api(
             'user_group.edit', action_data={'old_data': old_data},
             user=apiuser)
         Session().commit()
         return {
-            'msg': 'updated user group ID:%s %s' % (
+            'msg': 'updated user group ID:{} {}'.format(
                 user_group.users_group_id, user_group.users_group_name),
             'user_group': user_group.get_api_data(
                 include_secrets=include_secrets)
         }
     except Exception:
         log.exception("Error occurred during update of user group")
         raise JSONRPCError(
-            'failed to update user group `%s`' % (usergroupid,))
+            'failed to update user group `{}`'.format(usergroupid))
 
 
 @jsonrpc_method()
 def delete_user_group(request, apiuser, usergroupid):
     """
     Deletes the specified `user group`.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified repository.
 
     This command takes the following options:
 
     :param apiuser: filled automatically from apikey
     :type apiuser: AuthUser
     :param usergroupid:
     :type usergroupid: int
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "deleted user group ID:<user_group_id> <user_group_name>"
       }
       error :  null
 
     Example error output:
 
     .. code-block:: bash
 
       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 has_superadmin_permission(apiuser):
         # check if we have admin permission for this user group !
         _perms = ('usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
 
     old_data = user_group.get_api_data()
     try:
         UserGroupModel().delete(user_group)
         audit_logger.store_api(
             'user_group.delete', action_data={'old_data': old_data},
             user=apiuser)
         Session().commit()
         return {
-            'msg': 'deleted user group ID:%s %s' % (
+            'msg': 'deleted user group ID:{} {}'.format(
                 user_group.users_group_id, user_group.users_group_name),
             'user_group': None
         }
     except UserGroupAssignedException as e:
         log.exception("UserGroupAssigned error")
         raise JSONRPCError(str(e))
     except Exception:
         log.exception("Error occurred during deletion of user group")
         raise JSONRPCError(
             'failed to delete user group ID:%s %s' %(
                 user_group.users_group_id, user_group.users_group_name))
 
 
 @jsonrpc_method()
 def add_user_to_user_group(request, apiuser, usergroupid, userid):
     """
     Adds a user to a `user group`. If the user already exists in the group
     this command will return false.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified user group.
 
     This command takes the following options:
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param usergroupid: Set the name of the `user group` to which a
         user will be added.
     :type usergroupid: int
     :param userid: Set the `user_id` of the user to add to the group.
     :type userid: int
 
     Example output:
 
     .. code-block:: bash
 
       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
 
     Example error output:
 
     .. code-block:: bash
 
       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 has_superadmin_permission(apiuser):
         # check if we have admin permission for this user group !
         _perms = ('usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
-            raise JSONRPCError('user group `%s` does not exist' % (
-                usergroupid,))
+            raise JSONRPCError('user group `{}` does not exist'.format(
+                usergroupid))
 
     old_values = user_group.get_api_data()
     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`' % (
+        msg = 'added member `{}` to user group `{}`'.format(
             user.username, user_group.users_group_name
         )
         msg = msg if success else 'User is already in that group'
         if success:
             user_data = user.get_api_data()
             audit_logger.store_api(
                 'user_group.edit.member.add',
                 action_data={'user': user_data, 'old_data': old_values},
                 user=apiuser)
 
         Session().commit()
 
         return {
             'success': success,
             'msg': msg
         }
     except Exception:
         log.exception("Error occurred during adding a member to user group")
         raise JSONRPCError(
-            'failed to add member to user group `%s`' % (
+            'failed to add member to user group `{}`'.format(
                 user_group.users_group_name,
             )
         )
 
 
 @jsonrpc_method()
 def remove_user_from_user_group(request, apiuser, usergroupid, userid):
     """
     Removes a user from a user group.
 
     * If the specified user is not in the group, this command will return
       `false`.
 
     This command can only be run using an |authtoken| with admin rights to
     the specified user group.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param usergroupid: Sets the user group name.
     :type usergroupid: str or int
     :param userid: The user you wish to remove from |RCE|.
     :type userid: str or int
 
     Example output:
 
     .. code-block:: bash
 
         id : <id_given_in_input>
         result: {
                   "success":  True|False,  # depends on if member is in group
                   "msg": "removed member <username> from user group <groupname> |
                           User wasn't in group"
                 }
         error:  null
 
     """
 
     user = get_user_or_error(userid)
     user_group = get_user_group_or_error(usergroupid)
     if not has_superadmin_permission(apiuser):
         # check if we have admin permission for this user group !
         _perms = ('usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
 
     old_values = user_group.get_api_data()
     try:
         success = UserGroupModel().remove_user_from_group(user_group, user)
-        msg = 'removed member `%s` from user group `%s`' % (
+        msg = 'removed member `{}` from user group `{}`'.format(
             user.username, user_group.users_group_name
         )
         msg = msg if success else "User wasn't in group"
         if success:
             user_data = user.get_api_data()
             audit_logger.store_api(
                 'user_group.edit.member.delete',
                 action_data={'user': user_data, 'old_data': old_values},
                 user=apiuser)
 
         Session().commit()
         return {'success': success, 'msg': msg}
     except Exception:
         log.exception("Error occurred during removing an member from user group")
         raise JSONRPCError(
-            'failed to remove member from user group `%s`' % (
+            'failed to remove member from user group `{}`'.format(
                 user_group.users_group_name,
             )
         )
 
 
 @jsonrpc_method()
 def grant_user_permission_to_user_group(
         request, apiuser, usergroupid, userid, perm):
     """
     Set permissions for a user in a user group.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param usergroupid: Set the user group to edit permissions on.
     :type usergroupid: str or int
     :param userid: Set the user from whom you wish to set permissions.
     :type userid: str
     :param perm: (usergroup.(none|read|write|admin))
     :type perm: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "Granted perm: `<perm_name>` for user: `<username>` in user group: `<user_group_name>`",
         "success": true
       }
       error :  null
     """
 
     user_group = get_user_group_or_error(usergroupid)
 
     if not has_superadmin_permission(apiuser):
         # check if we have admin permission for this user group !
         _perms = ('usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
 
     user = get_user_or_error(userid)
     perm = get_perm_or_error(perm, prefix='usergroup.')
 
     try:
         changes = UserGroupModel().grant_user_permission(
             user_group=user_group, user=user, perm=perm)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'user_group.edit.permissions', action_data=action_data,
             user=apiuser)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             'msg':
-                'Granted perm: `%s` for user: `%s` in user group: `%s`' % (
+                'Granted perm: `{}` for user: `{}` in user group: `{}`'.format(
                     perm.permission_name, user.username,
                     user_group.users_group_name
                 ),
             'success': True
         }
     except Exception:
         log.exception("Error occurred during editing permissions "
                       "for user in user group")
         raise JSONRPCError(
             'failed to edit permission for user: '
             '`%s` in user group: `%s`' % (
                 userid, user_group.users_group_name))
 
 
 @jsonrpc_method()
 def revoke_user_permission_from_user_group(
         request, apiuser, usergroupid, userid):
     """
     Revoke a users permissions in a user group.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param usergroupid: Set the user group from which to revoke the user
         permissions.
     :type: usergroupid: str or int
     :param userid: Set the userid of the user whose permissions will be
         revoked.
     :type userid: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "Revoked perm for user: `<username>` in user group: `<user_group_name>`",
         "success": true
       }
       error :  null
     """
 
     user_group = get_user_group_or_error(usergroupid)
 
     if not has_superadmin_permission(apiuser):
         # check if we have admin permission for this user group !
         _perms = ('usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (usergroupid,))
+                'user group `{}` does not exist'.format(usergroupid))
 
     user = get_user_or_error(userid)
 
     try:
         changes = UserGroupModel().revoke_user_permission(
             user_group=user_group, user=user)
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'user_group.edit.permissions', action_data=action_data,
             user=apiuser)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
-            'msg': 'Revoked perm for user: `%s` in user group: `%s`' % (
+            'msg': 'Revoked perm for user: `{}` in user group: `{}`'.format(
                 user.username, user_group.users_group_name
             ),
             'success': True
         }
     except Exception:
         log.exception("Error occurred during editing permissions "
                       "for user in user group")
         raise JSONRPCError(
             'failed to edit permission for user: `%s` in user group: `%s`'
             % (userid, user_group.users_group_name))
 
 
 @jsonrpc_method()
 def grant_user_group_permission_to_user_group(
         request, apiuser, usergroupid, sourceusergroupid, perm):
     """
     Give one user group permissions to another user group.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param usergroupid: Set the user group on which to edit permissions.
     :type usergroupid: str or int
     :param sourceusergroupid: Set the source user group to which
         access/permissions will be granted.
     :type sourceusergroupid: str or int
     :param perm: (usergroup.(none|read|write|admin))
     :type perm: str
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "Granted perm: `<perm_name>` for user group: `<source_user_group_name>` in user group: `<user_group_name>`",
         "success": true
       }
       error :  null
     """
 
     user_group = get_user_group_or_error(sourceusergroupid)
     target_user_group = get_user_group_or_error(usergroupid)
     perm = get_perm_or_error(perm, prefix='usergroup.')
 
     if not has_superadmin_permission(apiuser):
         # check if we have admin permission for this user group !
         _perms = ('usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser,
                 user_group_name=target_user_group.users_group_name):
             raise JSONRPCError(
-                'to user group `%s` does not exist' % (usergroupid,))
+                'to user group `{}` does not exist'.format(usergroupid))
 
         # check if we have at least read permission for source user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (sourceusergroupid,))
+                'user group `{}` does not exist'.format(sourceusergroupid))
 
     try:
         changes = UserGroupModel().grant_user_group_permission(
             target_user_group=target_user_group,
             user_group=user_group, perm=perm)
 
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'user_group.edit.permissions', action_data=action_data,
             user=apiuser)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             'msg': 'Granted perm: `%s` for user group: `%s` '
                    'in user group: `%s`' % (
                        perm.permission_name, user_group.users_group_name,
                        target_user_group.users_group_name
                    ),
             'success': True
         }
     except Exception:
         log.exception("Error occurred during editing permissions "
                       "for user group in user group")
         raise JSONRPCError(
             'failed to edit permission for user group: `%s` in '
             'user group: `%s`' % (
                 sourceusergroupid, target_user_group.users_group_name
             )
         )
 
 
 @jsonrpc_method()
 def revoke_user_group_permission_from_user_group(
         request, apiuser, usergroupid, sourceusergroupid):
     """
     Revoke the permissions that one user group has to another.
 
     :param apiuser: This is filled automatically from the |authtoken|.
     :type apiuser: AuthUser
     :param usergroupid: Set the user group on which to edit permissions.
     :type usergroupid: str or int
     :param sourceusergroupid: Set the user group from which permissions
         are revoked.
     :type sourceusergroupid: str or int
 
     Example output:
 
     .. code-block:: bash
 
       id : <id_given_in_input>
       result : {
         "msg": "Revoked perm for user group: `<user_group_name>` in user group: `<target_user_group_name>`",
         "success": true
       }
       error :  null
     """
 
     user_group = get_user_group_or_error(sourceusergroupid)
     target_user_group = get_user_group_or_error(usergroupid)
 
     if not has_superadmin_permission(apiuser):
         # check if we have admin permission for this user group !
         _perms = ('usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser,
                 user_group_name=target_user_group.users_group_name):
             raise JSONRPCError(
-                'to user group `%s` does not exist' % (usergroupid,))
+                'to user group `{}` does not exist'.format(usergroupid))
 
         # check if we have at least read permission
         # for the source user group !
         _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)
         if not HasUserGroupPermissionAnyApi(*_perms)(
                 user=apiuser, user_group_name=user_group.users_group_name):
             raise JSONRPCError(
-                'user group `%s` does not exist' % (sourceusergroupid,))
+                'user group `{}` does not exist'.format(sourceusergroupid))
 
     try:
         changes = UserGroupModel().revoke_user_group_permission(
             target_user_group=target_user_group, user_group=user_group)
         action_data = {
             'added': changes['added'],
             'updated': changes['updated'],
             'deleted': changes['deleted'],
         }
         audit_logger.store_api(
             'user_group.edit.permissions', action_data=action_data,
             user=apiuser)
         Session().commit()
         PermissionModel().flush_user_permission_caches(changes)
 
         return {
             'msg': 'Revoked perm for user group: '
                    '`%s` in user group: `%s`' % (
                        user_group.users_group_name,
                        target_user_group.users_group_name
                    ),
             'success': True
         }
     except Exception:
         log.exception("Error occurred during editing permissions "
                       "for user group in user group")
         raise JSONRPCError(
             'failed to edit permission for user group: '
             '`%s` in user group: `%s`' % (
                 sourceusergroupid, target_user_group.users_group_name
             )
         )

rhodecode/lib/vcs/geventcurl.py

@@ -1,251 +1,251 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 This serves as a drop in replacement for pycurl. It implements the pycurl Curl
 class in a way that is compatible with gevent.
 """
 
 
 import logging
 import gevent
 import pycurl
 import greenlet
 
 # Import everything from pycurl.
 # This allows us to use this module as a drop in replacement of pycurl.
 from pycurl import *  # pragma: no cover
 
 from gevent import core
 from gevent.hub import Waiter
 
 
 log = logging.getLogger(__name__)
 
 
 class GeventCurlMulti(object):
     """
     Wrapper around pycurl.CurlMulti that integrates it into gevent's event
     loop.
 
     Parts of this class are a modified version of code copied from the Tornado
     Web Server project which is licensed under the Apache License, Version 2.0
     (the "License"). To be more specific the code originates from this file:
     https://github.com/tornadoweb/tornado/blob/stable/tornado/curl_httpclient.py
 
     This is the original license header of the origin:
 
         Copyright 2009 Facebook
 
         Licensed under the Apache License, Version 2.0 (the "License"); you may
         not use this file except in compliance with the License. You may obtain
         a copy of the License at
 
             http://www.apache.org/licenses/LICENSE-2.0
 
         Unless required by applicable law or agreed to in writing, software
         distributed under the License is distributed on an "AS IS" BASIS,
         WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
         implied. See the License for the specific language governing
         permissions and limitations under the License.
     """
 
     def __init__(self, loop=None):
         self._watchers = {}
         self._timeout = None
         self.loop = loop or gevent.get_hub().loop
 
         # Setup curl's multi instance.
         self._curl_multi = pycurl.CurlMulti()
         self.setopt(pycurl.M_TIMERFUNCTION, self._set_timeout)
         self.setopt(pycurl.M_SOCKETFUNCTION, self._handle_socket)
 
     def __getattr__(self, item):
         """
         The pycurl.CurlMulti class is final and we cannot subclass it.
         Therefore we are wrapping it and forward everything to it here.
         """
         return getattr(self._curl_multi, item)
 
     def add_handle(self, curl):
         """
         Add handle variant that also takes care about the initial invocation of
         socket action method. This is done by setting an immediate timeout.
         """
         result = self._curl_multi.add_handle(curl)
         self._set_timeout(0)
         return result
 
     def _handle_socket(self, event, fd, multi, data):
         """
         Called by libcurl when it wants to change the file descriptors it cares
         about.
         """
         event_map = {
             pycurl.POLL_NONE: core.NONE,
             pycurl.POLL_IN: core.READ,
             pycurl.POLL_OUT: core.WRITE,
             pycurl.POLL_INOUT: core.READ | core.WRITE
         }
 
         if event == pycurl.POLL_REMOVE:
             watcher = self._watchers.pop(fd, None)
             if watcher is not None:
                 watcher.stop()
         else:
             gloop_event = event_map[event]
             watcher = self._watchers.get(fd)
             if watcher is None:
                 watcher = self.loop.io(fd, gloop_event)
                 watcher.start(self._handle_events, fd, pass_events=True)
                 self._watchers[fd] = watcher
             else:
                 if watcher.events != gloop_event:
                     watcher.stop()
                     watcher.events = gloop_event
                     watcher.start(self._handle_events, fd, pass_events=True)
 
     def _set_timeout(self, msecs):
         """
         Called by libcurl to schedule a timeout.
         """
         if self._timeout is not None:
             self._timeout.stop()
         self._timeout = self.loop.timer(msecs/1000.0)
         self._timeout.start(self._handle_timeout)
 
     def _handle_events(self, events, fd):
         action = 0
         if events & core.READ:
             action |= pycurl.CSELECT_IN
         if events & core.WRITE:
             action |= pycurl.CSELECT_OUT
         while True:
             try:
                 ret, num_handles = self._curl_multi.socket_action(fd, action)
             except pycurl.error as e:
                 ret = e.args[0]
             if ret != pycurl.E_CALL_MULTI_PERFORM:
                 break
         self._finish_pending_requests()
 
     def _handle_timeout(self):
         """
         Called by IOLoop when the requested timeout has passed.
         """
         if self._timeout is not None:
             self._timeout.stop()
             self._timeout = None
         while True:
             try:
                 ret, num_handles = self._curl_multi.socket_action(
                     pycurl.SOCKET_TIMEOUT, 0)
             except pycurl.error as e:
                 ret = e.args[0]
             if ret != pycurl.E_CALL_MULTI_PERFORM:
                 break
         self._finish_pending_requests()
 
         # In theory, we shouldn't have to do this because curl will call
         # _set_timeout whenever the timeout changes. However, sometimes after
         # _handle_timeout we will need to reschedule immediately even though
         # nothing has changed from curl's perspective. This is because when
         # socket_action is called with SOCKET_TIMEOUT, libcurl decides
         # internally which timeouts need to be processed by using a monotonic
         # clock (where available) while tornado uses python's time.time() to
         # decide when timeouts have occurred. When those clocks disagree on
         # elapsed time (as they will whenever there is an NTP adjustment),
         # tornado might call _handle_timeout before libcurl is ready. After
         # each timeout, resync the scheduled timeout with libcurl's current
         # state.
         new_timeout = self._curl_multi.timeout()
         if new_timeout >= 0:
             self._set_timeout(new_timeout)
 
     def _finish_pending_requests(self):
         """
         Process any requests that were completed by the last call to
         multi.socket_action.
         """
         while True:
             num_q, ok_list, err_list = self._curl_multi.info_read()
             for curl in ok_list:
                 curl.waiter.switch(None)
             for curl, errnum, errmsg in err_list:
-                curl.waiter.throw(Exception('%s %s' % (errnum, errmsg)))
+                curl.waiter.throw(Exception('{} {}'.format(errnum, errmsg)))
             if num_q == 0:
                 break
 
 
 class GeventCurl(object):
     """
     Gevent compatible implementation of the pycurl.Curl class. Essentially a
     wrapper around pycurl.Curl with a customized perform method. It uses the
     GeventCurlMulti class to implement a blocking API to libcurl's "easy"
     interface.
     """
 
     # Reference to the GeventCurlMulti instance.
     _multi_instance = None
 
     def __init__(self):
         self._curl = pycurl.Curl()
 
     def __getattr__(self, item):
         """
         The pycurl.Curl class is final and we cannot subclass it. Therefore we
         are wrapping it and forward everything to it here.
         """
         return getattr(self._curl, item)
 
     @property
     def _multi(self):
         """
         Lazy property that returns the GeventCurlMulti instance. The value is
         cached as a class attribute. Therefore only one instance per process
         exists.
         """
         if GeventCurl._multi_instance is None:
             GeventCurl._multi_instance = GeventCurlMulti()
         return GeventCurl._multi_instance
 
     def perform(self):
         """
         This perform method is compatible with gevent because it uses gevent
         synchronization mechanisms to wait for the request to finish.
         """
         if getattr(self._curl, 'waiter', None) is not None:
             current = greenlet.getcurrent()
             msg = 'This curl object is already used by another greenlet, {}, \n' \
                   'this is {}'.format(self._curl.waiter, current)
             raise Exception(msg)
 
         waiter = self._curl.waiter = Waiter()
         try:
             self._multi.add_handle(self._curl)
             try:
                 return waiter.get()
             finally:
                 self._multi.remove_handle(self._curl)
         finally:
             del self._curl.waiter
 
 
 # Curl is originally imported from pycurl. At this point we override it with
 # our custom implementation.
 Curl = GeventCurl

rhodecode/lib/vcs/nodes.py

@@ -1,962 +1,960 b''
-
-
 # Copyright (C) 2014-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 """
 Module holding everything related to vcs nodes, with vcs2 architecture.
 """
 import functools
 import os
 import stat
 
 from zope.cachedescriptors.property import Lazy as LazyProperty
 
 from rhodecode.config.conf import LANGUAGES_EXTENSIONS_MAP
 from rhodecode.lib.str_utils import safe_str, safe_bytes
 from rhodecode.lib.hash_utils import md5
 from rhodecode.lib.vcs import path as vcspath
 from rhodecode.lib.vcs.backends.base import EmptyCommit, FILEMODE_DEFAULT
 from rhodecode.lib.vcs.conf.mtypes import get_mimetypes_db
 from rhodecode.lib.vcs.exceptions import NodeError, RemovedFileNodeError
 
 LARGEFILE_PREFIX = '.hglf'
 
 
 class NodeKind:
     SUBMODULE = -1
     DIR = 1
     FILE = 2
     LARGEFILE = 3
 
 
 class NodeState:
     ADDED = 'added'
     CHANGED = 'changed'
     NOT_CHANGED = 'not changed'
     REMOVED = 'removed'
 
 #TODO: not sure if that should be bytes or str ?
 # most probably bytes because content should be bytes and we check it
 BIN_BYTE_MARKER = b'\0'
 
 
 class NodeGeneratorBase(object):
     """
     Base class for removed added and changed filenodes, it's a lazy generator
     class that will create filenodes only on iteration or call
 
     The len method doesn't need to create filenodes at all
     """
 
     def __init__(self, current_paths, cs):
         self.cs = cs
         self.current_paths = current_paths
 
     def __call__(self):
         return [n for n in self]
 
     def __getitem__(self, key):
         if isinstance(key, slice):
             for p in self.current_paths[key.start:key.stop]:
                 yield self.cs.get_node(p)
 
     def __len__(self):
         return len(self.current_paths)
 
     def __iter__(self):
         for p in self.current_paths:
             yield self.cs.get_node(p)
 
 
 class AddedFileNodesGenerator(NodeGeneratorBase):
     """
     Class holding added files for current commit
     """
 
 
 class ChangedFileNodesGenerator(NodeGeneratorBase):
     """
     Class holding changed files for current commit
     """
 
 
 class RemovedFileNodesGenerator(NodeGeneratorBase):
     """
     Class holding removed files for current commit
     """
     def __iter__(self):
         for p in self.current_paths:
             yield RemovedFileNode(path=safe_bytes(p))
 
     def __getitem__(self, key):
         if isinstance(key, slice):
             for p in self.current_paths[key.start:key.stop]:
                 yield RemovedFileNode(path=safe_bytes(p))
 
 
 @functools.total_ordering
 class Node(object):
     """
     Simplest class representing file or directory on repository.  SCM backends
     should use ``FileNode`` and ``DirNode`` subclasses rather than ``Node``
     directly.
 
     Node's ``path`` cannot start with slash as we operate on *relative* paths
     only. Moreover, every single node is identified by the ``path`` attribute,
     so it cannot end with slash, too. Otherwise, path could lead to mistakes.
     """
     # RTLO marker allows swapping text, and certain
     # security attacks could be used with this
     RTLO_MARKER = "\u202E"
 
     commit = None
 
     def __init__(self, path: bytes, kind):
         self._validate_path(path)  # can throw exception if path is invalid
 
         self.bytes_path = path.rstrip(b'/')  # store for __repr__
         self.path = safe_str(self.bytes_path)  # we store paths as str
 
         if self.bytes_path == b'' and kind != NodeKind.DIR:
             raise NodeError("Only DirNode and its subclasses may be "
                             "initialized with empty path")
         self.kind = kind
 
         if self.is_root() and not self.is_dir():
             raise NodeError("Root node cannot be FILE kind")
 
     def __eq__(self, other):
         if type(self) is not type(other):
             return False
         for attr in ['name', 'path', 'kind']:
             if getattr(self, attr) != getattr(other, attr):
                 return False
         if self.is_file():
             # FileNode compare, we need to fallback to content compare
             return None
         else:
             # For DirNode's check without entering each dir
             self_nodes_paths = list(sorted(n.path for n in self.nodes))
             other_nodes_paths = list(sorted(n.path for n in self.nodes))
             if self_nodes_paths != other_nodes_paths:
                 return False
         return True
 
     def __lt__(self, other):
         if self.kind < other.kind:
             return True
         if self.kind > other.kind:
             return False
         if self.path < other.path:
             return True
         if self.path > other.path:
             return False
 
     # def __cmp__(self, other):
     #     """
     #     Comparator using name of the node, needed for quick list sorting.
     #     """
     #
     #     kind_cmp = cmp(self.kind, other.kind)
     #     if kind_cmp:
     #         if isinstance(self, SubModuleNode):
     #             # we make submodules equal to dirnode for "sorting" purposes
     #             return NodeKind.DIR
     #         return kind_cmp
     #     return cmp(self.name, other.name)
 
     def __repr__(self):
         maybe_path = getattr(self, 'path', 'UNKNOWN_PATH')
         return f'<{self.__class__.__name__} {maybe_path!r}>'
 
     def __str__(self):
         return self.name
 
     def _validate_path(self, path: bytes):
         self._assert_bytes(path)
 
         if path.startswith(b'/'):
             raise NodeError(
                 f"Cannot initialize Node objects with slash at "
                 f"the beginning as only relative paths are supported. "
                 f"Got {path}")
 
     def _assert_bytes(self, value):
         if not isinstance(value, bytes):
             raise TypeError(f"Bytes required as input, got {type(value)} of {value}.")
 
     @LazyProperty
     def parent(self):
         parent_path = self.get_parent_path()
         if parent_path:
             if self.commit:
                 return self.commit.get_node(parent_path)
             return DirNode(parent_path)
         return None
 
     @LazyProperty
     def str_path(self) -> str:
         return safe_str(self.path)
 
     @LazyProperty
     def has_rtlo(self):
         """Detects if a path has right-to-left-override marker"""
         return self.RTLO_MARKER in self.str_path
 
     @LazyProperty
     def dir_path(self):
         """
         Returns name of the directory from full path of this vcs node. Empty
         string is returned if there's no directory in the path
         """
         _parts = self.path.rstrip('/').rsplit('/', 1)
         if len(_parts) == 2:
             return _parts[0]
         return ''
 
     @LazyProperty
     def name(self):
         """
         Returns name of the node so if its path
         then only last part is returned.
         """
         return self.path.rstrip('/').split('/')[-1]
 
     @property
     def kind(self):
         return self._kind
 
     @kind.setter
     def kind(self, kind):
         if hasattr(self, '_kind'):
             raise NodeError("Cannot change node's kind")
         else:
             self._kind = kind
             # Post setter check (path's trailing slash)
             if self.path.endswith('/'):
                 raise NodeError("Node's path cannot end with slash")
 
     def get_parent_path(self) -> bytes:
         """
         Returns node's parent path or empty string if node is root.
         """
         if self.is_root():
             return b''
         str_path = vcspath.dirname(self.path.rstrip('/')) + '/'
 
         return safe_bytes(str_path)
 
     def is_file(self):
         """
         Returns ``True`` if node's kind is ``NodeKind.FILE``, ``False``
         otherwise.
         """
         return self.kind == NodeKind.FILE
 
     def is_dir(self):
         """
         Returns ``True`` if node's kind is ``NodeKind.DIR``, ``False``
         otherwise.
         """
         return self.kind == NodeKind.DIR
 
     def is_root(self):
         """
         Returns ``True`` if node is a root node and ``False`` otherwise.
         """
         return self.kind == NodeKind.DIR and self.path == ''
 
     def is_submodule(self):
         """
         Returns ``True`` if node's kind is ``NodeKind.SUBMODULE``, ``False``
         otherwise.
         """
         return self.kind == NodeKind.SUBMODULE
 
     def is_largefile(self):
         """
         Returns ``True`` if node's kind is ``NodeKind.LARGEFILE``, ``False``
         otherwise
         """
         return self.kind == NodeKind.LARGEFILE
 
     def is_link(self):
         if self.commit:
             return self.commit.is_link(self.path)
         return False
 
     @LazyProperty
     def added(self):
         return self.state is NodeState.ADDED
 
     @LazyProperty
     def changed(self):
         return self.state is NodeState.CHANGED
 
     @LazyProperty
     def not_changed(self):
         return self.state is NodeState.NOT_CHANGED
 
     @LazyProperty
     def removed(self):
         return self.state is NodeState.REMOVED
 
 
 class FileNode(Node):
     """
     Class representing file nodes.
 
     :attribute: path: path to the node, relative to repository's root
     :attribute: content: if given arbitrary sets content of the file
     :attribute: commit: if given, first time content is accessed, callback
     :attribute: mode: stat mode for a node. Default is `FILEMODE_DEFAULT`.
     """
     _filter_pre_load = []
 
     def __init__(self, path: bytes, content: bytes | None = None, commit=None, mode=None, pre_load=None):
         """
         Only one of ``content`` and ``commit`` may be given. Passing both
         would raise ``NodeError`` exception.
 
         :param path: relative path to the node
         :param content: content may be passed to constructor
         :param commit: if given, will use it to lazily fetch content
         :param mode: ST_MODE (i.e. 0100644)
         """
         if content and commit:
             raise NodeError("Cannot use both content and commit")
 
         super().__init__(path, kind=NodeKind.FILE)
 
         self.commit = commit
         if content and not isinstance(content, bytes):
             # File content is one thing that inherently must be bytes
             # we support passing str too, and convert the content
             content = safe_bytes(content)
         self._content = content
         self._mode = mode or FILEMODE_DEFAULT
 
         self._set_bulk_properties(pre_load)
 
     def __eq__(self, other):
         eq = super().__eq__(other)
         if eq is not None:
             return eq
         return self.content == other.content
 
     def __hash__(self):
         raw_id = getattr(self.commit, 'raw_id', '')
         return hash((self.path, raw_id))
 
     def __lt__(self, other):
         lt = super().__lt__(other)
         if lt is not None:
             return lt
         return self.content < other.content
 
     def __repr__(self):
         short_id = getattr(self.commit, 'short_id', '')
         return f'<{self.__class__.__name__} path={self.path!r}, short_id={short_id}>'
 
     def _set_bulk_properties(self, pre_load):
         if not pre_load:
             return
         pre_load = [entry for entry in pre_load
                     if entry not in self._filter_pre_load]
         if not pre_load:
             return
 
         remote = self.commit.get_remote()
         result = remote.bulk_file_request(self.commit.raw_id, self.path, pre_load)
 
         for attr, value in result.items():
             if attr == "flags":
                 self.__dict__['mode'] = safe_str(value)
             elif attr == "size":
                 self.__dict__['size'] = value
             elif attr == "data":
                 self.__dict__['_content'] = value
             elif attr == "is_binary":
                 self.__dict__['is_binary'] = value
             elif attr == "md5":
                 self.__dict__['md5'] = value
             else:
                 raise ValueError(f'Unsupported attr in bulk_property: {attr}')
 
     @LazyProperty
     def mode(self):
         """
         Returns lazily mode of the FileNode. If `commit` is not set, would
         use value given at initialization or `FILEMODE_DEFAULT` (default).
         """
         if self.commit:
             mode = self.commit.get_file_mode(self.path)
         else:
             mode = self._mode
         return mode
 
     @LazyProperty
     def raw_bytes(self) -> bytes:
         """
         Returns lazily the raw bytes of the FileNode.
         """
         if self.commit:
             if self._content is None:
                 self._content = self.commit.get_file_content(self.path)
             content = self._content
         else:
             content = self._content
         return content
 
     def content_uncached(self):
         """
         Returns lazily content of the FileNode.
         """
         if self.commit:
             content = self.commit.get_file_content(self.path)
         else:
             content = self._content
         return content
 
     def stream_bytes(self):
         """
         Returns an iterator that will stream the content of the file directly from
         vcsserver without loading it to memory.
         """
         if self.commit:
             return self.commit.get_file_content_streamed(self.path)
         raise NodeError("Cannot retrieve stream_bytes without related commit attribute")
 
     def metadata_uncached(self):
         """
         Returns md5, binary flag of the file node, without any cache usage.
         """
 
         content = self.content_uncached()
 
         is_binary = bool(content and BIN_BYTE_MARKER in content)
         size = 0
         if content:
             size = len(content)
 
         return is_binary, md5(content), size, content
 
     @LazyProperty
     def content(self) -> bytes:
         """
         Returns lazily content of the FileNode.
         """
         content = self.raw_bytes
         if content and not isinstance(content, bytes):
             raise ValueError(f'Content is of type {type(content)} instead of bytes')
         return content
 
     @LazyProperty
     def str_content(self) -> str:
         return safe_str(self.raw_bytes)
 
     @LazyProperty
     def size(self):
         if self.commit:
             return self.commit.get_file_size(self.path)
         raise NodeError(
             "Cannot retrieve size of the file without related "
             "commit attribute")
 
     @LazyProperty
     def message(self):
         if self.commit:
             return self.last_commit.message
         raise NodeError(
             "Cannot retrieve message of the file without related "
             "commit attribute")
 
     @LazyProperty
     def last_commit(self):
         if self.commit:
             pre_load = ["author", "date", "message", "parents"]
             return self.commit.get_path_commit(self.path, pre_load=pre_load)
         raise NodeError(
             "Cannot retrieve last commit of the file without "
             "related commit attribute")
 
     def get_mimetype(self):
         """
         Mimetype is calculated based on the file's content. If ``_mimetype``
         attribute is available, it will be returned (backends which store
         mimetypes or can easily recognize them, should set this private
         attribute to indicate that type should *NOT* be calculated).
         """
 
         if hasattr(self, '_mimetype'):
             if (isinstance(self._mimetype, (tuple, list)) and
                     len(self._mimetype) == 2):
                 return self._mimetype
             else:
                 raise NodeError('given _mimetype attribute must be an 2 '
                                 'element list or tuple')
 
         db = get_mimetypes_db()
         mtype, encoding = db.guess_type(self.name)
 
         if mtype is None:
             if not self.is_largefile() and self.is_binary:
                 mtype = 'application/octet-stream'
                 encoding = None
             else:
                 mtype = 'text/plain'
                 encoding = None
 
                 # try with pygments
                 try:
                     from pygments.lexers import get_lexer_for_filename
                     mt = get_lexer_for_filename(self.name).mimetypes
                 except Exception:
                     mt = None
 
                 if mt:
                     mtype = mt[0]
 
         return mtype, encoding
 
     @LazyProperty
     def mimetype(self):
         """
         Wrapper around full mimetype info. It returns only type of fetched
         mimetype without the encoding part. use get_mimetype function to fetch
         full set of (type,encoding)
         """
         return self.get_mimetype()[0]
 
     @LazyProperty
     def mimetype_main(self):
         return self.mimetype.split('/')[0]
 
     @classmethod
     def get_lexer(cls, filename, content=None):
         from pygments import lexers
 
         extension = filename.split('.')[-1]
         lexer = None
 
         try:
             lexer = lexers.guess_lexer_for_filename(
                 filename, content, stripnl=False)
         except lexers.ClassNotFound:
             pass
 
         # try our EXTENSION_MAP
         if not lexer:
             try:
                 lexer_class = LANGUAGES_EXTENSIONS_MAP.get(extension)
                 if lexer_class:
                     lexer = lexers.get_lexer_by_name(lexer_class[0])
             except lexers.ClassNotFound:
                 pass
 
         if not lexer:
             lexer = lexers.TextLexer(stripnl=False)
 
         return lexer
 
     @LazyProperty
     def lexer(self):
         """
         Returns pygment's lexer class. Would try to guess lexer taking file's
         content, name and mimetype.
         """
         # TODO: this is more proper, but super heavy on investigating the type based on the content
         #self.get_lexer(self.name, self.content)
 
         return self.get_lexer(self.name)
 
     @LazyProperty
     def lexer_alias(self):
         """
         Returns first alias of the lexer guessed for this file.
         """
         return self.lexer.aliases[0]
 
     @LazyProperty
     def history(self):
         """
         Returns a list of commit for this file in which the file was changed
         """
         if self.commit is None:
             raise NodeError('Unable to get commit for this FileNode')
         return self.commit.get_path_history(self.path)
 
     @LazyProperty
     def annotate(self):
         """
         Returns a list of three element tuples with lineno, commit and line
         """
         if self.commit is None:
             raise NodeError('Unable to get commit for this FileNode')
         pre_load = ["author", "date", "message", "parents"]
         return self.commit.get_file_annotate(self.path, pre_load=pre_load)
 
     @LazyProperty
     def state(self):
         if not self.commit:
             raise NodeError(
                 "Cannot check state of the node if it's not "
                 "linked with commit")
         elif self.path in (node.path for node in self.commit.added):
             return NodeState.ADDED
         elif self.path in (node.path for node in self.commit.changed):
             return NodeState.CHANGED
         else:
             return NodeState.NOT_CHANGED
 
     @LazyProperty
     def is_binary(self):
         """
         Returns True if file has binary content.
         """
         if self.commit:
             return self.commit.is_node_binary(self.path)
         else:
             raw_bytes = self._content
             return bool(raw_bytes and BIN_BYTE_MARKER in raw_bytes)
 
     @LazyProperty
     def md5(self):
         """
         Returns md5 of the file node.
         """
 
         if self.commit:
             return self.commit.node_md5_hash(self.path)
         else:
             raw_bytes = self._content
             # TODO: this sucks, we're computing md5 on potentially super big stream data...
             return md5(raw_bytes)
 
     @LazyProperty
     def extension(self):
         """Returns filenode extension"""
         return self.name.split('.')[-1]
 
     @property
     def is_executable(self):
         """
         Returns ``True`` if file has executable flag turned on.
         """
         return bool(self.mode & stat.S_IXUSR)
 
     def get_largefile_node(self):
         """
         Try to return a Mercurial FileNode from this node. It does internal
         checks inside largefile store, if that file exist there it will
         create special instance of LargeFileNode which can get content from
         LF store.
         """
         if self.commit:
             return self.commit.get_largefile_node(self.path)
 
     def count_lines(self, content: str | bytes, count_empty=False):
         if isinstance(content, str):
             newline_marker = '\n'
         elif isinstance(content, bytes):
             newline_marker = b'\n'
         else:
             raise ValueError('content must be bytes or str got {type(content)} instead')
 
         if count_empty:
             all_lines = 0
             empty_lines = 0
             for line in content.splitlines(True):
                 if line == newline_marker:
                     empty_lines += 1
                 all_lines += 1
 
             return all_lines, all_lines - empty_lines
         else:
             # fast method
             empty_lines = all_lines = content.count(newline_marker)
             if all_lines == 0 and content:
                 # one-line without a newline
                 empty_lines = all_lines = 1
 
         return all_lines, empty_lines
 
     def lines(self, count_empty=False):
         all_lines, empty_lines = 0, 0
 
         if not self.is_binary:
             content = self.content
             all_lines, empty_lines = self.count_lines(content, count_empty=count_empty)
         return all_lines, empty_lines
 
 
 class RemovedFileNode(FileNode):
     """
     Dummy FileNode class - trying to access any public attribute except path,
     name, kind or state (or methods/attributes checking those two) would raise
     RemovedFileNodeError.
     """
     ALLOWED_ATTRIBUTES = [
         'name', 'path', 'state', 'is_root', 'is_file', 'is_dir', 'kind',
         'added', 'changed', 'not_changed', 'removed', 'bytes_path'
     ]
 
     def __init__(self, path):
         """
         :param path: relative path to the node
         """
         super().__init__(path=path)
 
     def __getattribute__(self, attr):
         if attr.startswith('_') or attr in RemovedFileNode.ALLOWED_ATTRIBUTES:
             return super().__getattribute__(attr)
         raise RemovedFileNodeError(f"Cannot access attribute {attr} on RemovedFileNode. Not in allowed attributes")
 
     @LazyProperty
     def state(self):
         return NodeState.REMOVED
 
 
 class DirNode(Node):
     """
     DirNode stores list of files and directories within this node.
     Nodes may be used standalone but within repository context they
     lazily fetch data within same repository's commit.
     """
 
     def __init__(self, path, nodes=(), commit=None, default_pre_load=None):
         """
         Only one of ``nodes`` and ``commit`` may be given. Passing both
         would raise ``NodeError`` exception.
 
         :param path: relative path to the node
         :param nodes: content may be passed to constructor
         :param commit: if given, will use it to lazily fetch content
         """
         if nodes and commit:
             raise NodeError("Cannot use both nodes and commit")
         super().__init__(path, NodeKind.DIR)
         self.commit = commit
         self._nodes = nodes
         self.default_pre_load = default_pre_load or ['is_binary', 'size']
 
     def __iter__(self):
         yield from self.nodes
 
     def __eq__(self, other):
         eq = super().__eq__(other)
         if eq is not None:
             return eq
         # check without entering each dir
         self_nodes_paths = list(sorted(n.path for n in self.nodes))
         other_nodes_paths = list(sorted(n.path for n in self.nodes))
         return self_nodes_paths == other_nodes_paths
 
     def __lt__(self, other):
         lt = super().__lt__(other)
         if lt is not None:
             return lt
         # check without entering each dir
         self_nodes_paths = list(sorted(n.path for n in self.nodes))
         other_nodes_paths = list(sorted(n.path for n in self.nodes))
         return self_nodes_paths < other_nodes_paths
 
     @LazyProperty
     def content(self):
         raise NodeError(f"{self} represents a dir and has no `content` attribute")
 
     @LazyProperty
     def nodes(self):
         if self.commit:
             nodes = self.commit.get_nodes(self.path, pre_load=self.default_pre_load)
         else:
             nodes = self._nodes
         self._nodes_dict = {node.path: node for node in nodes}
         return sorted(nodes)
 
     @LazyProperty
     def files(self):
         return sorted(node for node in self.nodes if node.is_file())
 
     @LazyProperty
     def dirs(self):
         return sorted(node for node in self.nodes if node.is_dir())
 
     def get_node(self, path):
         """
         Returns node from within this particular ``DirNode``, so it is now
         allowed to fetch, i.e. node located at 'docs/api/index.rst' from node
         'docs'. In order to access deeper nodes one must fetch nodes between
         them first - this would work::
 
            docs = root.get_node('docs')
            docs.get_node('api').get_node('index.rst')
 
         :param: path - relative to the current node
 
         .. note::
            To access lazily (as in example above) node have to be initialized
            with related commit object - without it node is out of
            context and may know nothing about anything else than nearest
            (located at same level) nodes.
         """
         try:
             path = path.rstrip('/')
             if path == '':
                 raise NodeError("Cannot retrieve node without path")
             self.nodes  # access nodes first in order to set _nodes_dict
             paths = path.split('/')
             if len(paths) == 1:
                 if not self.is_root():
                     path = '/'.join((self.path, paths[0]))
                 else:
                     path = paths[0]
                 return self._nodes_dict[path]
             elif len(paths) > 1:
                 if self.commit is None:
                     raise NodeError("Cannot access deeper nodes without commit")
                 else:
                     path1, path2 = paths[0], '/'.join(paths[1:])
                     return self.get_node(path1).get_node(path2)
             else:
                 raise KeyError
         except KeyError:
             raise NodeError(f"Node does not exist at {path}")
 
     @LazyProperty
     def state(self):
         raise NodeError("Cannot access state of DirNode")
 
     @LazyProperty
     def size(self):
         size = 0
         for root, dirs, files in self.commit.walk(self.path):
             for f in files:
                 size += f.size
 
         return size
 
     @LazyProperty
     def last_commit(self):
         if self.commit:
             pre_load = ["author", "date", "message", "parents"]
             return self.commit.get_path_commit(self.path, pre_load=pre_load)
         raise NodeError(
             "Cannot retrieve last commit of the file without "
             "related commit attribute")
 
     def __repr__(self):
         short_id = getattr(self.commit, 'short_id', '')
         return f'<{self.__class__.__name__} {self.path!r} @ {short_id}>'
 
 
 class RootNode(DirNode):
     """
     DirNode being the root node of the repository.
     """
 
     def __init__(self, nodes=(), commit=None):
         super().__init__(path=b'', nodes=nodes, commit=commit)
 
     def __repr__(self):
         return f'<{self.__class__.__name__}>'
 
 
 class SubModuleNode(Node):
     """
     represents a SubModule of Git or SubRepo of Mercurial
     """
     is_binary = False
     size = 0
 
     def __init__(self, name, url=None, commit=None, alias=None):
         self.path = name
         self.kind = NodeKind.SUBMODULE
         self.alias = alias
 
         # we have to use EmptyCommit here since this can point to svn/git/hg
         # submodules we cannot get from repository
         self.commit = EmptyCommit(str(commit), alias=alias)
         self.url = url or self._extract_submodule_url()
 
     def __repr__(self):
         short_id = getattr(self.commit, 'short_id', '')
         return f'<{self.__class__.__name__} {self.path!r} @ {short_id}>'
 
     def _extract_submodule_url(self):
         # TODO: find a way to parse gits submodule file and extract the
         # linking URL
         return self.path
 
     @LazyProperty
     def name(self):
         """
         Returns name of the node so if its path
         then only last part is returned.
         """
         org = safe_str(self.path.rstrip('/').split('/')[-1])
         return f'{org} @ {self.commit.short_id}'
 
 
 class LargeFileNode(FileNode):
 
     def __init__(self, path, url=None, commit=None, alias=None, org_path=None):
         self._validate_path(path)  # can throw exception if path is invalid
         self.org_path = org_path  # as stored in VCS as LF pointer
 
         self.bytes_path = path.rstrip(b'/')  # store for __repr__
         self.path = safe_str(self.bytes_path)  # we store paths as str
 
         self.kind = NodeKind.LARGEFILE
         self.alias = alias
         self._content = b''
 
     def _validate_path(self, path: bytes):
         """
         we override check since the LargeFileNode path is system absolute, but we check for bytes only
         """
         self._assert_bytes(path)
 
     def __repr__(self):
         return f'<{self.__class__.__name__} {self.org_path} -> {self.path!r}>'
 
     @LazyProperty
     def size(self):
         return os.stat(self.path).st_size
 
     @LazyProperty
     def raw_bytes(self):
         with open(self.path, 'rb') as f:
             content = f.read()
         return content
 
     @LazyProperty
     def name(self):
         """
         Overwrites name to be the org lf path
         """
         return self.org_path
 
     def stream_bytes(self):
         with open(self.path, 'rb') as stream:
             while True:
                 data = stream.read(16 * 1024)
                 if not data:
                     break
                 yield data