# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# 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 General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.lib.celerylib

~~~~~~~~~~~~~~~~~~~~~~~

celery libs for Kallithea

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: Nov 27, 2010

:author: marcink

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

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

"""

import logging

import os

from hashlib import md5

from decorator import decorator

from tg import config

from kallithea import CELERY_EAGER, CELERY_ON

from kallithea.lib.pidlock import DaemonLock, LockHeld

from kallithea.lib.utils2 import safe_str

from kallithea.model import meta

log = logging.getLogger(__name__)

class FakeTask(object):

    """Fake a sync result to make it look like a finished task"""

    def __init__(self, result):

        self.result = result

    def failed(self):

        return False

    traceback = None # if failed

    task_id = None

def task(f_org):

    """Wrapper of celery.task.task, running async if CELERY_ON

    """

    if CELERY_ON:

        def f_async(*args, **kwargs):

            log.info('executing %s task', f_org.__name__)

            try:

                f_org(*args, **kwargs)

            finally:

                log.info('executed %s task', f_org.__name__)

        f_async.__name__ = f_org.__name__

        from kallithea.lib import celerypylons

        runner = celerypylons.task(ignore_result=True)(f_async)

        def f_wrapped(*args, **kwargs):

            t = runner.apply_async(args=args, kwargs=kwargs)

            log.info('executing task %s in async mode - id %s', f_org, t.task_id)

            return t

    else:

        def f_wrapped(*args, **kwargs):

            log.info('executing task %s in sync', f_org.__name__)

            try:

                result = f_org(*args, **kwargs)

            except Exception as e:

                log.error('exception executing sync task %s in sync: %r', f_org.__name__, e)

                raise # TODO: return this in FakeTask as with async tasks?

            return FakeTask(result)

    return f_wrapped

def __get_lockkey(func, *fargs, **fkwargs):

    params = list(fargs)

    params.extend(['%s-%s' % ar for ar in fkwargs.items()])

    func_name = str(func.__name__) if hasattr(func, '__name__') else str(func)

    lockkey = 'task_%s.lock' % \

    return lockkey

def locked_task(func):

    def __wrapper(func, *fargs, **fkwargs):

        lockkey = __get_lockkey(func, *fargs, **fkwargs)

        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:

            l = DaemonLock(os.path.join(lockkey_path, lockkey))

            ret = func(*fargs, **fkwargs)

            l.release()

            return ret

        except LockHeld:

            log.info('LockHeld')

            return 'Task with key %s already running' % lockkey

    return decorator(__wrapper, func)

def get_session():

    sa = meta.Session()

    return sa

def dbsession(func):

    def __wrapper(func, *fargs, **fkwargs):

        try:

            ret = func(*fargs, **fkwargs)

            return ret

        finally:

            if CELERY_ON and not CELERY_EAGER:

                meta.Session.remove()

    return decorator(__wrapper, func)

            if len(revs_ids) > revs_top_limit:

                morelinks += ', ...'

            cs_links.append(html_tmpl % (uniq_id, morelinks))

        if len(revs) > 1:

            cs_links.append(compare_view)

        return ''.join(cs_links)

    def get_fork_name():

        repo_name = action_params

        url_ = url('summary_home', repo_name=repo_name)

        return _('Fork name %s') % link_to(action_params, url_)

    def get_user_name():

        user_name = action_params

        return user_name

    def get_users_group():

        group_name = action_params

        return group_name

    def get_pull_request():

        from kallithea.model.db import PullRequest

        pull_request_id = action_params

        nice_id = PullRequest.make_nice_id(pull_request_id)

        deleted = user_log.repository is None

        if deleted:

            repo_name = user_log.repository_name

        else:

            repo_name = user_log.repository.repo_name

        return link_to(_('Pull request %s') % nice_id,

                    url('pullrequest_show', repo_name=repo_name,

                    pull_request_id=pull_request_id))

    def get_archive_name():

        archive_name = action_params

        return archive_name

    # action : translated str, callback(extractor), icon

    action_map = {

        'user_deleted_repo':           (_('[deleted] repository'),

                                        None, 'icon-trashcan'),

        'user_created_repo':           (_('[created] repository'),

                                        None, 'icon-plus'),

        'user_created_fork':           (_('[created] repository as fork'),

                                        None, 'icon-fork'),

        'user_forked_repo':            (_('[forked] repository'),

                                        get_fork_name, 'icon-fork'),

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

    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.

    """

    from tg import tmpl_context as c

    if not c.visual.use_gravatar:

        return ''

    src = gravatar_url(email_address, size * 2)

    if src:

        # here it makes sense to use style="width: ..." (instead of, say, a

        # stylesheet) because we using this to generate a high-res (retina) size

        html = ('<i class="icon-gravatar {cls}"'

                ' style="font-size: {size}px;background-size: {size}px;background-image: url(\'{src}\')"'

                '></i>').format(cls=cls, size=size, src=src)

    else:

        # if src is empty then there was no gravatar, so we use a font icon

        html = ("""<i class="icon-user {cls}" style="font-size: {size}px;"></i>"""

            .format(cls=cls, size=size, src=src))

    return literal(html)

