# -*- coding: utf-8 -*-
"""
rhodecode.lib.utils
~~~~~~~~~~~~~~~~~~~
Utilities library for RhodeCode
:created_on: Apr 18, 2010
:author: marcink
:copyright: (C) 2009-2011 Marcin Kuzminski <marcin@python-works.com>
:license: GPLv3, see COPYING for more details.
# 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; version 2
# of the License or (at your opinion) any later version of the license.
#
# 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, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
# MA 02110-1301, USA.
import os
import logging
import datetime
import traceback
import paste
import beaker
from paste.script.command import Command, BadCommand
from UserDict import DictMixin
from mercurial import ui, config, hg
from mercurial.error import RepoError
from webhelpers.text import collapse, remove_formatting, strip_tags
from vcs.backends.base import BaseChangeset
from vcs.utils.lazy import LazyProperty
from rhodecode.model import meta
from rhodecode.model.caching_query import FromCache
from rhodecode.model.db import Repository, User, RhodeCodeUi, UserLog, Group
from rhodecode.model.repo import RepoModel
from rhodecode.model.user import UserModel
log = logging.getLogger(__name__)
def recursive_replace(str, replace=' '):
"""Recursive replace of given sign to just one instance
:param str: given string
:param replace: char to find and replace multiple instances
Examples::
>>> recursive_replace("Mighty---Mighty-Bo--sstones",'-')
'Mighty-Mighty-Bo-sstones'
if str.find(replace * 2) == -1:
return str
else:
str = str.replace(replace * 2, replace)
return recursive_replace(str, replace)
def repo_name_slug(value):
"""Return slug of name of repository
This function is called on each creation/modification
of repository to prevent bad names in repo
slug = remove_formatting(value)
slug = strip_tags(slug)
for c in """=[]\;'"<>,/~!@#$%^&*()+{}|: """:
slug = slug.replace(c, '-')
slug = recursive_replace(slug, '-')
slug = collapse(slug, '-')
return slug
def get_repo_slug(request):
return request.environ['pylons.routes_dict'].get('repo_name')
def action_logger(user, action, repo, ipaddr='', sa=None):
Action logger for various actions made by users
:param user: user that made this action, can be a unique username string or
object containing user_id attribute
:param action: action to log, should be on of predefined unique actions for
easy translations
:param repo: string name of repository or object containing repo_id,
that action was made on
:param ipaddr: optional ip address from what the action was made
:param sa: optional sqlalchemy session
if not sa:
sa = meta.Session()
try:
um = UserModel()
if hasattr(user, 'user_id'):
user_obj = user
elif isinstance(user, basestring):
user_obj = um.get_by_username(user, cache=False)
raise Exception('You have to provide user object or username')
rm = RepoModel()
if hasattr(repo, 'repo_id'):
repo_obj = rm.get(repo.repo_id, cache=False)
repo_name = repo_obj.repo_name
elif isinstance(repo, basestring):
repo_name = repo.lstrip('/')
repo_obj = rm.get_by_repo_name(repo_name, cache=False)
raise Exception('You have to provide repository to action logger')
user_log = UserLog()
user_log.user_id = user_obj.user_id
user_log.action = action
user_log.repository_id = repo_obj.repo_id
user_log.repository_name = repo_name
user_log.action_date = datetime.datetime.now()
user_log.user_ip = ipaddr
sa.add(user_log)
sa.commit()
log.info('Adding user %s, action %s on %s', user_obj, action, repo)
except:
log.error(traceback.format_exc())
sa.rollback()
def get_repos(path, recursive=False):
Scans given path for repos and return (name,(type,path)) tuple
:param path: path to scann for repositories
:param recursive: recursive search and return names with subdirs in front
from vcs.utils.helpers import get_scm
from vcs.exceptions import VCSError
if path.endswith('/'):
#add ending slash for better results
path = path[:-1]
def _get_repos(p):
for dirpath in os.listdir(p):
if os.path.isfile(os.path.join(p, dirpath)):
continue
cur_path = os.path.join(p, dirpath)
scm_info = get_scm(cur_path)
yield scm_info[1].split(path)[-1].lstrip('/'), scm_info
yield scm_info[1].split(path)[-1].lstrip(os.sep), scm_info
except VCSError:
if not recursive:
#check if this dir containts other repos for recursive scan
rec_path = os.path.join(p, dirpath)
if os.path.isdir(rec_path):
for inner_scm in _get_repos(rec_path):
yield inner_scm
return _get_repos(path)
def check_repo_fast(repo_name, base_path):
Check given path for existence of directory
:param repo_name:
:param base_path:
:return False: if this directory is present
if os.path.isdir(os.path.join(base_path, repo_name)):return False
return True
def check_repo(repo_name, base_path, verify=True):
repo_path = os.path.join(base_path, repo_name)
if not check_repo_fast(repo_name, base_path):
return False
r = hg.repository(ui.ui(), repo_path)
if verify:
hg.verify(r)
#here we hnow that repo exists it was verified
log.info('%s repo is already created', repo_name)
except RepoError:
#it means that there is no valid repo there...
log.info('%s repo is free for creation', repo_name)
def ask_ok(prompt, retries=4, complaint='Yes or no, please!'):
while True:
ok = raw_input(prompt)
if ok in ('y', 'ye', 'yes'): return True
if ok in ('n', 'no', 'nop', 'nope'): return False
retries = retries - 1
if retries < 0: raise IOError
print complaint
#propagated from mercurial documentation
ui_sections = ['alias', 'auth',
'decode/encode', 'defaults',
'diff', 'email',
'extensions', 'format',
'merge-patterns', 'merge-tools',
'hooks', 'http_proxy',
'smtp', 'patch',
'paths', 'profiling',
'server', 'trusted',
'ui', 'web', ]
def make_ui(read_from='file', path=None, checkpaths=True):
"""A function that will read python rc files or database
and make an mercurial ui object from read options
:param path: path to mercurial config file
:param checkpaths: check the path
:param read_from: read from 'file' or 'db'
baseui = ui.ui()
#clean the baseui object
baseui._ocfg = config.config()
baseui._ucfg = config.config()
baseui._tcfg = config.config()
if read_from == 'file':
if not os.path.isfile(path):
log.warning('Unable to read config file %s' % path)
log.debug('reading hgrc from %s', path)
cfg = config.config()
cfg.read(path)
for section in ui_sections:
for k, v in cfg.items(section):
log.debug('settings ui from file[%s]%s:%s', section, k, v)
baseui.setconfig(section, k, v)
elif read_from == 'db':
ret = sa.query(RhodeCodeUi)\
.options(FromCache("sql_cache_short",
"get_hg_ui_settings")).all()
hg_ui = ret
for ui_ in hg_ui:
if ui_.ui_active:
log.debug('settings ui from db[%s]%s:%s', ui_.ui_section,
ui_.ui_key, ui_.ui_value)
baseui.setconfig(ui_.ui_section, ui_.ui_key, ui_.ui_value)
meta.Session.remove()
return baseui
def set_rhodecode_config(config):
"""Updates pylons config with new settings from database
:param config:
from rhodecode.model.settings import SettingsModel
hgsettings = SettingsModel().get_app_settings()
for k, v in hgsettings.items():
config[k] = v
def invalidate_cache(cache_key, *args):
"""Puts cache invalidation task into db for
further global cache invalidation
from rhodecode.model.scm import ScmModel
if cache_key.startswith('get_repo_cached_'):
name = cache_key.split('get_repo_cached_')[-1]
ScmModel().mark_for_invalidation(name)
class EmptyChangeset(BaseChangeset):
An dummy empty changeset. It's possible to pass hash when creating
an EmptyChangeset
def __init__(self, cs='0' * 40):
self._empty_cs = cs
self.revision = -1
self.message = ''
self.author = ''
self.date = ''
@LazyProperty
def raw_id(self):
"""Returns raw string identifying this changeset, useful for web
representation.
return self._empty_cs
def short_id(self):
return self.raw_id[:12]
def get_file_changeset(self, path):
return self
def get_file_content(self, path):
return u''
def get_file_size(self, path):
return 0
def map_groups(groups):
"""Checks for groups existence, and creates groups structures.
It returns last group in structure
:param groups: list of groups structure
parent = None
group = None
for lvl, group_name in enumerate(groups[:-1]):
group = sa.query(Group).filter(Group.group_name == group_name).scalar()
if group is None:
group = Group(group_name, parent)
sa.add(group)
parent = group
return group
def repo2db_mapper(initial_repo_list, remove_obsolete=False):
"""maps all repos given in initial_repo_list, non existing repositories
are created, if remove_obsolete is True it also check for db entries
that are not in initial_repo_list and removes them.
:param initial_repo_list: list of repositories found by scanning methods
:param remove_obsolete: check for obsolete entries in database
Status change: