kallithea Changeset - 5725fa4cfecd

Changeset - 5725fa4cfecd

Parent rev.

Child rev.

[Not reviewed]

default

0 14 0

Mads Kiilerich - 6 years ago 2019-08-04 00:07:10
mads@kiilerich.com

Grafted from: 7cc8ba8bc555

cleanup: minimize use of lambda expressions - we have 'def' for that purpose

Fix some flake8 warnings "E731 do not assign a lambda expression, use a def".

14 files changed with 71 insertions and 70 deletions:

kallithea/controllers/admin/repo_groups.py

kallithea/controllers/admin/user_groups.py

kallithea/controllers/admin/users.py

kallithea/lib/auth_modules/auth_ldap.py

kallithea/lib/celerylib/tasks.py

kallithea/lib/helpers.py

kallithea/lib/rcmail/response.py

kallithea/lib/vcs/backends/git/changeset.py

kallithea/lib/vcs/backends/git/repository.py

kallithea/lib/vcs/conf/settings.py

kallithea/lib/vcs/utils/paths.py

kallithea/model/repo.py

kallithea/model/validators.py

setup.py

0 comments (0 inline, 0 general)

kallithea/controllers/admin/repo_groups.py

➞

Show inline comments

@@ @@ -59,105 +59,104 @@ class RepoGroupsController(BaseControlle @@
         super(RepoGroupsController, self)._before(*args, **kwargs)
     def __load_defaults(self, extras=(), exclude=()):
         """extras is used for keeping current parent ignoring permissions
         exclude is used for not moving group to itself TODO: also exclude descendants
         Note: only admin can create top level groups
         """
         repo_groups = AvailableRepoGroupChoices([], 'admin', extras)
         exclude_group_ids = set(rg.group_id for rg in exclude)
         c.repo_groups = [rg for rg in repo_groups
                          if rg[0] not in exclude_group_ids]
     def __load_data(self, group_id):
         """
         Load defaults settings for edit, and update
         :param group_id:
         """
         repo_group = RepoGroup.get_or_404(group_id)
         data = repo_group.get_dict()
         data['group_name'] = repo_group.name
         # fill repository group users
         for p in repo_group.repo_group_to_perm:
             data.update({'u_perm_%s' % p.user.username:
                              p.permission.permission_name})
         # fill repository group groups
         for p in repo_group.users_group_to_perm:
             data.update({'g_perm_%s' % p.users_group.users_group_name:
                              p.permission.permission_name})
         return data
     def _revoke_perms_on_yourself(self, form_result):
         _up = [u for u in form_result['perms_updates'] if request.authuser.username == u[0]]
         _new = [u for u in form_result['perms_new'] if request.authuser.username == u[0]]
         if _new and _new[0][1] != 'group.admin' or _up and _up[0][1] != 'group.admin':
             return True
         return False
     def index(self, format='html'):
         _list = RepoGroup.query(sorted=True).all()
         group_iter = RepoGroupList(_list, perm_level='admin')
         repo_groups_data = []
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         repo_group_name = lambda repo_group_name, children_groups: (
             template.get_def("repo_group_name")
             .render_unicode(repo_group_name, children_groups, _=_, h=h, c=c)
+        )
         repo_group_actions = lambda repo_group_id, repo_group_name, gr_count: (
             template.get_def("repo_group_actions")
             .render_unicode(repo_group_id, repo_group_name, gr_count, _=_, h=h, c=c,
                     ungettext=ungettext)
+        )
         def repo_group_name(repo_group_name, children_groups):
             return template.get_def("repo_group_name") \
                 .render_unicode(repo_group_name, children_groups, _=_, h=h, c=c)
         def repo_group_actions(repo_group_id, repo_group_name, gr_count):
             return template.get_def("repo_group_actions") \
                 .render_unicode(repo_group_id, repo_group_name, gr_count, _=_, h=h, c=c,
                         ungettext=ungettext)
         for repo_gr in group_iter:
             children_groups = [g.name for g in repo_gr.parents] + [repo_gr.name]
             repo_count = repo_gr.repositories.count()
             repo_groups_data.append({
                 "raw_name": repo_gr.group_name,
                 "group_name": repo_group_name(repo_gr.group_name, children_groups),
                 "desc": h.escape(repo_gr.group_description),
                 "repos": repo_count,
                 "owner": h.person(repo_gr.owner),
                 "action": repo_group_actions(repo_gr.group_id, repo_gr.group_name,
                                              repo_count)
             })
         c.data = {
             "sort": None,
             "dir": "asc",
             "records": repo_groups_data
+        }
         return render('admin/repo_groups/repo_groups.html')
     def create(self):
         self.__load_defaults()
         # permissions for can create group based on parent_id are checked
         # here in the Form
         repo_group_form = RepoGroupForm(repo_groups=c.repo_groups)
         form_result = None
         try:
             form_result = repo_group_form.to_python(dict(request.POST))
             gr = RepoGroupModel().create(
                 group_name=form_result['group_name'],
                 group_description=form_result['group_description'],
                 parent=form_result['parent_group_id'],
                 owner=request.authuser.user_id, # TODO: make editable
                 copy_permissions=form_result['group_copy_permissions']
+            )
             Session().commit()
             # TODO: in future action_logger(, '', '', '')
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/repo_groups/repo_group_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)

kallithea/controllers/admin/user_groups.py

➞

Show inline comments

@@ @@ -44,106 +44,105 @@ from kallithea.lib.base import BaseContr @@
 from kallithea.lib.exceptions import RepoGroupAssignmentError, UserGroupsAssignedException
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import safe_int, safe_str
 from kallithea.model.db import User, UserGroup, UserGroupRepoGroupToPerm, UserGroupRepoToPerm, UserGroupToPerm
 from kallithea.model.forms import CustomDefaultPermissionsForm, UserGroupForm, UserGroupPermsForm
 from kallithea.model.meta import Session
 from kallithea.model.scm import UserGroupList
 from kallithea.model.user_group import UserGroupModel
 log = logging.getLogger(__name__)
 class UserGroupsController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     @LoginRequired(allow_default_user=True)
     def _before(self, *args, **kwargs):
         super(UserGroupsController, self)._before(*args, **kwargs)
     def __load_data(self, user_group_id):
         c.group_members_obj = sorted((x.user for x in c.user_group.members),
                                      key=lambda u: u.username.lower())
         c.group_members = [(x.user_id, x.username) for x in c.group_members_obj]
         c.available_members = sorted(((x.user_id, x.username) for x in
                                       User.query().all()),
                                      key=lambda u: u[1].lower())
     def __load_defaults(self, user_group_id):
         """
         Load defaults settings for edit, and update
         :param user_group_id:
         """
         user_group = UserGroup.get_or_404(user_group_id)
         data = user_group.get_dict()
         return data
     def index(self, format='html'):
         _list = UserGroup.query() \
                         .order_by(func.lower(UserGroup.users_group_name)) \
                         .all()
         group_iter = UserGroupList(_list, perm_level='admin')
         user_groups_data = []
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         user_group_name = lambda user_group_id, user_group_name: (
             template.get_def("user_group_name")
             .render_unicode(user_group_id, user_group_name, _=_, h=h, c=c)
+        )
         user_group_actions = lambda user_group_id, user_group_name: (
             template.get_def("user_group_actions")
             .render_unicode(user_group_id, user_group_name, _=_, h=h, c=c)
+        )
         def user_group_name(user_group_id, user_group_name):
             return template.get_def("user_group_name") \
                 .render_unicode(user_group_id, user_group_name, _=_, h=h, c=c)
         def user_group_actions(user_group_id, user_group_name):
             return template.get_def("user_group_actions") \
                 .render_unicode(user_group_id, user_group_name, _=_, h=h, c=c)
         for user_gr in group_iter:
             user_groups_data.append({
                 "raw_name": user_gr.users_group_name,
                 "group_name": user_group_name(user_gr.users_group_id,
                                               user_gr.users_group_name),
                 "desc": h.escape(user_gr.user_group_description),
                 "members": len(user_gr.members),
                 "active": h.boolicon(user_gr.users_group_active),
                 "owner": h.person(user_gr.owner.username),
                 "action": user_group_actions(user_gr.users_group_id, user_gr.users_group_name)
             })
         c.data = {
             "sort": None,
             "dir": "asc",
             "records": user_groups_data
+        }
         return render('admin/user_groups/user_groups.html')
     @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')
     def create(self):
         users_group_form = UserGroupForm()()
         try:
             form_result = users_group_form.to_python(dict(request.POST))
             ug = UserGroupModel().create(name=form_result['users_group_name'],
                                          description=form_result['user_group_description'],
                                          owner=request.authuser.user_id,
                                          active=form_result['users_group_active'])
             gr = form_result['users_group_name']
             action_logger(request.authuser,
                           'admin_created_users_group:%s' % gr,
                           None, request.ip_addr)
             h.flash(h.HTML(_('Created user group %s')) % h.link_to(gr, url('edit_users_group', id=ug.users_group_id)),
                 category='success')
             Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/user_groups/user_group_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except Exception:
             log.error(traceback.format_exc())
             h.flash(_('Error occurred during creation of user group %s')
                     % request.POST.get('users_group_name'), category='error')

kallithea/controllers/admin/users.py

➞

Show inline comments

@@ @@ -30,103 +30,103 @@ import traceback @@
 import formencode
 from formencode import htmlfill
 from sqlalchemy.sql.expression import func
 from tg import app_globals, request
 from tg import tmpl_context as c
 from tg.i18n import ugettext as _
 from webob.exc import HTTPFound, HTTPNotFound
 from kallithea.config.routing import url
 from kallithea.lib import auth_modules
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import AuthUser, HasPermissionAnyDecorator, LoginRequired
 from kallithea.lib.base import BaseController, IfSshEnabled, render
 from kallithea.lib.exceptions import DefaultUserException, UserCreationError, UserOwnsReposException
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import datetime_to_time, generate_api_key, safe_int
 from kallithea.model.api_key import ApiKeyModel
 from kallithea.model.db import User, UserEmailMap, UserIpMap, UserToPerm
 from kallithea.model.forms import CustomDefaultPermissionsForm, UserForm
 from kallithea.model.meta import Session
 from kallithea.model.ssh_key import SshKeyModel, SshKeyModelException
 from kallithea.model.user import UserModel
 log = logging.getLogger(__name__)
 class UsersController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     @LoginRequired()
     @HasPermissionAnyDecorator('hg.admin')
     def _before(self, *args, **kwargs):
         super(UsersController, self)._before(*args, **kwargs)
     def index(self, format='html'):
         c.users_list = User.query().order_by(User.username) \
                         .filter_by(is_default_user=False) \
                         .order_by(func.lower(User.username)) \
                         .all()
         users_data = []
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         grav_tmpl = '<div class="gravatar">%s</div>'
         username = lambda user_id, username: (
                 template.get_def("user_name")
                 .render_unicode(user_id, username, _=_, h=h, c=c))
         def username(user_id, username):
             return template.get_def("user_name") \
                 .render_unicode(user_id, username, _=_, h=h, c=c)
         user_actions = lambda user_id, username: (
                 template.get_def("user_actions")
                 .render_unicode(user_id, username, _=_, h=h, c=c))
         def user_actions(user_id, username):
             return template.get_def("user_actions") \
                 .render_unicode(user_id, username, _=_, h=h, c=c)
         for user in c.users_list:
             users_data.append({
                 "gravatar": grav_tmpl % h.gravatar(user.email, size=20),
                 "raw_name": user.username,
                 "username": username(user.user_id, user.username),
                 "firstname": h.escape(user.name),
                 "lastname": h.escape(user.lastname),
                 "last_login": h.fmt_date(user.last_login),
                 "last_login_raw": datetime_to_time(user.last_login),
                 "active": h.boolicon(user.active),
                 "admin": h.boolicon(user.admin),
                 "extern_type": user.extern_type,
                 "extern_name": user.extern_name,
                 "action": user_actions(user.user_id, user.username),
             })
         c.data = {
             "sort": None,
             "dir": "asc",
             "records": users_data
+        }
         return render('admin/users/users.html')
     def create(self):
         c.default_extern_type = User.DEFAULT_AUTH_TYPE
         c.default_extern_name = ''
         user_model = UserModel()
         user_form = UserForm()()
         try:
             form_result = user_form.to_python(dict(request.POST))
             user = user_model.create(form_result)
             action_logger(request.authuser, 'admin_created_user:%s' % user.username,
                           None, request.ip_addr)
             h.flash(_('Created user %s') % user.username,
                     category='success')
             Session().commit()
         except formencode.Invalid as errors:
             return htmlfill.render(
                 render('admin/users/user_add.html'),
                 defaults=errors.value,
                 errors=errors.error_dict or {},
                 prefix_error=False,
                 encoding="UTF-8",
                 force_defaults=False)
         except UserCreationError as e:
             h.flash(e, 'error')

kallithea/lib/auth_modules/auth_ldap.py

➞

Show inline comments