def gravatar_url(email_address, size=30, default=''):

    # doh, we need to re-import those to mock it later

    from kallithea.config.routing import url

    from kallithea.model.db import User

    from tg import tmpl_context as c

    if not c.visual.use_gravatar:

        return ""

    _def = 'anonymous@kallithea-scm.org'  # default gravatar

    email_address = email_address or _def

    if email_address == _def:

        return default

    parsed_url = urlparse.urlparse(url.current(qualified=True))

    url = (c.visual.gravatar_url or User.DEFAULT_GRAVATAR_URL) \

               .replace('{email}', email_address) \

               .replace('{md5email}', hashlib.md5(safe_str(email_address).lower()).hexdigest()) \

               .replace('{netloc}', parsed_url.netloc) \

               .replace('{scheme}', parsed_url.scheme) \

    return url

def changed_tooltip(nodes):

    """

    Generates a html string for changed nodes in changeset page.

    It limits the output to 30 entries

    :param nodes: LazyNodesGenerator

    """

    if nodes:

        pref = ': <br/> '

        suf = ''

        if len(nodes) > 30:

            suf = '<br/>' + _(' and %s more') % (len(nodes) - 30)

        return literal(pref + '<br/> '.join([safe_unicode(x.path)

                                             for x in nodes[:30]]) + suf)

    else:

        return ': ' + _('No files')

def fancy_file_stats(stats):

    """

    Displays a fancy two colored bar for number of added/deleted

    lines of code on file

    :param stats: two element list of added/deleted lines of code

    """

    from kallithea.lib.diffs import NEW_FILENODE, DEL_FILENODE, \

        MOD_FILENODE, RENAMED_FILENODE, CHMOD_FILENODE, BIN_FILENODE

    a, d = stats['added'], stats['deleted']

    width = 100

    if stats['binary']:

        # binary mode

        lbl = ''

        bin_op = 1

        if BIN_FILENODE in stats['ops']:

            lbl = 'bin+'

        if NEW_FILENODE in stats['ops']:

            lbl += _('new file')

            bin_op = NEW_FILENODE

        elif MOD_FILENODE in stats['ops']:

            lbl += _('mod')

            bin_op = MOD_FILENODE

        elif DEL_FILENODE in stats['ops']:

            lbl += _('del')

            bin_op = DEL_FILENODE

        elif RENAMED_FILENODE in stats['ops']:

            lbl += _('rename')

            bin_op = RENAMED_FILENODE

        # chmod can go with other operations

        if CHMOD_FILENODE in stats['ops']:

            _org_lbl = _('chmod')

            lbl += _org_lbl if lbl.endswith('+') else '+%s' % _org_lbl

        #import ipdb;ipdb.set_trace()

        b_d = '<div class="bin bin%s progress-bar" style="width:100%%">%s</div>' % (bin_op, lbl)

        b_a = '<div class="bin bin1" style="width:0%"></div>'

        return literal('<div style="width:%spx" class="progress">%s%s</div>' % (width, b_a, b_d))

    t = stats['added'] + stats['deleted']

    unit = float(width) / (t or 1)

    # needs > 9% of width to be visible or 0 to be hidden

    a_p = max(9, unit * a) if a > 0 else 0

    d_p = max(9, unit * d) if d > 0 else 0

    p_sum = a_p + d_p

    if p_sum > width:

        # adjust the percentage to be == 100% since we adjusted to 9

        if a_p > d_p:

            a_p = a_p - (p_sum - width)

        else:

            d_p = d_p - (p_sum - width)

    a_v = a if a > 0 else ''

    d_v = d if d > 0 else ''

    d_a = '<div class="added progress-bar" style="width:%s%%">%s</div>' % (

        a_p, a_v

    )

    d_d = '<div class="deleted progress-bar" style="width:%s%%">%s</div>' % (

        d_p, d_v

    )

    return literal('<div class="progress" style="width:%spx">%s%s</div>' % (width, d_a, d_d))

