Files
@ a83a1799480c
Branch filter:
Location: kallithea/pylons_app/model/hg_model.py
a83a1799480c
3.8 KiB
text/x-python
Reimplemented way of caching repos list, hg model now get's repos objects right from cached dict, this way we skipp creating instances of MercurialRepository and gain performance. Some import cleanup
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | #!/usr/bin/env python
# encoding: utf-8
#
# Copyright (c) 2010 marcink. All rights reserved.
#
'''
Created on Apr 9, 2010
@author: marcink
'''
from beaker.cache import cache_region
from mercurial import ui
from mercurial.hgweb.hgwebdir_mod import findrepos
from pylons import app_globals as g
from vcs.exceptions import RepositoryError, VCSError
import logging
import os
import sys
log = logging.getLogger(__name__)
try:
from vcs.backends.hg import MercurialRepository
except ImportError:
sys.stderr.write('You have to import vcs module')
raise Exception('Unable to import vcs')
@cache_region('long_term', 'cached_repo_list')
def _get_repos_cached():
"""
return cached dict with repos
"""
return HgModel.repo_scan(g.paths[0][0], g.paths[0][1], g.baseui)
@cache_region('long_term', 'full_changelog')
def _full_changelog_cached(repo_name):
log.info('getting full changelog for %s', repo_name)
return list(reversed(list(HgModel().get_repo(repo_name))))
class HgModel(object):
"""
Mercurial Model
"""
def __init__(self):
"""
Constructor
"""
pass
@staticmethod
def repo_scan(repos_prefix, repos_path, baseui):
"""
Listing of repositories in given path. This path should not be a
repository itself. Return a dictionary of repository objects
:param repos_path: path to directory it could take syntax with
* or ** for deep recursive displaying repositories
"""
def check_repo_dir(path):
"""
Checks the repository
:param path:
"""
repos_path = path.split('/')
if repos_path[-1] in ['*', '**']:
repos_path = repos_path[:-1]
if repos_path[0] != '/':
repos_path[0] = '/'
if not os.path.isdir(os.path.join(*repos_path)):
raise RepositoryError('Not a valid repository in %s' % path[0][1])
if not repos_path.endswith('*'):
raise VCSError('You need to specify * or ** at the end of path '
'for recursive scanning')
check_repo_dir(repos_path)
log.info('scanning for repositories in %s', repos_path)
repos = findrepos([(repos_prefix, repos_path)])
if not isinstance(baseui, ui.ui):
baseui = ui.ui()
repos_list = {}
for name, path in repos:
try:
repos_list[name] = MercurialRepository(path, baseui=baseui)
except OSError:
continue
return repos_list
def get_repos(self):
for name, repo in _get_repos_cached().items():
if repo._get_hidden():
#skip hidden web repository
continue
last_change = repo.last_change
try:
tip = repo.get_changeset('tip')
except RepositoryError:
from pylons_app.lib.utils import EmptyChangeset
tip = EmptyChangeset()
tmp_d = {}
tmp_d['name'] = repo.name
tmp_d['name_sort'] = tmp_d['name'].lower()
tmp_d['description'] = repo.description
tmp_d['description_sort'] = tmp_d['description']
tmp_d['last_change'] = last_change
tmp_d['last_change_sort'] = last_change[1] - last_change[0]
tmp_d['tip'] = tip.raw_id
tmp_d['tip_sort'] = tip.revision
tmp_d['rev'] = tip.revision
tmp_d['contact'] = repo.contact
tmp_d['contact_sort'] = tmp_d['contact']
tmp_d['repo_archives'] = list(repo._get_archives())
yield tmp_d
def get_repo(self, repo_name):
return _get_repos_cached()[repo_name]
|