@@ @@ -282,78 +282,79 @@ class KallitheaAuthPlugin(auth_modules.K @@
+        ]
         return settings
     def use_fake_password(self):
         return True
     def auth(self, userobj, username, password, settings, **kwargs):
         """
         Given a user object (which may be null), username, a plaintext password,
         and a settings object (containing all the keys needed as listed in settings()),
         authenticate this user's login attempt.
         Return None on failure. On success, return a dictionary of the form:
             see: KallitheaAuthPluginBase.auth_func_attrs
         This is later validated for correctness
         """
         if not username or not password:
             log.debug('Empty username or password skipping...')
             return None
         kwargs = {
             'server': settings.get('host', ''),
             'base_dn': settings.get('base_dn', ''),
             'port': settings.get('port'),
             'bind_dn': settings.get('dn_user'),
             'bind_pass': settings.get('dn_pass'),
             'tls_kind': settings.get('tls_kind'),
             'tls_reqcert': settings.get('tls_reqcert'),
             'cacertdir': settings.get('cacertdir'),
             'ldap_filter': settings.get('filter'),
             'search_scope': settings.get('search_scope'),
             'attr_login': settings.get('attr_login'),
             'ldap_version': 3,
+        }
         if kwargs['bind_dn'] and not kwargs['bind_pass']:
             log.debug('Using dynamic binding.')
             kwargs['bind_dn'] = kwargs['bind_dn'].replace('$login', username)
             kwargs['bind_pass'] = password
         log.debug('Checking for ldap authentication')
         try:
             aldap = AuthLdap(**kwargs)
             (user_dn, ldap_attrs) = aldap.authenticate_ldap(username, password)
             log.debug('Got ldap DN response %s', user_dn)
             get_ldap_attr = lambda k: ldap_attrs.get(settings.get(k), [''])[0]
             def get_ldap_attr(k):
                 return ldap_attrs.get(settings.get(k), [''])[0]
             # old attrs fetched from Kallithea database
             admin = getattr(userobj, 'admin', False)
             email = getattr(userobj, 'email', '')
             firstname = getattr(userobj, 'firstname', '')
             lastname = getattr(userobj, 'lastname', '')
             user_data = {
                 'username': username,
                 'firstname': get_ldap_attr('attr_firstname') or firstname,
                 'lastname': get_ldap_attr('attr_lastname') or lastname,
                 'groups': [],
                 'email': get_ldap_attr('attr_email') or email,
                 'admin': admin,
                 'extern_name': user_dn,
+            }
             log.info('user %s authenticated correctly', user_data['username'])
             return user_data
         except LdapUsernameError:
             log.info('Error authenticating %s with LDAP: User not found', username)
         except LdapPasswordError:
             log.info('Error authenticating %s with LDAP: Password error', username)
         except LdapImportError:
             log.error('Error authenticating %s with LDAP: LDAP not available', username)
         return None
     def get_managed_fields(self):
         return ['username', 'firstname', 'lastname', 'email', 'password']

kallithea/lib/celerylib/tasks.py

➞

Show inline comments

@@ @@ -21,112 +21,114 @@ by celery daemon @@
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Oct 6, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import email.utils
 import os
 import traceback
 from collections import OrderedDict
 from operator import itemgetter
 from time import mktime
 import celery.utils.log
 from tg import config
 import kallithea
 from kallithea.lib import celerylib, ext_json
 from kallithea.lib.helpers import person
 from kallithea.lib.hooks import log_create_repository
 from kallithea.lib.rcmail.smtp_mailer import SmtpMailer
 from kallithea.lib.utils import action_logger
 from kallithea.lib.utils2 import ascii_bytes, str2bool
 from kallithea.lib.vcs.utils import author_email
 from kallithea.model.db import RepoGroup, Repository, Statistics, User
 __all__ = ['whoosh_index', 'get_commits_stats', 'send_email']
 log = celery.utils.log.get_task_logger(__name__)
 @celerylib.task
 @celerylib.locked_task
 @celerylib.dbsession
 def whoosh_index(repo_location, full_index):
     from kallithea.lib.indexers.daemon import WhooshIndexingDaemon
     celerylib.get_session() # initialize database connection
     index_location = config['index_dir']
     WhooshIndexingDaemon(index_location=index_location,
                          repo_location=repo_location) \
                          .run(full_index=full_index)
 # for js data compatibility cleans the key for person from '
 def akc(k):
     return person(k).replace('"', '')
 @celerylib.task
 @celerylib.dbsession
 def get_commits_stats(repo_name, ts_min_y, ts_max_y, recurse_limit=100):
     DBS = celerylib.get_session()
     lockkey = celerylib.__get_lockkey('get_commits_stats', repo_name, ts_min_y,
                             ts_max_y)
     lockkey_path = config.get('cache_dir') or config['app_conf']['cache_dir']  # Backward compatibility for TurboGears < 2.4
     log.info('running task with lockkey %s', lockkey)
     try:
         lock = celerylib.DaemonLock(os.path.join(lockkey_path, lockkey))
         # for js data compatibility cleans the key for person from '
         akc = lambda k: person(k).replace('"', "")
         co_day_auth_aggr = {}
         commits_by_day_aggregate = {}
         repo = Repository.get_by_repo_name(repo_name)
         if repo is None:
             return True
         repo = repo.scm_instance
         repo_size = repo.count()
         # return if repo have no revisions
         if repo_size < 1:
             lock.release()
             return True
         skip_date_limit = True
         parse_limit = int(config.get('commit_parse_limit'))
         last_rev = None
         last_cs = None
         timegetter = itemgetter('time')
         dbrepo = DBS.query(Repository) \
             .filter(Repository.repo_name == repo_name).scalar()
         cur_stats = DBS.query(Statistics) \
             .filter(Statistics.repository == dbrepo).scalar()
         if cur_stats is not None:
             last_rev = cur_stats.stat_on_revision
         if last_rev == repo.get_changeset().revision and repo_size > 1:
             # pass silently without any work if we're not on first revision or
             # current state of parsing revision(from db marker) is the
             # last revision
             lock.release()
             return True
         if cur_stats:
             commits_by_day_aggregate = OrderedDict(ext_json.loads(
                                         cur_stats.commit_activity_combined))
             co_day_auth_aggr = ext_json.loads(cur_stats.commit_activity)
         log.debug('starting parsing %s', parse_limit)
         last_rev = last_rev + 1 if last_rev and last_rev >= 0 else 0
         log.debug('Getting revisions from %s to %s',
              last_rev, last_rev + parse_limit
+        )
         for cs in repo[last_rev:last_rev + parse_limit]:
             log.debug('parsing %s', cs)
             last_cs = cs  # remember last parsed changeset

kallithea/lib/helpers.py

➞

Show inline comments

