Source

trac-hgdirmanager / hgdirmanager / hgdirmanager.py

from trac.core import *
from trac.perm import IPermissionRequestor
from trac.util.html import html
from trac.util.translation import _
from trac.web import IRequestHandler, HTTPNotFound
from trac.web.chrome import ITemplateProvider, add_warning, add_notice
from trac.versioncontrol.api import IRepositoryProvider

import os, os.path
from string import ascii_lowercase, digits

class HgDirProvider(Component):
    """Provides Mercurial repositories from a directory."""

    implements(IRepositoryProvider)

    # IRepositoryProvider
    def get_repositories(self):
        """Retrieve repositories specified in the repository DB table."""

        basepath = self.config.get('hgdir', 'base_path')
        if not basepath:
            return []

        recurse = self.config.getbool('hgdir', 'recurse', False)
        url_prefix = self.config.get('hgdir', 'url_prefix')

        repos = []

        def add_repo(topdir):
            relpath = os.path.relpath(topdir, basepath)
            info = {
                'dir': topdir,
                'type': 'hg'
                }

            try:
                hgrc_path = os.path.join(topdir, '.hg', 'hgrc')

                from mercurial.config import config
                cfg = config()
                cfg.read(hgrc_path)
                desc = cfg.get('web', 'description', None)
                if desc:
                    info['description'] = desc
            except:
                pass

            info['url'] = url_prefix + relpath
            repos.append((relpath, info))

        if recurse:
            for topdir, dirs, files in os.walk(basepath):
                if '.hg' in dirs:
                    add_repo(topdir)
        else:
            for name in os.listdir(basepath):
                if os.path.isdir(os.path.join(basepath, name, '.hg')):
                    add_repo(os.path.join(basepath, name))

        return repos

class HgDirManager(Component):
    """Provides the ability to create and delete Mercurial repositories."""
    implements(IRequestHandler, ITemplateProvider, IPermissionRequestor)

    # IPermissionRequestor
    def get_permission_actions(self):
        return ['REPO_CREATE', 'REPO_DELETE', 'REPO_DELETE_NONEMPTY']

    def _new_repo(self, req):
        req.perm.require('REPO_CREATE')
        if req.method == 'POST':
            reponame = req.args.get('field_name')
            if reponame == '':
                add_warning(req, _('The name of the new repository must be non-empty.'))
            elif any(ch != '_' and ch not in digits and ch not in ascii_lowercase for ch in reponame):
                add_warning(req, _('Repository names may only contain lowercase letters, digits and underscores.'))
            elif reponame[0] == '_':
                add_warning(req, _('Repository names may not begin with underscores.'))
            elif reponame and '/' not in reponame:
                basepath = self.config.get('hgdir', 'base_path')
                repodir = os.path.join(basepath, reponame)

                if os.path.exists(repodir):
                    add_warning(req, _('Cannot create repository: the directory already exists.'))
                else:
                    from mercurial.localrepo import localrepository
                    from mercurial import ui

                    try:
                        localrepository(ui.ui(), repodir, create=True)
                    except Exception, e:
                        add_warning(req, _('Failed to create repository: ') + str(e))
                    else:
                        req.redirect(req.href.wiki())

        return 'newhgrepo.html', {}, None

    def _delete_repo(self, req, reponame):
        req.perm.require('REPO_DELETE')

        basepath = self.config.get('hgdir', 'base_path')
        repodir = os.path.join(basepath, reponame)
        if not os.path.exists(repodir):
            raise HTTPNotFound(_('There is no such repository'))

        nonempty = True
        try:
            from mercurial.localrepo import localrepository
            from mercurial import ui
            hgrepo = localrepository(ui.ui(), repodir)
            if len(hgrepo.changelog) == 0:
                nonempty = False
            hgrepo.close()
        except:
            pass

        if nonempty:
            req.perm.require('REPO_DELETE_NONEMPTY')

        if req.method == 'POST':
            if 'confirm' not in req.args:
                req.redirect(req.href.wiki())

            # delete the directory here
            import shutil
            add_notice(req, _('Removed repository: ') + reponame)
            shutil.rmtree(repodir)
            req.redirect(req.href.wiki())

        return 'deletehgrepo.html', {'reponame': reponame, 'nonempty': nonempty}, None

    # IRequestHandler
    def match_request(self, req):
        return req.path_info == '/newhgrepo' or req.path_info.startswith('/deletehgrepo/')

    # IRequestHandler
    def process_request(self, req):
        if req.path_info == '/newhgrepo':
            return self._new_repo(req)
        elif req.path_info.startswith('/deletehgrepo/'):
            return self._delete_repo(req, req.path_info.split('/', 2)[2])

    # ITemplateProvider
    def get_templates_dirs(self):
        from pkg_resources import resource_filename
        return [resource_filename(__name__, 'templates')]
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.