_URLIFY_RE = re.compile(r'''

# URL markup

(?P<url>%s) |

# @mention markup

(?P<mention>%s) |

# Changeset hash markup

(?<!\w|[-_])

  (?P<hash>[0-9a-f]{12,40})

(?!\w|[-_]) |

# Markup of *bold text*

(?:

  (?:^|(?<=\s))

  (?P<bold> [*] (?!\s) [^*\n]* (?<!\s) [*] )

  (?![*\w])

) |

# "Stylize" markup

\[see\ \=&gt;\ *(?P<seen>[a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\] |

\[license\ \=&gt;\ *(?P<license>[a-zA-Z0-9\/\=\?\&\ \:\/\.\-]*)\] |

\[(?P<tagtype>requires|recommends|conflicts|base)\ \=&gt;\ *(?P<tagvalue>[a-zA-Z0-9\-\/]*)\] |

\[(?:lang|language)\ \=&gt;\ *(?P<lang>[a-zA-Z\-\/\#\+]*)\] |

\[(?P<tag>[a-z]+)\]

''' % (url_re.pattern, MENTIONS_REGEX.pattern),

    re.VERBOSE | re.MULTILINE | re.IGNORECASE)

def urlify_text(s, repo_name=None, link_=None, truncate=None, stylize=False, truncatef=truncate):

    """

    Parses given text message and make literal html with markup.

    The text will be truncated to the specified length.

    Hashes are turned into changeset links to specified repository.

    URLs links to what they say.

    Issues are linked to given issue-server.

    If link_ is provided, all text not already linking somewhere will link there.

    """

    def _replace(match_obj):

        url = match_obj.group('url')

        if url is not None:

            return '<a href="%(url)s">%(url)s</a>' % {'url': url}

        mention = match_obj.group('mention')

        if mention is not None:

            return '<b>%s</b>' % mention

        hash_ = match_obj.group('hash')

        if hash_ is not None and repo_name is not None:

            from kallithea.config.routing import url  # doh, we need to re-import url to mock it later

            return '<a class="changeset_hash" href="%(url)s">%(hash)s</a>' % {

                 'url': url('changeset_home', repo_name=repo_name, revision=hash_),

                 'hash': hash_,

                }

        bold = match_obj.group('bold')

        if bold is not None:

            return '<b>*%s*</b>' % _urlify(bold[1:-1])

        if stylize:

            seen = match_obj.group('seen')

            if seen:

                return '<div class="label label-meta" data-tag="see">see =&gt; %s</div>' % seen

            license = match_obj.group('license')

            if license:

                return '<div class="label label-meta" data-tag="license"><a href="http://www.opensource.org/licenses/%s">%s</a></div>' % (license, license)

            tagtype = match_obj.group('tagtype')

            if tagtype:

                tagvalue = match_obj.group('tagvalue')

                return '<div class="label label-meta" data-tag="%s">%s =&gt; <a href="/%s">%s</a></div>' % (tagtype, tagtype, tagvalue, tagvalue)

            lang = match_obj.group('lang')

            if lang:

                return '<div class="label label-meta" data-tag="lang">%s</div>' % lang

            tag = match_obj.group('tag')

            if tag:

                return '<div class="label label-meta" data-tag="%s">%s</div>' % (tag, tag)

        return match_obj.group(0)

    def _urlify(s):

        """

        Extract urls from text and make html links out of them

        """

        return _URLIFY_RE.sub(_replace, s)

    if truncate is None:

        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

    # { risk to be seen as an issue reference due to the presence of '#'.)

    s = s.replace("'", "'")

    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.

# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# 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 General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

"""

kallithea.model.db

~~~~~~~~~~~~~~~~~~

Database Models for Kallithea

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: Apr 08, 2010

:author: marcink

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

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

"""

import base64

import collections

import datetime

import functools

import hashlib

import logging

import os

import time

import traceback

import ipaddr

import sqlalchemy

from beaker.cache import cache_region, region_invalidate

from sqlalchemy import *

from sqlalchemy.ext.hybrid import hybrid_property

from sqlalchemy.orm import class_mapper, joinedload, relationship, validates

from tg.i18n import lazy_ugettext as _

from webob.exc import HTTPNotFound

import kallithea

from kallithea.lib.caching_query import FromCache

from kallithea.lib.compat import json

from kallithea.lib.exceptions import DefaultUserException

from kallithea.lib.utils2 import Optional, aslist, get_changeset_safe, get_clone_url, remove_prefix, safe_int, safe_str, safe_unicode, str2bool, urlreadable

from kallithea.lib.vcs import get_backend

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

from kallithea.lib.vcs.utils.helpers import get_scm

from kallithea.lib.vcs.utils.lazy import LazyProperty

from kallithea.model.meta import Base, Session

URL_SEP = '/'

log = logging.getLogger(__name__)

#==============================================================================

# BASE CLASSES

#==============================================================================

_hash_key = lambda k: hashlib.md5(safe_str(k)).hexdigest()

class BaseDbModel(object):

    """

    Base Model for all classes

    """

    @classmethod

    def _get_keys(cls):

        """return column names for this model """

        return class_mapper(cls).c.keys()

    def get_dict(self):

        """

        return dict with keys and values corresponding

        to this model data """

        d = {}

        for k in self._get_keys():

            d[k] = getattr(self, k)

        # also use __json__() if present to get additional fields

        _json_attr = getattr(self, '__json__', None)

        if _json_attr:

            # update with attributes from __json__

            if callable(_json_attr):

                _json_attr = _json_attr()

            for k, val in _json_attr.iteritems():

                d[k] = val

        return d

    def get_appstruct(self):

        """return list with keys and values tuples corresponding

        to this model data """

        return [

            (k, getattr(self, k))

            for k in self._get_keys()

        ]

    def populate_obj(self, populate_dict):

        """populate model with data from given populate_dict"""

        for k in self._get_keys():

            if k in populate_dict:

                setattr(self, k, populate_dict[k])

    @classmethod

    def query(cls):

        return Session().query(cls)

    @classmethod

    def get(cls, id_):

        if id_:

            return cls.query().get(id_)

    @classmethod

    def guess_instance(cls, value, callback=None):

        """Haphazardly attempt to convert `value` to a `cls` instance.

        If `value` is None or already a `cls` instance, return it. If `value`

        is a number (or looks like one if you squint just right), assume it's

        a database primary key and let SQLAlchemy sort things out. Otherwise,

        fall back to resolving it using `callback` (if specified); this could

        e.g. be a function that looks up instances by name (though that won't

        work if the name begins with a digit). Otherwise, raise Exception.

        """

        if value is None:

            return None

        if isinstance(value, cls):

            return value

            return cls.get(value)

        if callback is not None:

            return callback(value)

        raise Exception(

            'given object must be int, long or Instance of %s '

            'got %s, no callback provided' % (cls, type(value))

        )

    @classmethod

    def get_or_404(cls, id_):

        try:

            id_ = int(id_)

        except (TypeError, ValueError):

            raise HTTPNotFound

        res = cls.query().get(id_)

        if res is None:

            raise HTTPNotFound

        return res

    @classmethod

    def delete(cls, id_):

        obj = cls.query().get(id_)

        Session().delete(obj)

    def __repr__(self):

        if hasattr(self, '__unicode__'):

            # python repr needs to return str

            try:

                return safe_str(self.__unicode__())

            except UnicodeDecodeError:

                pass

        return '<DB:%s>' % (self.__class__.__name__)

_table_args_default_dict = {'extend_existing': True,

                            'mysql_engine': 'InnoDB',

                            'mysql_charset': 'utf8',

                            'sqlite_autoincrement': True,

                           }

