The :mod:`models` module

========================

.. automodule:: kallithea.model

   :members:

.. automodule:: kallithea.model.comment

   :members:

.. automodule:: kallithea.model.permission

   :members:

.. automodule:: kallithea.model.repo

   :members:

.. automodule:: kallithea.model.repo_group

   :members:

.. automodule:: kallithea.model.scm

   :members:

.. automodule:: kallithea.model.user

   :members:

    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)

        )

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

        )

        for user_gr in group_iter:

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

        user_actions = lambda user_id, username: (

                template.get_def("user_actions")

                .render_unicode(user_id, username, _=_, h=h, c=c))

        if (pull_request.owner_id == request.authuser.user_id or

            h.HasPermissionAny('hg.admin')() or

            h.HasRepoPermissionLevel('admin')(pull_request.org_repo.repo_name) or

            h.HasRepoPermissionLevel('admin')(pull_request.other_repo.repo_name)

        ) and not pull_request.is_closed():

            PullRequestModel().delete(pull_request)

            Session().commit()

            h.flash(_('Successfully deleted pull request %s') % pull_request_id,

                    category='success')

            return {

               'location': h.url('my_pullrequests'), # or repo pr list?

            }

        raise HTTPForbidden()

    text = request.POST.get('text', '').strip()

    comment = ChangesetCommentsModel().create(

        text=text,

        repo=c.db_repo.repo_id,

        author=request.authuser.user_id,

        revision=revision,

        pull_request=pull_request_id,

        f_path=f_path or None,

        line_no=line_no or None,

from kallithea.model.user import UserModel

log = logging.getLogger(__name__)

class DbManage(object):

    def __init__(self, dbconf, root, tests=False, SESSION=None, cli_args=None):

        self.dbname = dbconf.split('/')[-1]

        self.tests = tests

        self.root = root

        self.dburi = dbconf

        self.cli_args = cli_args or {}

        self.init_db(SESSION=SESSION)

    def _ask_ok(self, msg):

        """Invoke ask_ok unless the force_ask option provides the answer"""

        force_ask = self.cli_args.get('force_ask')

        if force_ask is not None:

            return force_ask

        from kallithea.lib.utils2 import ask_ok

        return ask_ok(msg)

    def init_db(self, SESSION=None):

class NonRelativePathError(Exception):

    pass

class IMCCommitError(Exception):

    pass

class UserCreationError(Exception):

    pass

class HgsubversionImportError(Exception):

    pass

from kallithea.lib.rcmail.exceptions import BadHeaders, InvalidMessage

from kallithea.lib.rcmail.response import MailResponse

class Message(object):

    """

    Encapsulates an email message.

    :param subject: email subject header

    :param recipients: list of email addresses

    :param body: plain text message

    :param html: HTML message

    :param sender: email sender address

    :param cc: CC list

    :param bcc: BCC list

    :param extra_headers: dict of extra email headers

            encoded = content.encode('ascii')

            charset = 'ascii'

        except UnicodeError:

            encoded = content.encode('utf-8')

            charset = 'utf-8'

        self.set_payload(encoded, charset=charset)

    def extract_payload(self, mail):

        if mail.body is None:

            return  # only None, '' is still ok

        cdisp, cdisp_params = mail.content_encoding['Content-Disposition']

        assert ctype, ("Extract payload requires that mail.content_encoding "

                       "have a valid Content-Type.")

        if ctype.startswith("text/"):

            self.add_text(mail.body)

        else:

            if cdisp:

                # replicate the content-disposition settings

                self.add_header('Content-Disposition', cdisp, **cdisp_params)

        Commit api function that will remove given ``FileNode`` into this

        repository.

        :raises ``EmptyRepositoryError``: if there are no changesets yet

        :raises ``NodeDoesNotExistError``: if there is no file with given path

        """

        raise NotImplementedError

    def commit(self, message, **kwargs):

        """

        Persists current changes made on this repository and returns newly

        created changeset.

        """

        raise NotImplementedError

    def get_state(self):

        """

        Returns dictionary with ``added``, ``changed`` and ``removed`` lists

        containing ``FileNode`` objects.

        """

        raise NotImplementedError

    def get_config_value(self, section, name, config_file=None):

        """

class EmptyRepositoryError(RepositoryError):

    pass

class TagAlreadyExistError(RepositoryError):

    pass

class TagDoesNotExistError(RepositoryError):

    pass

class BranchDoesNotExistError(RepositoryError):

    pass

class ChangesetError(RepositoryError):

    pass

class ChangesetDoesNotExistError(ChangesetError):

    pass

class CommitError(RepositoryError):

    pass

class NodeError(VCSError):

    pass

class RemovedFileNodeError(NodeError):

    pass

class NodeAlreadyExistsError(CommitError):

    pass

class NodeAlreadyAddedError(CommitError):

    pass

class NodeAlreadyRemovedError(CommitError):

    pass

class ImproperArchiveTypeError(VCSError):

    pass

    def data_added_event(self):

        return self.worker.data_added

    @property

    def data_added(self):

        return self.worker.data_added.is_set()

    @property

    def reading_paused(self):

        return not self.worker.keep_reading.is_set()

    @property

    def done_reading(self):

        """

        Done_reading does not mean that the iterator's buffer is empty.

        Iterator might have done reading from underlying source, but the read

        chunks might still be available for serving through .next() method.

        :returns: An Bool value.

        """

        return self.worker.EOF.is_set()

    @property

    def length(self):

class _Missing(object):

    def __repr__(self):

        return 'no value'

    def __reduce__(self):

        return '_missing'

_missing = _Missing()

        self.__module__ = func.__module__

        self.__name__ = func.__name__

        self.__doc__ = func.__doc__

    def __get__(self, obj, klass=None):

        if obj is None:

            return self

        value = obj.__dict__.get(self.__name__, _missing)

        if value is _missing:

            value = self._func(obj)

            obj.__dict__[self.__name__] = value

        return value

abspath = lambda * p: 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:

            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:

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import datetime

import logging

from tg import app_globals

from tg import tmpl_context as c

from tg.i18n import ugettext as _

from kallithea.lib import helpers as h

from kallithea.model.db import User

log = logging.getLogger(__name__)

class NotificationModel(object):

    TYPE_CHANGESET_COMMENT = 'cs_comment'

    TYPE_MESSAGE = 'message'

    TYPE_MENTION = 'mention' # not used

    TYPE_CHANGESET_COMMENT = NotificationModel.TYPE_CHANGESET_COMMENT

    TYPE_MESSAGE = NotificationModel.TYPE_MESSAGE # only used for testing

    # NotificationModel.TYPE_MENTION is not used

    TYPE_PASSWORD_RESET = 'password_link'

    TYPE_REGISTRATION = NotificationModel.TYPE_REGISTRATION

    TYPE_PULL_REQUEST = NotificationModel.TYPE_PULL_REQUEST

    TYPE_PULL_REQUEST_COMMENT = NotificationModel.TYPE_PULL_REQUEST_COMMENT

    TYPE_DEFAULT = 'default'

    def __init__(self):

        super(EmailNotificationModel, self).__init__()

        self._tmpl_lookup = app_globals.mako_lookup

        self.email_types = {

            self.TYPE_CHANGESET_COMMENT: 'changeset_comment',

            self.TYPE_PASSWORD_RESET: 'password_reset',

            self.TYPE_REGISTRATION: 'registration',

            self.TYPE_DEFAULT: 'default',

            self.TYPE_PULL_REQUEST: 'pull_request',

            self.TYPE_PULL_REQUEST_COMMENT: 'pull_request_comment',

        }

        self._subj_map = {

            self.TYPE_CHANGESET_COMMENT: _('[Comment] %(repo_name)s changeset %(short_id)s "%(message_short)s" on %(branch)s'),

            self.TYPE_MESSAGE: 'Test Message',

class SshKeyModel(object):

    def create(self, user, description, public_key):

        """

        :param user: user or user_id

        :param description: description of SshKey

        :param publickey: public key text

        Will raise SshKeyModelException on errors

        """

        try:

        except ssh.SshKeyParseError as e:

            raise SshKeyModelException(_('SSH key %r is invalid: %s') % (public_key, e.args[0]))

        if not description.strip():

            description = comment.strip()

        user = User.guess_instance(user)

        new_ssh_key = UserSshKeys()

        new_ssh_key.user_id = user.user_id

        new_ssh_key.description = description

        new_ssh_key.public_key = public_key

                            # set none for default when updating to

                            # private repo protects against form manipulation

                            v = EMPTY_PERM

                    perms_update.add((member, v, t))

            value['perms_updates'] = list(perms_update)

            value['perms_new'] = list(perms_new)

            # update permissions

            for k, v, t in perms_new:

                try:

                    if t == 'user':

                            .filter(User.active == True) \

                            .filter(User.username == k).one()

                    if t == 'users_group':

                            .filter(UserGroup.users_group_active == True) \

                            .filter(UserGroup.users_group_name == k).one()

                except Exception:

                    log.exception('Updated permission failed')

                    msg = self.message('perm_new_member_type', state)

                    raise formencode.Invalid(msg, value, state,

                        error_dict=dict(perm_new_member_name=msg)

                    )

            return value

    return _validator