@@ @@ -457,205 +457,210 @@ def _session_flash_messages(append=None, @@
         if append is None:  # common fast path - also used for clearing empty queue
             return []  # don't bother saving
         flash_messages = []
         session[key] = flash_messages
     if append is not None and append not in flash_messages:
         flash_messages.append(append)
     if clear:
         session.pop(key, None)
     session.save()
     return flash_messages
 def flash(message, category, logf=None):
     """
     Show a message to the user _and_ log it through the specified function
     category: notice (default), warning, error, success
     logf: a custom log function - such as log.debug
     logf defaults to log.info, unless category equals 'success', in which
     case logf defaults to log.debug.
     """
     assert category in ('error', 'success', 'warning'), category
     if hasattr(message, '__html__'):
         # render to HTML for storing in cookie
         safe_message = str(message)
     else:
         # Apply str - the message might be an exception with __str__
         # Escape, so we can trust the result without further escaping, without any risk of injection
         safe_message = html_escape(str(message))
     if logf is None:
         logf = log.info
         if category == 'success':
             logf = log.debug
     logf('Flash %s: %s', category, safe_message)
     _session_flash_messages(append=(category, safe_message))
 def pop_flash_messages():
     """Return all accumulated messages and delete them from the session.
     The return value is a list of ``Message`` objects.
     """
     return [_Message(category, message) for category, message in _session_flash_messages(clear=True)]
 age = lambda x, y=False: _age(x, y)
 capitalize = lambda x: x.capitalize()
 def age(x, y=False):
     return _age(x, y)
 def capitalize(x):
     return x.capitalize()
 email = author_email
 short_id = lambda x: x[:12]
 hide_credentials = lambda x: ''.join(credentials_filter(x))
 def short_id(x):
     return x[:12]
 def hide_credentials(x):
     return ''.join(credentials_filter(x))
 def show_id(cs):
     """
     Configurable function that shows ID
     by default it's r123:fffeeefffeee
     :param cs: changeset instance
     """
     from kallithea import CONFIG
     def_len = safe_int(CONFIG.get('show_sha_length', 12))
     show_rev = str2bool(CONFIG.get('show_revision_number', False))
     raw_id = cs.raw_id[:def_len]
     if show_rev:
         return 'r%s:%s' % (cs.revision, raw_id)
     else:
         return raw_id
 def fmt_date(date):
     if date:
         return date.strftime("%Y-%m-%d %H:%M:%S")
     return ""
 def is_git(repository):
     if hasattr(repository, 'alias'):
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
     return _type == 'git'
 def is_hg(repository):
     if hasattr(repository, 'alias'):
         _type = repository.alias
     elif hasattr(repository, 'repo_type'):
         _type = repository.repo_type
     else:
         _type = repository
     return _type == 'hg'
 @cache_region('long_term', 'user_attr_or_none')
 def user_attr_or_none(author, show_attr):
     """Try to match email part of VCS committer string with a local user and return show_attr
     - or return None if user not found"""
     email = author_email(author)
     if email:
         from kallithea.model.db import User
         user = User.get_by_email(email, cache=True) # cache will only use sql_cache_short
         if user is not None:
             return getattr(user, show_attr)
     return None
 def email_or_none(author):
     """Try to match email part of VCS committer string with a local user.
     Return primary email of user, email part of the specified author name, or None."""
     if not author:
         return None
     email = user_attr_or_none(author, 'email')
     if email is not None:
         return email # always use user's main email address - not necessarily the one used to find user
     # extract email from the commit string
     email = author_email(author)
     if email:
         return email
     # No valid email, not a valid user in the system, none!
     return None
 def person(author, show_attr="username"):
     """Find the user identified by 'author', return one of the users attributes,
     default to the username attribute, None if there is no user"""
     from kallithea.model.db import User
     # if author is already an instance use it for extraction
     if isinstance(author, User):
         return getattr(author, show_attr)
     value = user_attr_or_none(author, show_attr)
     if value is not None:
         return value
     # Still nothing?  Just pass back the author name if any, else the email
     return author_name(author) or email(author)
 def person_by_id(id_, show_attr="username"):
     from kallithea.model.db import User
     # attr to return from fetched user
     person_getter = lambda usr: getattr(usr, show_attr)
     # maybe it's an ID ?
     if str(id_).isdigit() or isinstance(id_, int):
         id_ = int(id_)
         user = User.get(id_)
         if user is not None:
-            return person_getter(user)
+            return getattr(user, show_attr)
     return id_
 def boolicon(value):
     """Returns boolean value of a value, represented as small html image of true/false
     icons
     :param value: value
     """
     if value:
         return HTML.tag('i', class_="icon-ok")
     else:
         return HTML.tag('i', class_="icon-minus-circled")
 def action_parser(user_log, feed=False, parse_cs=False):
     """
     This helper will action_map the specified string action into translated
     fancy names with icons and links
     :param user_log: user log instance
     :param feed: use output for feeds (no html and fancy icons)
     :param parse_cs: parse Changesets into VCS instances
     """
     action = user_log.action
     action_params = ' '
     x = action.split(':')
     if len(x) > 1:
         action, action_params = x
     def get_cs_links():
         revs_limit = 3  # display this amount always
         revs_top_limit = 50  # show upto this amount of changesets hidden
         revs_ids = action_params.split(',')
         deleted = user_log.repository is None
         if deleted:
             return ','.join(revs_ids)
         repo_name = user_log.repository.repo_name
         def lnk(rev, repo_name):
             lazy_cs = False
             title_ = None
             url_ = '#'
@@ @@ -817,100 +822,97 @@ def action_parser(user_log, feed=False, @@
         'user_updated_repo':           (_('[updated] repository'),
                                         None, 'icon-pencil'),
         'user_downloaded_archive':      (_('[downloaded] archive from repository'),
                                         get_archive_name, 'icon-download-cloud'),
         'admin_deleted_repo':          (_('[delete] repository'),
                                         None, 'icon-trashcan'),
         'admin_created_repo':          (_('[created] repository'),
                                         None, 'icon-plus'),
         'admin_forked_repo':           (_('[forked] repository'),
                                         None, 'icon-fork'),
         'admin_updated_repo':          (_('[updated] repository'),
                                         None, 'icon-pencil'),
         'admin_created_user':          (_('[created] user'),
                                         get_user_name, 'icon-user'),
         'admin_updated_user':          (_('[updated] user'),
                                         get_user_name, 'icon-user'),
         'admin_created_users_group':   (_('[created] user group'),
                                         get_users_group, 'icon-pencil'),
         'admin_updated_users_group':   (_('[updated] user group'),
                                         get_users_group, 'icon-pencil'),
         'user_commented_revision':     (_('[commented] on revision in repository'),
                                         get_cs_links, 'icon-comment'),
         'user_commented_pull_request': (_('[commented] on pull request for'),
                                         get_pull_request, 'icon-comment'),
         'user_closed_pull_request':    (_('[closed] pull request for'),
                                         get_pull_request, 'icon-ok'),
         'push':                        (_('[pushed] into'),
                                         get_cs_links, 'icon-move-up'),
         'push_local':                  (_('[committed via Kallithea] into repository'),
                                         get_cs_links, 'icon-pencil'),
         'push_remote':                 (_('[pulled from remote] into repository'),
                                         get_cs_links, 'icon-move-up'),
         'pull':                        (_('[pulled] from'),
                                         None, 'icon-move-down'),
         'started_following_repo':      (_('[started following] repository'),
                                         None, 'icon-heart'),
         'stopped_following_repo':      (_('[stopped following] repository'),
                                         None, 'icon-heart-empty'),
+    }
     action_str = action_map.get(action, action)
     if feed:
         action = action_str[0].replace('[', '').replace(']', '')
     else:
         action = action_str[0] \
             .replace('[', '<b>') \
             .replace(']', '</b>')
     action_params_func = lambda: ""
     if callable(action_str[1]):
         action_params_func = action_str[1]
     action_params_func = action_str[1] if callable(action_str[1]) else (lambda: "")
     def action_parser_icon():
         action = user_log.action
         action_params = None
         x = action.split(':')
         if len(x) > 1:
             action, action_params = x
         ico = action_map.get(action, ['', '', ''])[2]
         html = """<i class="%s"></i>""" % ico
         return literal(html)
     # returned callbacks we need to call to get
     return [lambda: literal(action), action_params_func, action_parser_icon]
 #==============================================================================
 # GRAVATAR URL
 #==============================================================================
 def gravatar_div(email_address, cls='', size=30, **div_attributes):
     """Return an html literal with a span around a gravatar if they are enabled.
     Extra keyword parameters starting with 'div_' will get the prefix removed
     and '_' changed to '-' and be used as attributes on the div. The default
     class is 'gravatar'.
     """
     from tg import tmpl_context as c
     if not c.visual.use_gravatar:
         return ''
     if 'div_class' not in div_attributes:
         div_attributes['div_class'] = "gravatar"
     attributes = []
     for k, v in sorted(div_attributes.items()):
         assert k.startswith('div_'), k
         attributes.append(' %s="%s"' % (k[4:].replace('_', '-'), escape(v)))
     return literal("""<span%s>%s</span>""" %
                    (''.join(attributes),
                     gravatar(email_address, cls=cls, size=size)))
 def gravatar(email_address, cls='', size=30):
     """return html element of the gravatar
     This method will return an <img> with the resolution double the size (for
     retina screens) of the image. If the url returned from gravatar_url is
     empty then we fallback to using an icon.
     """
@@ @@ -1131,151 +1133,152 @@ def urlify_text(s, repo_name=None, link_ @@
         s = s.rstrip()
     else:
         s = truncatef(s, truncate, whole_word=True)
     s = html_escape(s)
     s = _urlify(s)
     if repo_name is not None:
         s = urlify_issues(s, repo_name)
     if link_ is not None:
         # make href around everything that isn't a href already
         s = linkify_others(s, link_)
     s = s.replace('\r\n', '<br/>').replace('\n', '<br/>')
     # Turn HTML5 into more valid HTML4 as required by some mail readers.
     # (This is not done in one step in html_escape, because character codes like
     # &#123; risk to be seen as an issue reference due to the presence of '#'.)
     s = s.replace("&apos;", "&#39;")
     return literal(s)
 def linkify_others(t, l):
     """Add a default link to html with links.
     HTML doesn't allow nesting of links, so the outer link must be broken up
     in pieces and give space for other links.
     """
     urls = re.compile(r'(\<a.*?\<\/a\>)',)
     links = []
     for e in urls.split(t):
         if e.strip() and not urls.match(e):
             links.append('<a class="message-link" href="%s">%s</a>' % (l, e))
         else:
             links.append(e)
     return ''.join(links)
 # Global variable that will hold the actual urlify_issues function body.
 # Will be set on first use when the global configuration has been read.
 _urlify_issues_f = None
 def urlify_issues(newtext, repo_name):
     """Urlify issue references according to .ini configuration"""
     global _urlify_issues_f
     if _urlify_issues_f is None:
         from kallithea import CONFIG
         from kallithea.model.db import URL_SEP
         assert CONFIG['sqlalchemy.url'] # make sure config has been loaded
         # Build chain of urlify functions, starting with not doing any transformation
         tmp_urlify_issues_f = lambda s: s
         def tmp_urlify_issues_f(s):
             return s
         issue_pat_re = re.compile(r'issue_pat(.*)')
         for k in CONFIG:
             # Find all issue_pat* settings that also have corresponding server_link and prefix configuration
             m = issue_pat_re.match(k)
             if m is None:
                 continue
             suffix = m.group(1)
             issue_pat = CONFIG.get(k)
             issue_server_link = CONFIG.get('issue_server_link%s' % suffix)
             issue_sub = CONFIG.get('issue_sub%s' % suffix)
             if not issue_pat or not issue_server_link or issue_sub is None: # issue_sub can be empty but should be present
                 log.error('skipping incomplete issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_sub)
                 continue
             # Wrap tmp_urlify_issues_f with substitution of this pattern, while making sure all loop variables (and compiled regexpes) are bound
             try:
                 issue_re = re.compile(issue_pat)
             except re.error as e:
                 log.error('skipping invalid issue pattern %r: %r -> %r %r. Error: %s', suffix, issue_pat, issue_server_link, issue_sub, str(e))
                 continue
             log.debug('issue pattern %r: %r -> %r %r', suffix, issue_pat, issue_server_link, issue_sub)
             def issues_replace(match_obj,
                                issue_server_link=issue_server_link, issue_sub=issue_sub):
                 try:
                     issue_url = match_obj.expand(issue_server_link)
                 except (IndexError, re.error) as e:
                     log.error('invalid issue_url setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
                     issue_url = issue_server_link
                 issue_url = issue_url.replace('{repo}', repo_name)
                 issue_url = issue_url.replace('{repo_name}', repo_name.split(URL_SEP)[-1])
                 # if issue_sub is empty use the matched issue reference verbatim
                 if not issue_sub:
                     issue_text = match_obj.group()
                 else:
                     try:
                         issue_text = match_obj.expand(issue_sub)
                     except (IndexError, re.error) as e:
                         log.error('invalid issue_sub setting %r -> %r %r. Error: %s', issue_pat, issue_server_link, issue_sub, str(e))
                         issue_text = match_obj.group()
                 return (
                     '<a class="issue-tracker-link" href="%(url)s">'
                     '%(text)s'
                     '</a>'
                     ) % {
                      'url': issue_url,
                      'text': issue_text,
+                    }
             tmp_urlify_issues_f = (lambda s,
                                           issue_re=issue_re, issues_replace=issues_replace, chain_f=tmp_urlify_issues_f:
                                    issue_re.sub(issues_replace, chain_f(s)))
             def tmp_urlify_issues_f(s, issue_re=issue_re, issues_replace=issues_replace, chain_f=tmp_urlify_issues_f):
                 return issue_re.sub(issues_replace, chain_f(s))
         # Set tmp function globally - atomically
         _urlify_issues_f = tmp_urlify_issues_f
     return _urlify_issues_f(newtext)
 def render_w_mentions(source, repo_name=None):
     """
     Render plain text with revision hashes and issue references urlified
     and with @mention highlighting.
     """
     s = safe_str(source)
     s = urlify_text(s, repo_name=repo_name)
     return literal('<div class="formatted-fixed">%s</div>' % s)
 def short_ref(ref_type, ref_name):
     if ref_type == 'rev':
         return short_id(ref_name)
     return ref_name
 def link_to_ref(repo_name, ref_type, ref_name, rev=None):
     """
     Return full markup for a href to changeset_home for a changeset.
     If ref_type is branch it will link to changelog.
     ref_name is shortened if ref_type is 'rev'.
     if rev is specified show it too, explicitly linking to that revision.
     """
     txt = short_ref(ref_type, ref_name)
     if ref_type == 'branch':
         u = url('changelog_home', repo_name=repo_name, branch=ref_name)
     else:
         u = url('changeset_home', repo_name=repo_name, revision=ref_name)
     l = link_to(repo_name + '#' + txt, u)
     if rev and ref_type != 'rev':
         l = literal('%s (%s)' % (l, link_to(short_id(rev), url('changeset_home', repo_name=repo_name, revision=rev))))
     return l
 def changeset_status(repo, revision):
     from kallithea.model.changeset_status import ChangesetStatusModel
     return ChangesetStatusModel().get_status(repo, revision)
 def changeset_status_lbl(changeset_status):
     from kallithea.model.db import ChangesetStatus

kallithea/lib/rcmail/response.py

➞

Show inline comments

 # The code in this module is entirely lifted from the Lamson project
 # (http://lamsonproject.org/).  Its copyright is:
 # Copyright (c) 2008, Zed A. Shaw
 # All rights reserved.
 # It is provided under this license:
 # Redistribution and use in source and binary forms, with or without
 # modification, are permitted provided that the following conditions are met:
 # * Redistributions of source code must retain the above copyright notice, this
 #   list of conditions and the following disclaimer.
 # * Redistributions in binary form must reproduce the above copyright notice,
 #   this list of conditions and the following disclaimer in the documentation
 #   and/or other materials provided with the distribution.
 # * Neither the name of the Zed A. Shaw nor the names of its contributors may
 #   be used to endorse or promote products derived from this software without
 #   specific prior written permission.
 # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
 # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
 # COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT,
 # INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 # SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
 # STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 # POSSIBILITY OF SUCH DAMAGE.
 import mimetypes
 import os
 import string
 from email import encoders
 from email.charset import Charset
 from email.mime.base import MIMEBase
 from email.utils import parseaddr
 ADDRESS_HEADERS_WHITELIST = ['From', 'To', 'Delivered-To', 'Cc']
 DEFAULT_ENCODING = "utf-8"
 VALUE_IS_EMAIL_ADDRESS = lambda v: '@' in v
 def VALUE_IS_EMAIL_ADDRESS(v):
     return '@' in v
 def normalize_header(header):
     return string.capwords(header.lower(), '-')
 class EncodingError(Exception):
     """Thrown when there is an encoding error."""
     pass
 class MailBase(object):
     """MailBase is used as the basis of lamson.mail and contains the basics of
     encoding an email.  You actually can do all your email processing with this
     class, but it's more raw.
     """
     def __init__(self, items=()):
         self.headers = dict(items)
         self.parts = []
         self.body = None
         self.content_encoding = {'Content-Type': (None, {}),
                                  'Content-Disposition': (None, {}),
                                  'Content-Transfer-Encoding': (None, {})}
     def __getitem__(self, key):
         return self.headers.get(normalize_header(key), None)
     def __len__(self):
         return len(self.headers)
     def __iter__(self):
         return iter(self.headers)
     def __contains__(self, key):
         return normalize_header(key) in self.headers
     def __setitem__(self, key, value):
         self.headers[normalize_header(key)] = value
     def __delitem__(self, key):
         del self.headers[normalize_header(key)]
     def __bool__(self):
         return self.body is not None or len(self.headers) > 0 or len(self.parts) > 0
     def keys(self):
         """Returns the sorted keys."""
         return sorted(self.headers.keys())

kallithea/lib/vcs/backends/git/changeset.py

➞

Show inline comments

@@ @@ -390,98 +390,98 @@ class GitChangeset(BaseChangeset): @@
         path = path.rstrip('/')
         id = self._get_id_for_path(path)
         tree = self.repository._repo[id]
         dirnodes = []
         filenodes = []
         als = self.repository.alias
         for name, stat, id in tree.items():
             obj_path = safe_str(name)
             if path != '':
                 obj_path = '/'.join((path, obj_path))
             if objects.S_ISGITLINK(stat):
                 root_tree = self.repository._repo[self._tree_id]
                 cf = ConfigFile.from_file(BytesIO(self.repository._repo.get_object(root_tree[b'.gitmodules'][1]).data))
                 url = ascii_str(cf.get(('submodule', obj_path), 'url'))
                 dirnodes.append(SubModuleNode(obj_path, url=url, changeset=ascii_str(id),
                                               alias=als))
                 continue
             obj = self.repository._repo.get_object(id)
             if obj_path not in self._stat_modes:
                 self._stat_modes[obj_path] = stat
             if isinstance(obj, objects.Tree):
                 dirnodes.append(DirNode(obj_path, changeset=self))
             elif isinstance(obj, objects.Blob):
                 filenodes.append(FileNode(obj_path, changeset=self, mode=stat))
             else:
                 raise ChangesetError("Requested object should be Tree "
                                      "or Blob, is %r" % type(obj))
         nodes = dirnodes + filenodes
         for node in nodes:
             if node.path not in self.nodes:
                 self.nodes[node.path] = node
         nodes.sort()
         return nodes
     def get_node(self, path):
         """
         Returns ``Node`` object from the given ``path``. If there is no node at
         the given ``path``, ``ChangesetError`` would be raised.
         """
         path = path.rstrip('/')
         if path not in self.nodes:
             try:
                 id_ = self._get_id_for_path(path)
             except ChangesetError:
                 raise NodeDoesNotExistError("Cannot find one of parents' "
                     "directories for a given path: %s" % path)
             _GL = lambda m: m and objects.S_ISGITLINK(m)
             if _GL(self._stat_modes.get(path)):
             stat = self._stat_modes.get(path)
             if stat and objects.S_ISGITLINK(stat):
                 tree = self.repository._repo[self._tree_id]
                 cf = ConfigFile.from_file(BytesIO(self.repository._repo.get_object(tree[b'.gitmodules'][1]).data))
                 url = ascii_str(cf.get(('submodule', path), 'url'))
                 node = SubModuleNode(path, url=url, changeset=ascii_str(id_),
                                      alias=self.repository.alias)
             else:
                 obj = self.repository._repo.get_object(id_)
                 if isinstance(obj, objects.Tree):
                     if path == '':
                         node = RootNode(changeset=self)
                     else:
                         node = DirNode(path, changeset=self)
                     node._tree = obj
                 elif isinstance(obj, objects.Blob):
                     node = FileNode(path, changeset=self)
                     node._blob = obj
                 else:
                     raise NodeDoesNotExistError("There is no file nor directory "
                         "at the given path: '%s' at revision %s"
                         % (path, self.short_id))
             # cache node
             self.nodes[path] = node
         return self.nodes[path]
     @LazyProperty
     def affected_files(self):
         """
         Gets a fast accessible file changes for given changeset
         """
         added, modified, deleted = self._changes_cache
         return list(added.union(modified).union(deleted))
     @LazyProperty
     def _changes_cache(self):
         added = set()
         modified = set()
         deleted = set()
         _r = self.repository._repo
         parents = self.parents
         if not self.parents:
             parents = [EmptyChangeset()]
         for parent in parents:
             if isinstance(parent, EmptyChangeset):
                 oid = None
             else:
                 oid = _r[parent._commit.id].tree

kallithea/lib/vcs/backends/git/repository.py

➞

Show inline comments

@@ @@ -313,117 +313,114 @@ class GitRepository(BaseRepository): @@
             yield {"type": i[0], "extension": i[1], "node": archive_name}
     def _get_url(self, url):
         """
         Returns normalized url. If schema is not given, would fall to
         filesystem (``file:///``) schema.
         """
         if url != 'default' and '://' not in url:
             url = ':///'.join(('file', url))
         return url
     @LazyProperty
     def name(self):
         return os.path.basename(self.path)
     @LazyProperty
     def last_change(self):
         """
         Returns last change made on this repository as datetime object
         """
         return date_fromtimestamp(self._get_mtime(), makedate()[1])
     def _get_mtime(self):
         try:
             return time.mktime(self.get_changeset().date.timetuple())
         except RepositoryError:
             idx_loc = '' if self.bare else '.git'
             # fallback to filesystem
             in_path = os.path.join(self.path, idx_loc, "index")
             he_path = os.path.join(self.path, idx_loc, "HEAD")
             if os.path.exists(in_path):
                 return os.stat(in_path).st_mtime
             else:
                 return os.stat(he_path).st_mtime
     @LazyProperty
     def description(self):
         return safe_str(self._repo.get_description() or b'unknown')
     @LazyProperty
     def contact(self):
         undefined_contact = 'Unknown'
         return undefined_contact
     @property
     def branches(self):
         if not self.revisions:
             return {}
         sortkey = lambda ctx: ctx[0]
         _branches = [(safe_str(key), ascii_str(sha))
                      for key, (sha, type_) in self._parsed_refs.items() if type_ == b'H']
-        return OrderedDict(sorted(_branches, key=sortkey, reverse=False))
+        return OrderedDict(sorted(_branches, key=(lambda ctx: ctx[0]), reverse=False))
     @LazyProperty
     def closed_branches(self):
         return {}
     @LazyProperty
     def tags(self):
         return self._get_tags()
     def _get_tags(self):
         if not self.revisions:
             return {}
         sortkey = lambda ctx: ctx[0]
         _tags = [(safe_str(key), ascii_str(sha))
                  for key, (sha, type_) in self._parsed_refs.items() if type_ == b'T']
-        return OrderedDict(sorted(_tags, key=sortkey, reverse=True))
+        return OrderedDict(sorted(_tags, key=(lambda ctx: ctx[0]), reverse=True))
     def tag(self, name, user, revision=None, message=None, date=None,
             **kwargs):
         """
         Creates and returns a tag for the given ``revision``.
         :param name: name for new tag
         :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"
         :param revision: changeset id for which new tag would be created
         :param message: message of the tag's commit
         :param date: date of tag's commit
         :raises TagAlreadyExistError: if tag with same name already exists
         """
         if name in self.tags:
             raise TagAlreadyExistError("Tag %s already exists" % name)
         changeset = self.get_changeset(revision)
         message = message or "Added tag %s for commit %s" % (name,
             changeset.raw_id)
         self._repo.refs[b"refs/tags/%s" % safe_bytes(name)] = changeset._commit.id
         self._parsed_refs = self._get_parsed_refs()
         self.tags = self._get_tags()
         return changeset
     def remove_tag(self, name, user, message=None, date=None):
         """
         Removes tag with the given ``name``.
         :param name: name of the tag to be removed
         :param user: full username, i.e.: "Joe Doe <joe.doe@example.com>"
         :param message: message of the tag's removal commit
         :param date: date of tag's removal commit
         :raises TagDoesNotExistError: if tag with given name does not exists
         """
         if name not in self.tags:
             raise TagDoesNotExistError("Tag %s does not exist" % name)
         # self._repo.refs is a DiskRefsContainer, and .path gives the full absolute path of '.git'
         tagpath = os.path.join(safe_str(self._repo.refs.path), 'refs', 'tags', name)
         try:
             os.remove(tagpath)
             self._parsed_refs = self._get_parsed_refs()
             self.tags = self._get_tags()
         except OSError as e:
             raise RepositoryError(e.strerror)
     @LazyProperty

kallithea/lib/vcs/conf/settings.py

➞

Show inline comments

 import os
 import tempfile
 from kallithea.lib.vcs.utils import aslist
 from kallithea.lib.vcs.utils.paths import get_user_home
 abspath = lambda * p: os.path.abspath(os.path.join(*p))
 def abspath(*p):
     return os.path.abspath(os.path.join(*p))
 VCSRC_PATH = os.environ.get('VCSRC_PATH')
 if not VCSRC_PATH:
     HOME_ = get_user_home()
     if not HOME_:
         HOME_ = tempfile.gettempdir()
 VCSRC_PATH = VCSRC_PATH or abspath(HOME_, '.vcsrc')
 if os.path.isdir(VCSRC_PATH):
     VCSRC_PATH = os.path.join(VCSRC_PATH, '__init__.py')
 # list of default encoding used in safe_str/safe_bytes methods
 DEFAULT_ENCODINGS = aslist('utf-8')
 # path to git executable run by run_git_command function
 GIT_EXECUTABLE_PATH = 'git'
 # can be also --branches --tags
 GIT_REV_FILTER = '--all'
 BACKENDS = {
     'hg': 'kallithea.lib.vcs.backends.hg.MercurialRepository',
     'git': 'kallithea.lib.vcs.backends.git.GitRepository',
+}
 ARCHIVE_SPECS = {
     'tar': ('application/x-tar', '.tar'),
     'tbz2': ('application/x-bzip2', '.tar.bz2'),
     'tgz': ('application/x-gzip', '.tar.gz'),
     'zip': ('application/zip', '.zip'),
+}

kallithea/lib/vcs/utils/paths.py

➞

Show inline comments

 import os
 abspath = lambda * p: os.path.abspath(os.path.join(*p))
 def abspath(*p):
     return os.path.abspath(os.path.join(*p))
 def get_dirs_for_path(*paths):
     """
     Returns list of directories, including intermediate.
     """
     for path in paths:
         head = path
         while head:
             head, _tail = os.path.split(head)
             if head:
                 yield head
             else:
                 # We don't need to yield empty path
                 break
 def get_dir_size(path):
     root_path = path
     size = 0
     for path, dirs, files in os.walk(root_path):
         for f in files:
             try:
                 size += os.path.getsize(os.path.join(path, f))
             except OSError:
                 pass
     return size
 def get_user_home():
     """
     Returns home path of the user.
     """
     return os.getenv('HOME', os.getenv('USERPROFILE')) or ''

kallithea/model/repo.py

➞

Show inline comments

@@ @@ -64,102 +64,101 @@ class RepoModel(object): @@
         default_perm = 'repository.none' if private else default
         repo_to_perm = UserRepoToPerm()
         repo_to_perm.permission = Permission.get_by_key(default_perm)
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         Session().add(repo_to_perm)
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = Ui.query().filter(Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id, cache=False):
         repo = Repository.query() \
             .filter(Repository.repo_id == repo_id)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_id))
         return repo.scalar()
     def get_repo(self, repository):
         return Repository.guess_instance(repository)
     def get_by_repo_name(self, repo_name, cache=False):
         repo = Repository.query() \
             .filter(Repository.repo_name == repo_name)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_name))
         return repo.scalar()
     def get_all_user_repos(self, user):
         """
         Gets all repositories that user have at least read access
         :param user:
         """
         from kallithea.lib.auth import AuthUser
         user = User.guess_instance(user)
         repos = AuthUser(dbuser=user).permissions['repositories']
         access_check = lambda r: r[1] in ['repository.read',
                                           'repository.write',
                                           'repository.admin']
         repos = [x[0] for x in filter(access_check, repos.items())]
         auth_user = AuthUser(dbuser=User.guess_instance(user))
         repos = [repo_name
             for repo_name, perm in auth_user.permissions['repositories'].items()
             if perm in ['repository.read', 'repository.write', 'repository.admin']
+            ]
         return Repository.query().filter(Repository.repo_name.in_(repos))
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         from tg import tmpl_context as c, request, app_globals
         from tg.i18n import ugettext as _
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         tmpl = template.get_def(tmpl)
         kwargs.update(dict(_=_, h=h, c=c, request=request))
         return tmpl.render_unicode(*args, **kwargs)
     def get_repos_as_dict(self, repos_list, repo_groups_list=None,
                           admin=False,
                           short_name=False):
         """Return repository list for use by DataTable.
         repos_list: list of repositories - but will be filtered for read permission.
         repo_groups_list: added at top of list without permission check.
         admin: return data for action column.
         """
         _render = self._render_datatable
         from tg import tmpl_context as c, request
         from kallithea.model.scm import ScmModel
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=short_name)
         def following(repo_id, is_following):
             return _render('following', repo_id, is_following)
         def last_change(last_change):
             return _render("last_change", last_change)
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
         def desc(desc):