class Setting(Base, BaseDbModel):

    __tablename__ = 'settings'

    __table_args__ = (

        _table_args_default_dict,

    )

    SETTINGS_TYPES = {

        'str': safe_str,

        'int': safe_int,

        'unicode': safe_unicode,

        'bool': str2bool,

        'list': functools.partial(aslist, sep=',')

    }

    DEFAULT_UPDATE_URL = ''

    app_settings_id = Column(Integer(), primary_key=True)

    app_settings_name = Column(String(255), nullable=False, unique=True)

    _app_settings_value = Column("app_settings_value", Unicode(4096), nullable=False)

    _app_settings_type = Column("app_settings_type", String(255), nullable=True) # FIXME: not nullable?

    def __init__(self, key='', val='', type='unicode'):

        self.app_settings_name = key

        self.app_settings_value = val

        self.app_settings_type = type

    @validates('_app_settings_value')

    def validate_settings_value(self, key, val):

        assert isinstance(val, unicode)

        return val

    @hybrid_property

    def app_settings_value(self):

        v = self._app_settings_value

        _type = self.app_settings_type

        converter = self.SETTINGS_TYPES.get(_type) or self.SETTINGS_TYPES['unicode']

        return converter(v)

    @app_settings_value.setter

    def app_settings_value(self, val):

        """

        Setter that will always make sure we use unicode in app_settings_value

        :param val:

        """

        self._app_settings_value = safe_unicode(val)

    @hybrid_property

    def app_settings_type(self):

        return self._app_settings_type

    @app_settings_type.setter

    def app_settings_type(self, val):

        if val not in self.SETTINGS_TYPES:

            raise Exception('type must be one of %s got %s'

                            % (list(self.SETTINGS_TYPES), val))

        self._app_settings_type = val

    def __unicode__(self):

        return u"<%s('%s:%s[%s]')>" % (

            self.__class__.__name__,

            self.app_settings_name, self.app_settings_value, self.app_settings_type

        )

    @classmethod

    def get_by_name(cls, key):

        return cls.query() \

            .filter(cls.app_settings_name == key).scalar()

    @classmethod

    def get_by_name_or_create(cls, key, val='', type='unicode'):

        res = cls.get_by_name(key)

        if res is None:

            res = cls(key, val, type)

        return res

    @classmethod

    def create_or_update(cls, key, val=Optional(''), type=Optional('unicode')):

        """

        Creates or updates Kallithea setting. If updates are triggered, it will only

        update parameters that are explicitly set. Optional instance will be skipped.

        :param key:

        :param val:

        :param type:

        :return:

        """

        res = cls.get_by_name(key)

        if res is None:

            val = Optional.extract(val)

            type = Optional.extract(type)

            res = cls(key, val, type)

            Session().add(res)

        else:

            res.app_settings_name = key

            if not isinstance(val, Optional):

                # update if set

                res.app_settings_value = val

            if not isinstance(type, Optional):

                # update if set

                res.app_settings_type = type

        return res

    @classmethod

    def get_app_settings(cls, cache=False):

        ret = cls.query()

        if cache:

            ret = ret.options(FromCache("sql_cache_short", "get_hg_settings"))

        if ret is None:

            raise Exception('Could not get application settings !')

        settings = {}

        for each in ret:

            settings[each.app_settings_name] = \

                each.app_settings_value

        return settings

    @classmethod

    def get_auth_settings(cls, cache=False):

        ret = cls.query() \

                .filter(cls.app_settings_name.startswith('auth_')).all()

        fd = {}

        for row in ret:

            fd[row.app_settings_name] = row.app_settings_value

        return fd

    @classmethod

    def get_default_repo_settings(cls, cache=False, strip_prefix=False):

        ret = cls.query() \

                .filter(cls.app_settings_name.startswith('default_')).all()

        fd = {}

        for row in ret:

            key = row.app_settings_name

            if strip_prefix:

                key = remove_prefix(key, prefix='default_')

            fd.update({key: row.app_settings_value})

        return fd

    @classmethod

    def get_server_info(cls):

        import pkg_resources

        import platform

        from kallithea.lib.utils import check_git_version

        mods = [(p.project_name, p.version) for p in pkg_resources.working_set]

        info = {

            'modules': sorted(mods, key=lambda k: k[0].lower()),

            'py_version': platform.python_version(),

            'platform': safe_unicode(platform.platform()),

# -*- coding: utf-8 -*-

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,