kallithea Changeset - 0e5455fda8fd

Changeset - 0e5455fda8fd

Parent rev.

Child rev.

[Not reviewed]

default

0 13 1

Marcin Kuzminski - 15 years ago 2010-06-07 00:18:33
marcin@python-works.com

Implemented basic repository managment. Implemented repo2db mappings, model, helpers updates and code cleanups

14 files changed with 327 insertions and 75 deletions:

pylons_app/config/environment.py

pylons_app/controllers/repos.py

pylons_app/controllers/users.py

pylons_app/lib/auth.py

pylons_app/lib/helpers.py

pylons_app/lib/utils.py

pylons_app/model/db.py

pylons_app/model/forms.py

pylons_app/model/hg_model.py

pylons_app/model/repo_model.py

106

pylons_app/model/user_model.py

pylons_app/templates/admin/repos/repo_add.html

pylons_app/templates/admin/repos/repo_edit.html

pylons_app/templates/summary/summary.html

0 comments (0 inline, 0 general)

pylons_app/config/environment.py

➞

Show inline comments

 """Pylons environment configuration"""
 from mako.lookup import TemplateLookup
 from pylons.configuration import PylonsConfig
 from pylons.error import handle_mako_error
 from pylons_app.config.routing import make_map
 from pylons_app.lib.auth import set_available_permissions
 from pylons_app.lib.utils import repo2db_mapper
 from pylons_app.model import init_model
 from pylons_app.model.hg_model import _get_repos_cached_initial
 from sqlalchemy import engine_from_config
 import logging
 import os
 import pylons_app.lib.app_globals as app_globals
 import pylons_app.lib.helpers
 log = logging.getLogger(__name__)
 def load_environment(global_conf, app_conf):
     """Configure the Pylons environment via the ``pylons.config``
     object
     """
     config = PylonsConfig()
     # Pylons paths
     root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
     paths = dict(root=root,
                  controllers=os.path.join(root, 'controllers'),
                  static_files=os.path.join(root, 'public'),
                  templates=[os.path.join(root, 'templates')])
     # Initialize config with the basic options
     config.init_app(global_conf, app_conf, package='pylons_app', paths=paths)