kallithea/model/validators.py

➞

Show inline comments

@@ @@ -141,101 +141,97 @@ def ValidUserGroup(edit=False, old_data= @@
                 msg = self.message('invalid_group', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(users_group_name=msg)
+                )
             # check if group is unique
             old_ugname = None
             if edit:
                 old_id = old_data.get('users_group_id')
                 old_ugname = UserGroup.get(old_id).users_group_name
             if old_ugname != value or not edit:
                 is_existing_group = UserGroup.get_by_group_name(value,
                                                         case_insensitive=True)
                 if is_existing_group:
                     msg = self.message('group_exist', state, usergroup=value)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(users_group_name=msg)
+                    )
             if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+$', value) is None:
                 msg = self.message('invalid_usergroup_name', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(users_group_name=msg)
+                )
     return _validator
 def ValidRepoGroup(edit=False, old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'parent_group_id': _('Cannot assign this group as parent'),
             'group_exists': _('Group "%(group_name)s" already exists'),
             'repo_exists':
                 _('Repository with name "%(group_name)s" already exists')
+        }
         def _validate_python(self, value, state):
             # TODO WRITE VALIDATIONS
             group_name = value.get('group_name')
             parent_group_id = value.get('parent_group_id')
             # slugify repo group just in case :)
             slug = repo_name_slug(group_name)
             # check for parent of self
             parent_of_self = lambda: (
                 old_data['group_id'] == parent_group_id
                 if parent_group_id else False
+            )
             if edit and parent_of_self():
             if edit and parent_group_id and old_data['group_id'] == parent_group_id:
                 msg = self.message('parent_group_id', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(parent_group_id=msg)
+                )
             old_gname = None
             if edit:
                 old_gname = RepoGroup.get(old_data.get('group_id')).group_name
             if old_gname != group_name or not edit:
                 # check group
                 gr = RepoGroup.query() \
                       .filter(func.lower(RepoGroup.group_name) == func.lower(slug)) \
                       .filter(RepoGroup.parent_group_id == parent_group_id) \
                       .scalar()
                 if gr is not None:
                     msg = self.message('group_exists', state, group_name=slug)
                     raise formencode.Invalid(msg, value, state,
                             error_dict=dict(group_name=msg)
+                    )
                 # check for same repo
                 repo = Repository.query() \
                       .filter(func.lower(Repository.repo_name) == func.lower(slug)) \
                       .scalar()
                 if repo is not None:
                     msg = self.message('repo_exists', state, group_name=slug)
                     raise formencode.Invalid(msg, value, state,
                             error_dict=dict(group_name=msg)
+                    )
     return _validator
 def ValidPassword():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_password':
                 _('Invalid characters (non-ascii) in password')