@@ @@ -39,30 +40,30 @@ def load_environment(global_conf, app_co @@
     pylons.cache._push_object(config['pylons.app_globals'].cache)
     # Create the Mako TemplateLookup, with the default auto-escaping
     config['pylons.app_globals'].mako_lookup = TemplateLookup(
         directories=paths['templates'],
         error_handler=handle_mako_error,
         module_directory=os.path.join(app_conf['cache_dir'], 'templates'),
         input_encoding='utf-8', default_filters=['escape'],
         imports=['from webhelpers.html import escape'])
     #sets the c attribute access when don't existing attribute are accessed
     config['pylons.strict_tmpl_context'] = True
     #MULTIPLE DB configs
     # Setup the SQLAlchemy database engine
     if config['debug']:
         #use query time debugging.
         from pylons_app.lib.timerproxy import TimerProxy
         sa_engine_db1 = engine_from_config(config, 'sqlalchemy.db1.',
                                                             proxy=TimerProxy())
     else:
         sa_engine_db1 = engine_from_config(config, 'sqlalchemy.db1.')
     init_model(sa_engine_db1)
     repo2db_mapper()
+    repo2db_mapper(_get_repos_cached_initial(config['pylons.app_globals']))
     set_available_permissions(config)
     # CONFIGURATION OPTIONS HERE (note: all config options will override
     # any Pylons config options)
     return config

pylons_app/controllers/repos.py

➞

Show inline comments

 #!/usr/bin/env python
 # encoding: utf-8
 # repos controller for pylons
 # Copyright (C) 2009-2010 Marcin Kuzminski <marcin@python-works.com>
 # 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.
 """
 Created on April 7, 2010
 admin controller for pylons
 @author: marcink
 """
-import logging
+from operator import itemgetter
 from pylons import request, response, session, tmpl_context as c, url, \
     app_globals as g
 from pylons.controllers.util import abort, redirect
 from pylons_app.lib.auth import LoginRequired
 from pylons.i18n.translation import _
 from pylons_app.lib import helpers as h
 from pylons_app.lib.auth import LoginRequired
 from pylons_app.lib.base import BaseController, render
 from pylons_app.lib.filters import clean_repo
 from pylons_app.lib.utils import check_repo, invalidate_cache
 from pylons_app.lib.utils import invalidate_cache
 from pylons_app.model.repo_model import RepoModel
 from pylons_app.model.hg_model import HgModel
 from pylons_app.model.forms import RepoForm
 from pylons_app.model.meta import Session
 from datetime import datetime
 import formencode
 from formencode import htmlfill
 import logging
 import os
 import shutil
 from operator import itemgetter
 log = logging.getLogger(__name__)
 class ReposController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('repo', 'repos')
     @LoginRequired()
     def __before__(self):
         c.admin_user = session.get('admin_user')
         c.admin_username = session.get('admin_username')
         super(ReposController, self).__before__()
     def index(self, format='html'):
         """GET /repos: All items in the collection"""
         # url('repos')
         cached_repo_list = HgModel().get_repos()
         c.repos_list = sorted(cached_repo_list, key=itemgetter('name_sort'))
         return render('admin/repos/repos.html')
     def create(self):
         """POST /repos: Create a new item"""
         # url('repos')
         name = request.POST.get('name')
         repo_model = RepoModel()
         _form = RepoForm()()
         try:
             self._create_repo(name)
             #clear our cached list for refresh with new repo
             form_result = _form.to_python(dict(request.POST))
             repo_model.create(form_result, c.hg_app_user)
             invalidate_cache('cached_repo_list')
             h.flash(_('created repository %s') % name, category='success')
         except Exception as e:
             log.error(e)
             h.flash(_('created repository %s') % form_result['repo_name'],
                     category='success')
         except formencode.Invalid as errors:
             c.form_errors = errors.error_dict
             c.new_repo = errors.value['repo_name']
             return htmlfill.render(
                  render('admin/repos/repo_add.html'),
                 defaults=errors.value,
                 encoding="UTF-8")
         except Exception:
             h.flash(_('error occured during creation of repository %s') \
                     % form_result['repo_name'], category='error')
         return redirect('repos')
     def _create_repo(self, repo_name):
         repo_path = os.path.join(g.base_path, repo_name)
         if check_repo(repo_name, g.base_path):
             log.info('creating repo %s in %s', repo_name, repo_path)
             from vcs.backends.hg import MercurialRepository
             MercurialRepository(repo_path, create=True)
     def new(self, format='html'):
         """GET /repos/new: Form to create a new item"""
         new_repo = request.GET.get('repo', '')
-        c.new_repo = clean_repo(new_repo)
+        c.new_repo = h.repo_name_slug(new_repo)
         return render('admin/repos/repo_add.html')
     def update(self, id):
         """PUT /repos/id: Update an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="PUT" />
         # Or using helpers:
         #    h.form(url('repo', id=ID),
         #           method='put')
         # url('repo', id=ID)
         repo_model = RepoModel()
         _form = RepoForm(edit=True)()
         try:
             form_result = _form.to_python(dict(request.POST))
             repo_model.update(id, form_result)
             invalidate_cache('cached_repo_list')
             h.flash(_('Repository updated succesfully'), category='success')
         except formencode.Invalid as errors:
             c.repo_info = repo_model.get(id)
             c.form_errors = errors.error_dict
             return htmlfill.render(
                  render('admin/repos/repo_edit.html'),
                 defaults=errors.value,
                 encoding="UTF-8")
         except Exception:
             h.flash(_('error occured during update of repository %s') \
                     % form_result['repo_name'], category='error')
         return redirect(url('repos'))
     def delete(self, id):
         """DELETE /repos/id: Delete an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="DELETE" />
         # Or using helpers:
         #    h.form(url('repo', id=ID),
         #           method='delete')
         # url('repo', id=ID)
         from datetime import datetime
         path = g.paths[0][1].replace('*', '')
         rm_path = os.path.join(path, id)
         log.info("Removing %s", rm_path)
         shutil.move(os.path.join(rm_path, '.hg'), os.path.join(rm_path, 'rm__.hg'))
         shutil.move(rm_path, os.path.join(path, 'rm__%s-%s' % (datetime.today(), id)))
         repo_model = RepoModel()
         repo = repo_model.get(id)
         if not repo:
             h.flash(_('%s repository is not mapped to db perhaps'
                       ' it was moved or renamed please run the application again'
                       ' in order to rescan repositories') % id, category='error')
         #clear our cached list for refresh with new repo
         invalidate_cache('cached_repo_list')
         h.flash(_('deleted repository %s') % rm_path, category='success')
             return redirect(url('repos'))
         try:
             repo_model.delete(repo)
             invalidate_cache('cached_repo_list')
             h.flash(_('deleted repository %s') % id, category='success')
         except Exception:
             h.flash(_('An error occured during deletion of %s') % id,
                     category='error')
         return redirect(url('repos'))
     def show(self, id, format='html'):
         """GET /repos/id: Show a specific item"""
         # url('repo', id=ID)
     def edit(self, id, format='html'):
         """GET /repos/id/edit: Form to edit an existing item"""
         # url('edit_repo', id=ID)
         c.new_repo = id
         return render('admin/repos/repo_edit.html')
         repo_model = RepoModel()
         c.repo_info = repo_model.get(id)
         defaults = c.repo_info.__dict__
         defaults.update({'user':c.repo_info.user.username})
         return htmlfill.render(
             render('admin/repos/repo_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )

pylons_app/controllers/users.py

➞

Show inline comments

@@ @@ -2,138 +2,146 @@ @@
 # encoding: utf-8
 # users controller for pylons
 # Copyright (C) 2009-2010 Marcin Kuzminski <marcin@python-works.com>
 # 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.
 """
 Created on April 4, 2010
 users controller for pylons
 @author: marcink
 """
 import logging
 from formencode import htmlfill
 from pylons import request, session, tmpl_context as c, url
 from pylons.controllers.util import abort, redirect
 from pylons.i18n.translation import _
 from pylons_app.lib import helpers as h
 from pylons_app.lib.auth import LoginRequired, CheckPermissionAll
 from pylons_app.lib.base import BaseController, render
 from pylons_app.model.db import User, UserLog
 from pylons_app.model.forms import UserForm
 from pylons_app.model.user_model import UserModel
 import formencode
 from formencode import htmlfill
 log = logging.getLogger(__name__)
 class UsersController(BaseController):
     """REST Controller styled on the Atom Publishing Protocol"""
     # To properly map this controller, ensure your config/routing.py
     # file has a resource setup:
     #     map.resource('user', 'users')
     @LoginRequired()
     def __before__(self):
         c.admin_user = session.get('admin_user')
         c.admin_username = session.get('admin_username')
         super(UsersController, self).__before__()
     def index(self, format='html'):
         """GET /users: All items in the collection"""
         # url('users')
         c.users_list = self.sa.query(User).all()
         return render('admin/users/users.html')
     def create(self):
         """POST /users: Create a new item"""
         # url('users')
         user_model = UserModel()
         login_form = UserForm()()
         try:
             form_result = login_form.to_python(dict(request.POST))
             user_model.create(form_result)
             h.flash(_('created user %s') % form_result['username'], category='success')
             return redirect(url('users'))
             h.flash(_('created user %s') % form_result['username'],
                     category='success')
         except formencode.Invalid as errors:
             c.form_errors = errors.error_dict
             return htmlfill.render(
                  render('admin/users/user_add.html'),
                 defaults=errors.value,
                 encoding="UTF-8")
         except Exception:
             h.flash(_('error occured during creation of user %s') \
                     % form_result['username'], category='error')
         return redirect(url('users'))
     def new(self, format='html'):
         """GET /users/new: Form to create a new item"""
         # url('new_user')
         return render('admin/users/user_add.html')
     def update(self, id):
         """PUT /users/id: Update an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="PUT" />
         # Or using helpers:
         #    h.form(url('user', id=ID),
         #           method='put')
         # url('user', id=ID)
         user_model = UserModel()
-        login_form = UserForm(edit=True)()
+        _form = UserForm(edit=True)()
         try:
-            form_result = login_form.to_python(dict(request.POST))
+            form_result = _form.to_python(dict(request.POST))
             user_model.update(id, form_result)
             h.flash(_('User updated succesfully'), category='success')
             return redirect(url('users'))
         except formencode.Invalid as errors:
             c.user = user_model.get_user(id)
             c.form_errors = errors.error_dict
             return htmlfill.render(
                  render('admin/users/user_edit.html'),
                 defaults=errors.value,
                 encoding="UTF-8")
         except Exception:
             h.flash(_('error occured during update of user %s') \
                     % form_result['username'], category='error')
         return redirect(url('users'))
     def delete(self, id):
         """DELETE /users/id: Delete an existing item"""
         # Forms posted to this method should contain a hidden field:
         #    <input type="hidden" name="_method" value="DELETE" />
         # Or using helpers:
         #    h.form(url('user', id=ID),
         #           method='delete')
         # url('user', id=ID)
         user_model = UserModel()
         try:
             self.sa.delete(self.sa.query(User).get(id))
             self.sa.commit()
             user_model.delete(id)
             h.flash(_('sucessfully deleted user'), category='success')
         except:
             self.sa.rollback()
             raise
         except Exception:
             h.flash(_('An error occured during deletion of user'),
                     category='error')
         return redirect(url('users'))
     def show(self, id, format='html'):
         """GET /users/id: Show a specific item"""
         # url('user', id=ID)
     def edit(self, id, format='html'):
         """GET /users/id/edit: Form to edit an existing item"""
         # url('edit_user', id=ID)
         c.user = self.sa.query(User).get(id)
         defaults = c.user.__dict__
         return htmlfill.render(
             render('admin/users/user_edit.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )

pylons_app/lib/auth.py

➞

Show inline comments

@@ @@ -44,48 +44,49 @@ def get_crypt_password(password): @@
 def authfunc(environ, username, password):
     sa = meta.Session
     password_crypt = get_crypt_password(password)
     try:
         user = sa.query(User).filter(User.username == username).one()
     except (NoResultFound, MultipleResultsFound, OperationalError) as e:
         log.error(e)
         user = None
     if user:
         if user.active:
             if user.username == username and user.password == password_crypt:
                 log.info('user %s authenticated correctly', username)
                 return True
         else:
             log.error('user %s is disabled', username)
     return False
 class  AuthUser(object):
     """
     A simple object that handles a mercurial username for authentication
     """
     username = 'None'
     user_id = None
     is_authenticated = False
     is_admin = False
     permissions = set()
     group = set()
     def __init__(self):
         pass
 def set_available_permissions(config):
     """
     This function will propagate pylons globals with all available defined
     permission given in db. We don't wannt to check each time from db for new
     permissions since adding a new permission also requires application restart
     ie. to decorate new views with the newly created permission
     @param config:
     """
     from pylons_app.model.meta import Session
     from pylons_app.model.db import Permission
     logging.info('getting information about all available permissions')
     sa = Session()
     all_perms = sa.query(Permission).all()
     config['pylons.app_globals'].available_permissions = [x.permission_name for x in all_perms]

pylons_app/lib/helpers.py

➞

Show inline comments

@@ @@ -93,35 +93,33 @@ def pygmentize_annotation(filenode, **kw @@
     return literal(annotate_highlight(filenode, url_func, **kwargs))
 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
     """
     slug = urlify(value)
     for c in """=[]\;',/~!@#$%^&*()+{}|:""":
+    for c in """=[]\;'"<>,/~!@#$%^&*()+{}|:""":
         slug = slug.replace(c, '-')
     print slug
     slug = recursive_replace(slug, '-')
     print slug
     return slug
 files_breadcrumbs = _FilesBreadCrumbs()
 link = _Link()
 flash = _Flash()
 get_error = _GetError()

pylons_app/lib/utils.py

➞

Show inline comments

@@ @@ -7,48 +7,49 @@ @@
 # 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.
 """
 Created on April 18, 2010
 Utilities for hg app
 @author: marcink
 """
 import os
 import logging
 from mercurial import ui, config, hg
 from mercurial.error import RepoError
 from pylons_app.model.db import Repository, User
 log = logging.getLogger(__name__)
 def get_repo_slug(request):
     return request.environ['pylons.routes_dict'].get('repo_name')
 def is_mercurial(environ):
     """
     Returns True if request's target is mercurial server - header
     ``HTTP_ACCEPT`` of such request would start with ``application/mercurial``.
     """
     http_accept = environ.get('HTTP_ACCEPT')
     if http_accept and http_accept.startswith('application/mercurial'):
         return True
     return False
 def check_repo_dir(paths):
     repos_path = paths[0][1].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 Exception('Not a valid repository in %s' % paths[0][1])
@@ @@ -131,30 +132,49 @@ def invalidate_cache(name, *args): @@
     if name == 'cached_repo_list':
         from pylons_app.model.hg_model import _get_repos_cached
         region_invalidate(_get_repos_cached, None, *args)
     if name == 'full_changelog':
         from pylons_app.model.hg_model import _full_changelog_cached
         region_invalidate(_full_changelog_cached, None, *args)
 from vcs.backends.base import BaseChangeset
 from vcs.utils.lazy import LazyProperty
 class EmptyChangeset(BaseChangeset):
     revision = -1
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifing this changeset, useful for web
         representation.
         """
         return '0' * 12
 def repo2db_mapper():
 def repo2db_mapper(initial_repo_list):
     """
     maps all found repositories into db
     """
     scann all dirs for .hgdbid
     if some dir doesn't have one generate one.
     """
     pass
     from pylons_app.model.meta import Session
     sa = Session()
     user = sa.query(User).filter(User.admin == True).first()
     for name, repo in initial_repo_list.items():
         if not sa.query(Repository).get(name):
             log.info('%s not found creating default', name)
             try:
                 new_repo = Repository()
                 new_repo.repo_name = name
                 desc = repo.description if repo.description != 'unknown' else \
                                             'auto description for %s' % name
                 new_repo.description = desc
                 new_repo.user_id = user.user_id
                 new_repo.private = False
                 sa.add(new_repo)
                 sa.commit()
             except:
                 sa.rollback()
                 raise

pylons_app/model/db.py

➞

Show inline comments

 from pylons_app.model.meta import Base
 from sqlalchemy.orm import relation, backref
 from sqlalchemy import *
 from vcs.utils.lazy import LazyProperty
 class User(Base):
     __tablename__ = 'users'
     __table_args__ = {'useexisting':True}
-    user_id = Column("user_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=1)
+    user_id = Column("user_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=True)
     username = Column("username", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     password = Column("password", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     active = Column("active", BOOLEAN(), nullable=True, unique=None, default=None)
     admin = Column("admin", BOOLEAN(), nullable=True, unique=None, default=None)
     name = Column("name", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     lastname = Column("lastname", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     email = Column("email", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     last_login = Column("last_login", DATETIME(timezone=False), nullable=True, unique=None, default=None)
     user_log = relation('UserLog')
     @LazyProperty
     def full_contact(self):
         return '%s %s <%s>' % (self.name, self.lastname, self.email)
     def __repr__(self):
         return "<User('%s:%s')>" % (self.user_id, self.username)
 class UserLog(Base):
     __tablename__ = 'user_logs'
     __table_args__ = {'useexisting':True}
     user_log_id = Column("user_log_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=1)
-    user_id = Column("user_id", INTEGER(), ForeignKey(u'users.user_id'), nullable=True, unique=None, default=None)
+    user_id = Column("user_id", INTEGER(), ForeignKey(u'users.user_id'), nullable=False, unique=None, default=None)
     repository = Column("repository", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     action = Column("action", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     action_date = Column("action_date", DATETIME(timezone=False), nullable=True, unique=None, default=None)
     user = relation('User')
 class Repository(Base):
     __tablename__ = 'repositories'
     repo_id = Column("repo_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=1)
     repo_name = Column("repo_name", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     user_id = Column("user_id", INTEGER(), ForeignKey(u'users.user_id'), nullable=True, unique=None, default=None)
     repo_name = Column("repo_name", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=False, unique=True, default=None, primary_key=True)
     user_id = Column("user_id", INTEGER(), ForeignKey(u'users.user_id'), nullable=False, unique=False, default=None)
     private = Column("private", BOOLEAN(), nullable=True, unique=None, default=None)
+    description = Column("description", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     user = relation('User')
 class Permission(Base):
     __tablename__ = 'permissions'
     __table_args__ = {'useexisting':True}
     permission_id = Column("id", INTEGER(), nullable=False, unique=True, default=None, primary_key=1)
     permission_name = Column("permission_name", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     permission_longname = Column("permission_longname", TEXT(length=None, convert_unicode=False, assert_unicode=None), nullable=True, unique=None, default=None)
     def __repr__(self):
         return "<Permission('%s:%s')>" % (self.permission_id, self.permission_name)

pylons_app/model/forms.py

➞

Show inline comments

@@ @@ -4,50 +4,51 @@ for list off all availible validators @@
 we can create our own validators
 The table below outlines the options which can be used in a schema in addition to the validators themselves
 pre_validators          []     These validators will be applied before the schema
 chained_validators      []     These validators will be applied after the schema
 allow_extra_fields      False     If True, then it is not an error when keys that aren't associated with a validator are present
 filter_extra_fields     False     If True, then keys that aren't associated with a validator are removed
 if_key_missing          NoDefault If this is given, then any keys that aren't available but are expected will be replaced with this value (and then validated). This does not override a present .if_missing attribute on validators. NoDefault is a special FormEncode class to mean that no default values has been specified and therefore missing keys shouldn't take a default value.
 ignore_key_missing      False     If True, then missing keys will be missing in the result, if the validator doesn't have .if_missing on it already
 <name> = formencode.validators.<name of validator>
 <name> must equal form name
 list=[1,2,3,4,5]
 for SELECT use formencode.All(OneOf(list), Int())
 """
 from formencode import All
 from formencode.validators import UnicodeString, OneOf, Int, Number, Regex, \
     Email, Bool, StringBoolean
 from pylons import session
 from pylons.i18n.translation import _
 from pylons_app.lib.auth import get_crypt_password
 import pylons_app.lib.helpers as h
 from pylons_app.model import meta
 from pylons_app.model.db import User
+from pylons_app.model.db import User, Repository
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound
 from webhelpers.pylonslib.secure_form import authentication_token
 import datetime
 import formencode
 import logging
 log = logging.getLogger(__name__)
 #this is needed to translate the messages using _() in validators
 class State_obj(object):
     _ = staticmethod(_)
 #===============================================================================
 # VALIDATORS
 #===============================================================================
 class ValidAuthToken(formencode.validators.FancyValidator):
     messages = {'invalid_token':_('Token mismatch')}
     def validate_python(self, value, state):
         if value != authentication_token():
             raise formencode.Invalid(self.message('invalid_token', state,
                                             search_number=value), value, state)
@@ @@ -72,94 +73,131 @@ class ValidAuth(formencode.validators.Fa @@
     e_dict = {'username':messages['invalid_login'],
               'password':messages['invalid_password']}
     e_dict_disable = {'username':messages['disabled_account']}
     def validate_python(self, value, state):
         sa = meta.Session
         crypted_passwd = get_crypt_password(value['password'])
         username = value['username']
         try:
             user = sa.query(User).filter(User.username == username).one()
         except (NoResultFound, MultipleResultsFound, OperationalError) as e:
             log.error(e)
             user = None
             raise formencode.Invalid(self.message('invalid_password',
                                      state=State_obj), value, state,
                                      error_dict=self.e_dict)
         if user:
             if user.active:
                 if user.username == username and user.password == crypted_passwd:
                     from pylons_app.lib.auth import AuthUser
                     auth_user = AuthUser()
                     auth_user.username = username
                     auth_user.is_authenticated = True
                     auth_user.is_admin = user.admin
                     auth_user.user_id = user.user_id
                     session['hg_app_user'] = auth_user
                     session.save()
                     log.info('user %s is now authenticated', username)
                     try:
                         user.last_login = datetime.datetime.now()
                         sa.add(user)
                         sa.commit()
                     except (OperationalError) as e:
                         log.error(e)
                         sa.rollback()
                     return value
                 else:
                     log.warning('user %s not authenticated', username)
                     raise formencode.Invalid(self.message('invalid_password',
                                              state=State_obj), value, state,
                                              error_dict=self.e_dict)
             else:
                 log.warning('user %s is disabled', username)
                 raise formencode.Invalid(self.message('disabled_account',
                                          state=State_obj),
                                          value, state,
                                          error_dict=self.e_dict_disable)
 class ValidRepoUser(formencode.validators.FancyValidator):
     def to_python(self, value, state):
         sa = meta.Session
         try:
             self.user_db = sa.query(User).filter(User.username == value).one()
         except Exception:
             raise formencode.Invalid(_('This username is not valid'),
                                      value, state)
         return self.user_db.user_id
 def ValidRepoName(edit=False):
     class _ValidRepoName(formencode.validators.FancyValidator):
         def to_python(self, value, state):
             slug = h.repo_name_slug(value)
             sa = meta.Session
             if sa.query(Repository).get(slug) and not edit:
                 raise formencode.Invalid(_('This repository already exists'),
                                          value, state)
             return slug
     return _ValidRepoName
 #===============================================================================
 # FORMS
 #===============================================================================
 class LoginForm(formencode.Schema):
     allow_extra_fields = True
     filter_extra_fields = True
     username = UnicodeString(
                              strip=True,
                              min=3,
                              not_empty=True,
                              messages={
                                        'empty':_('Please enter a login'),
                                        'tooShort':_('Enter a value %(min)i characters long or more')}
+                            )
     password = UnicodeString(
                             strip=True,
                             min=3,
                             not_empty=True,
                             messages={
                                       'empty':_('Please enter a password'),
                                       'tooShort':_('Enter a value %(min)i characters long or more')}
+                                )
     #chained validators have access to all data
     chained_validators = [ValidAuth]
 def UserForm(edit=False):
     class _UserForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(UnicodeString(strip=True, min=3, not_empty=True), ValidUsername)
         if edit:
             new_password = All(UnicodeString(strip=True, min=3, not_empty=False), ValidPassword)
         else:
             password = All(UnicodeString(strip=True, min=3, not_empty=False), ValidPassword)
         active = StringBoolean(if_missing=False)
         name = UnicodeString(strip=True, min=3, not_empty=True)
         lastname = UnicodeString(strip=True, min=3, not_empty=True)
         email = Email(not_empty=True)
     return _UserForm
 def RepoForm(edit=False):
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         repo_name = All(UnicodeString(strip=True, min=1, not_empty=True), ValidRepoName(edit))
         description = UnicodeString(strip=True, min=3, not_empty=True)
         private = StringBoolean(if_missing=False)
         if edit:
             user = All(Int(not_empty=True), ValidRepoUser)
     return _RepoForm

pylons_app/model/hg_model.py

➞

Show inline comments

@@ @@ -6,125 +6,138 @@ @@
 # 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.
 """
 Created on April 9, 2010
 Model for hg app
 @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
 from pylons_app.model.meta import Session
 from pylons_app.model.db import Repository
 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')
 def _get_repos_cached_initial(app_globals):
     """
     return cached dict with repos
     """
     g = app_globals
     return HgModel.repo_scan(g.paths[0][0], g.paths[0][1], g.baseui)
 @cache_region('long_term', 'cached_repo_list')
 def _get_repos_cached():
     """
     return cached dict with repos
     """
     from pylons import app_globals as g
     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
         """
         sa = Session()
         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:
                 #name = name.split('/')[-1]
                 if repos_list.has_key(name):
                     raise RepositoryError('Duplicate repository name %s found in'
                                     ' %s' % (name, path))
                 else:
                     repos_list[name] = MercurialRepository(path, baseui=baseui)
                     repos_list[name].name = name
                     dbrepo = sa.query(Repository).get(name)
                     if dbrepo:
                         repos_list[name].description = dbrepo.description
                         repos_list[name].contact = dbrepo.user.full_contact
             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]

pylons_app/model/repo_model.py

➞

Show inline comments

@@ new file 100644 @@
 #!/usr/bin/env python
 # encoding: utf-8
 # model for handling repositories actions
 # Copyright (C) 2009-2010 Marcin Kuzminski <marcin@python-works.com>
 # 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.
 """
 Created on Jun 5, 2010
 model for handling repositories actions
 @author: marcink
 """
 from pylons_app.model.meta import Session
 from pylons_app.model.db import Repository
 import shutil
 import os
 from datetime import datetime
 from pylons_app.lib.utils import check_repo
 from pylons import app_globals as g
 import logging
 log = logging.getLogger(__name__)
 class RepoModel(object):
     def __init__(self):
         self.sa = Session()
     def get(self, id):
         return self.sa.query(Repository).get(id)
     def update(self, id, form_data):
         try:
             if id != form_data['repo_name']:
                 self.__rename_repo(id, form_data['repo_name'])
             cur_repo = self.sa.query(Repository).get(id)
             for k, v in form_data.items():
                 if k == 'user':
                     cur_repo.user_id = v
                 else:
                     setattr(cur_repo, k, v)
             self.sa.add(cur_repo)
             self.sa.commit()
         except Exception as e:
             log.error(e)
             self.sa.rollback()
             raise
     def create(self, form_data, cur_user):
         try:
             new_repo = Repository()
             for k, v in form_data.items():
                 setattr(new_repo, k, v)
             new_repo.user_id = cur_user.user_id
             self.sa.add(new_repo)
             self.sa.commit()
             self.__create_repo(form_data['repo_name'])
         except Exception as e:
             log.error(e)
             self.sa.rollback()
             raise
     def delete(self, repo):
         try:
             self.sa.delete(repo)
             self.sa.commit()
             self.__delete_repo(repo.repo_name)
         except Exception as e:
             log.error(e)
             self.sa.rollback()
             raise
     def __create_repo(self, repo_name):
         repo_path = os.path.join(g.base_path, repo_name)
         if check_repo(repo_name, g.base_path):
             log.info('creating repo %s in %s', repo_name, repo_path)
             from vcs.backends.hg import MercurialRepository
             MercurialRepository(repo_path, create=True)
     def __rename_repo(self, old, new):
         log.info('renaming repoo from %s to %s', old, new)
         old_path = os.path.join(g.base_path, old)
         new_path = os.path.join(g.base_path, new)
         shutil.move(old_path, new_path)
     def __delete_repo(self, name):
         rm_path = os.path.join(g.base_path, name)
         log.info("Removing %s", rm_path)
         #disable hg
         shutil.move(os.path.join(rm_path, '.hg'), os.path.join(rm_path, 'rm__.hg'))
         #disable repo
         shutil.move(rm_path, os.path.join(g.base_path, 'rm__%s-%s' % (datetime.today(), id)))

pylons_app/model/user_model.py

➞

Show inline comments

@@ @@ -5,60 +5,73 @@ @@
 # 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.
 """
 Created on April 9, 2010
 Model for users
 @author: marcink
 """
 from pylons_app.model.db import User
 from pylons_app.model.meta import Session
 import logging
 log = logging.getLogger(__name__)
 class UserModel(object):
     def __init__(self):
         self.sa = Session()
     def get_user(self, id):
         return self.sa.query(User).get(id)
     def create(self, form_data):
         try:
             new_user = User()
             for k, v in form_data.items():
                 setattr(new_user, k, v)
             self.sa.add(new_user)
             self.sa.commit()
         except:
         except Exception as e:
             log.error(e)
             self.sa.rollback()
             raise
     def update(self, id, form_data):
         try:
             new_user = self.sa.query(User).get(id)
             for k, v in form_data.items():
                 if k == 'new_password' and v != '':
                     new_user.password = v
                 else:
                     setattr(new_user, k, v)
             self.sa.add(new_user)
             self.sa.commit()
         except:
         except Exception as e:
             log.error(e)
             self.sa.rollback()
             raise
     def delete(self, id):
         try:
             self.sa.delete(self.sa.query(User).get(id))
             self.sa.commit()
         except Exception as e:
             log.error(e)
             self.sa.rollback()
             raise

pylons_app/templates/admin/repos/repo_add.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="/base/base.html"/>
 <%def name="title()">
     ${_('Repositories administration')}
 </%def>
 <%def name="breadcrumbs()">
 	${h.link_to(u'Admin',h.url('admin_home'))}
+	 /
 	${_('Repos')}
 </%def>
 <%def name="page_nav()">
 	${self.menu('admin')}
 	${self.submenu('repos')}
 </%def>
 <%def name="main()">
 	<div>
         <h2>${_('Repositories')} - ${_('add new')}</h2>
         ${h.form(url('repos'))}
         <table>
         	<tr>
         		<td>${_('Name')}</td>
         		<td>${h.text('name',c.new_repo)}</td>
         		<td>${h.text('repo_name',c.new_repo)}</td>
         		<td>${self.get_form_error('repo_name')}</td>
         	</tr>
         	<tr>
         		<td>${_('Description')}</td>
         		<td>${h.textarea('description',cols=23,rows=5)}</td>
         		<td>${self.get_form_error('description')}</td>
         	</tr>
         	<tr>
         		<td>${_('Private')}</td>
         		<td>${h.checkbox('private')}</td>
         		<td>${self.get_form_error('private')}</td>
         	</tr>
         	<tr>
         		<td></td>
         		<td>${h.submit('add','add')}</td>
         	</tr>
         </table>
         ${h.end_form()}
     </div>
 </%def>

pylons_app/templates/admin/repos/repo_edit.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="/base/base.html"/>
 <%def name="title()">
     ${_('Repositories administration')}
 </%def>
 <%def name="breadcrumbs()">
 	${h.link_to(u'Admin',h.url('admin_home'))}
+	 /
 	 ${_('Repos')}
 </%def>
 <%def name="page_nav()">
 	${self.menu('admin')}
 	${self.submenu('repos')}
 </%def>
 <%def name="main()">
 	<div>
         <h2>${_('Repositories')} - ${_('edit')}</h2>
-        ${h.form(url('repo', id=c.new_repo),method='put')}
+        ${h.form(url('repo', id=c.repo_info.repo_name),method='put')}
         <table>
         	<tr>
         		<td>${_('Name')}</td>
         		<td>${h.text('name',c.new_repo)}</td>
         		<td>${h.text('repo_name')}</td>
         		<td>${self.get_form_error('repo_name')}</td>
         	</tr>
         	<tr>
         		<td>${_('Description')}</td>
         		<td>${h.textarea('description',cols=23,rows=5)}</td>
         		<td>${self.get_form_error('description')}</td>
         	</tr>
         	<tr>
         		<td>${_('Private')}</td>
         		<td>${h.checkbox('private')}</td>
         		<td>${self.get_form_error('private')}</td>
         	</tr>
         	<tr>
         		<td>${_('Owner')}</td>
         		<td>${h.text('user')}</td>
         		<td>${self.get_form_error('user')}</td>
         	</tr>
         	<tr>
         		<td></td>
         		<td>${h.submit('update','update')}</td>
         	</tr>
         </table>
         ${h.end_form()}
     </div>
 </%def>

pylons_app/templates/summary/summary.html

➞

Show inline comments

@@ @@ -49,49 +49,49 @@ E.onDOMReady(function(e){ @@
        	%for cnt,archive in enumerate(c.repo_info._get_archives()):
        		 %if cnt >=1:
+       		 |
        		 %endif
        		 ${h.link_to(c.repo_info.name+'.'+archive['type'],
        			h.url('files_archive_home',repo_name=c.repo_info.name,
        			revision='tip',fileformat=archive['extension']),class_="archive_logo")}
 		%endfor
         </dd>
         <dt>${_('feeds')}</dt>
         <dd>
         	${h.link_to(_('RSS'),h.url('rss_feed_home',repo_name=c.repo_info.name),class_='rss_logo')}
 			${h.link_to(_('Atom'),h.url('atom_feed_home',repo_name=c.repo_info.name),class_='atom_logo')}
         </dd>
     </dl>
     <h2>${h.link_to(_('Changes'),h.url('changelog_home',repo_name=c.repo_name))}</h2>
     <table>
 	%for cnt,cs in enumerate(c.repo_changesets):
 		<tr class="parity${cnt%2}">
 			<td>${cs._ctx.date()|n,filters.age}</td>
 			<td>${cs.author|n,filters.person}</td>
 			<td>r${cs.revision}</td>
 			<td>
 				${h.link_to(truncate(cs.message,60),
+				${h.link_to(h.truncate(cs.message,60),
 				h.url('changeset_home',repo_name=c.repo_name,revision=cs._short),
 				title=cs.message)}
 			</td>
 			<td>
 				<span class="logtags">
 					<span class="branchtag">${cs.branch}</span>
 					%for tag in cs.tags:
 						<span class="tagtag">${tag}</span>
 					%endfor
 				</span>
 			</td>
 			<td class="nowrap">
 			${h.link_to(_('changeset'),h.url('changeset_home',repo_name=c.repo_name,revision=cs._short))}
+			|
 			${h.link_to(_('files'),h.url('files_home',repo_name=c.repo_name,revision=cs._short))}
 			</td>
 		</tr>
 	%endfor
     </table>
     <h2>${h.link_to(_('Tags'),h.url('tags_home',repo_name=c.repo_name))}</h2>
     <table>
 		%for cnt,tag in enumerate(c.repo_tags):
 		<tr class="parity${cnt%2}">

0 comments (0 inline, 0 general)