+        }
         def _validate_python(self, value, state):
             try:
                 (value or '').encode('ascii')
             except UnicodeError:
                 msg = self.message('invalid_password', state)
                 raise formencode.Invalid(msg, value, state,)

setup.py

➞

Show inline comments

 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 import os
 import platform
 import sys
 import setuptools
 # monkey patch setuptools to use distutils owner/group functionality
 from setuptools.command import sdist
 if sys.version_info < (3, 6):
     raise Exception('Kallithea requires Python 3.6 or later')
 here = os.path.abspath(os.path.dirname(__file__))
 def _get_meta_var(name, data, callback_handler=None):
     import re
     matches = re.compile(r'(?:%s)\s*=\s*(.*)' % name).search(data)
     if matches:
         if not callable(callback_handler):
             callback_handler = lambda v: v
         return callback_handler(eval(matches.groups()[0]))
         s = eval(matches.groups()[0])
         if callable(callback_handler):
             return callback_handler(s)
         return s
 _meta = open(os.path.join(here, 'kallithea', '__init__.py'), 'r')
 _metadata = _meta.read()
 _meta.close()
 callback = lambda V: ('.'.join(map(str, V[:3])) + '.'.join(V[3:]))
 def callback(V):
     return '.'.join(map(str, V[:3])) + '.'.join(V[3:])
 __version__ = _get_meta_var('VERSION', _metadata, callback)
 __license__ = _get_meta_var('__license__', _metadata)
 __author__ = _get_meta_var('__author__', _metadata)
 __url__ = _get_meta_var('__url__', _metadata)
 # defines current platform
 __platform__ = platform.system()
 is_windows = __platform__ in ['Windows']
 requirements = [
     "alembic >= 1.0.10, < 1.5",
     "gearbox >= 0.1.0, < 1",
     "waitress >= 0.8.8, < 1.5",
     "WebOb >= 1.8, < 1.9",
     "backlash >= 0.1.2, < 1",
     "TurboGears2 >= 2.4, < 2.5",
     "tgext.routes >= 0.2.0, < 1",
     "Beaker >= 1.10.1, < 2",
     "WebHelpers2 >= 2.0, < 2.1",
     "FormEncode >= 1.3.1, < 1.4",
     "SQLAlchemy >= 1.2.9, < 1.4",
     "Mako >= 0.9.1, < 1.2",
     "Pygments >= 2.2.0, < 2.6",
     "Whoosh >= 2.7.1, < 2.8",
     "celery >= 4.3, < 4.5",
     "Babel >= 1.3, < 2.9",
     "python-dateutil >= 2.1.0, < 2.9",
     "Markdown >= 2.2.1, < 3.2",
     "docutils >= 0.11, < 0.17",
     "URLObject >= 2.3.4, < 2.5",
     "Routes >= 2.0, < 2.5",
     "dulwich >= 0.19.0, < 0.20",
     "mercurial >= 5.2, < 5.4",
     "decorator >= 4.2.1, < 4.5",
     "Paste >= 2.0.3, < 3.4",
     "bleach >= 3.0, < 3.2",
     "Click >= 7.0, < 8",
     "ipaddr >= 2.2.0, < 2.3",
     "paginate >= 0.5, < 0.6",
     "paginate_sqlalchemy >= 0.3.0, < 0.4",
+]
 if not is_windows:
     requirements.append("bcrypt >= 3.1.0, < 3.2")
 dependency_links = [
+]

0 comments (0 inline, 0 general)