kallithea Changeset - 664a5b8c551a

Changeset - 664a5b8c551a

Parent rev.

Child rev.

[Not reviewed]

default

0 12 0

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

Added application settings, are now customizable from database
fixed all instances of sqlachemy to be removed() after execution.

12 files changed with 106 insertions and 27 deletions:

pylons_app/controllers/admin/admin.py

pylons_app/controllers/admin/settings.py

pylons_app/controllers/admin/users.py

pylons_app/lib/auth.py

pylons_app/lib/db_manage.py

pylons_app/lib/middleware/simplehg.py

pylons_app/lib/utils.py

pylons_app/model/forms.py

pylons_app/model/hg_model.py

pylons_app/templates/admin/settings/settings.html

pylons_app/templates/index.html

pylons_app/templates/login.html

0 comments (0 inline, 0 general)

pylons_app/controllers/admin/admin.py

➞

Show inline comments

 #!/usr/bin/env python
 # encoding: utf-8
 # admin 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 pylons import request, response, session, tmpl_context as c
 from pylons_app.lib.base import BaseController, render
 from pylons_app.model import meta
 from pylons_app.model.db import UserLog
 from webhelpers.paginate import Page
 from pylons_app.lib.auth import LoginRequired, HasPermissionAllDecorator
 log = logging.getLogger(__name__)
 class AdminController(BaseController):
     @LoginRequired()
     def __before__(self):
         super(AdminController, self).__before__()
     @HasPermissionAllDecorator('hg.admin')
     def index(self):
         sa = meta.Session
         users_log = sa.query(UserLog).order_by(UserLog.action_date.desc())
+        users_log = self.sa.query(UserLog).order_by(UserLog.action_date.desc())
         p = int(request.params.get('page', 1))
         c.users_log = Page(users_log, page=p, items_per_page=10)
         c.log_data = render('admin/admin_log.html')
         if request.params.get('partial'):
             return c.log_data
         return render('admin/admin.html')

pylons_app/controllers/admin/settings.py

➞

Show inline comments

 #!/usr/bin/env python
 # encoding: utf-8
 # settings 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 July 14, 2010
 settings controller for pylons
 @author: marcink
 """
 from formencode import htmlfill
 from pylons import request, session, tmpl_context as c, url, app_globals as g
 from pylons import request, session, tmpl_context as c, url, app_globals as g, \
     config
 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, HasPermissionAllDecorator
 from pylons_app.lib.base import BaseController, render
 from pylons_app.lib.utils import repo2db_mapper, invalidate_cache
 from pylons_app.model.db import User, UserLog
 from pylons_app.model.forms import UserForm
 from pylons_app.lib.utils import repo2db_mapper, invalidate_cache, \
     set_hg_app_config
 from pylons_app.model.db import User, UserLog, HgAppSettings
 from pylons_app.model.forms import UserForm, ApplicationSettingsForm
 from pylons_app.model.hg_model import HgModel
 from pylons_app.model.user_model import UserModel
 import formencode
 import logging
 import traceback
 log = logging.getLogger(__name__)
 class SettingsController(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('setting', 'settings', controller='admin/settings',
     #         path_prefix='/admin', name_prefix='admin_')
     @LoginRequired()
     #@HasPermissionAllDecorator('hg.admin')
     def __before__(self):
         c.admin_user = session.get('admin_user')
         c.admin_username = session.get('admin_username')
         super(SettingsController, self).__before__()
     def index(self, format='html'):
         """GET /admin/settings: All items in the collection"""
         # url('admin_settings')
         return render('admin/settings/settings.html')
         hgsettings = self.sa.query(HgAppSettings).scalar()
         defaults = hgsettings.__dict__ if hgsettings else {}
         return htmlfill.render(
             render('admin/settings/settings.html'),
             defaults=defaults,
             encoding="UTF-8",
             force_defaults=False
+        )
     def create(self):
         """POST /admin/settings: Create a new item"""
         # url('admin_settings')
     def new(self, format='html'):
         """GET /admin/settings/new: Form to create a new item"""
         # url('admin_new_setting')
     def update(self, id):
         """PUT /admin/settings/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('admin_setting', id=ID),
         #           method='put')
         # url('admin_setting', id=ID)
         if id == 'mapping':
             rm_obsolete = request.POST.get('destroy', False)
             log.debug('Rescanning directories with destroy=%s', rm_obsolete)
             initial = HgModel.repo_scan(g.paths[0][0], g.paths[0][1], g.baseui)
             repo2db_mapper(initial, rm_obsolete)
             invalidate_cache('cached_repo_list')
         if id == 'global':
             application_form = ApplicationSettingsForm()()
             try:
                 form_result = application_form.to_python(dict(request.POST))
                 title = form_result['app_title']
                 realm = form_result['app_auth_realm']
                 try:
                     hgsettings = self.sa.query(HgAppSettings).get(1)
                     hgsettings.app_auth_realm = realm
                     hgsettings.app_title = title
                     self.sa.add(hgsettings)
                     self.sa.commit()
                     set_hg_app_config(config)
                     h.flash(_('Updated application settings'),
                             category='success')
                 except:
                     log.error(traceback.format_exc())
                     h.flash(_('error occured during chaning application settings'),
                             category='error')
                     self.sa.rollback()
             except formencode.Invalid as errors:
                 c.form_errors = errors.error_dict
                 return htmlfill.render(
                      render('admin/settings/settings.html'),
                     defaults=errors.value,
                     encoding="UTF-8")
         return redirect(url('admin_settings'))
     def delete(self, id):
         """DELETE /admin/settings/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('admin_setting', id=ID),
         #           method='delete')
         # url('admin_setting', id=ID)
     def show(self, id, format='html'):
         """GET /admin/settings/id: Show a specific item"""
         # url('admin_setting', id=ID)
     def edit(self, id, format='html'):
         """GET /admin/settings/id/edit: Form to edit an existing item"""
         # url('admin_edit_setting', id=ID)

pylons_app/controllers/admin/users.py

➞

Show inline comments

@@ @@ -31,97 +31,96 @@ from pylons_app.lib.auth import LoginReq @@
 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, DefaultUserException
 import formencode
 import logging
 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()
     @HasPermissionAllDecorator('hg.admin')
     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')
         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') \
                     % request.POST.get('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()
         _form = UserForm(edit=True)()
         try:
             form_result = _form.to_python(dict(request.POST))
             user_model.update(id, form_result)
             h.flash(_('User updated succesfully'), category='success')
         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:
             user_model.delete(id)

pylons_app/lib/auth.py

➞

Show inline comments

@@ @@ -2,190 +2,198 @@ @@
 # encoding: utf-8
 # authentication and permission libraries
 # 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
 @author: marcink
 """
 from beaker.cache import cache_region
 from functools import wraps
 from pylons import config, session, url, request
 from pylons.controllers.util import abort, redirect
 from pylons_app.lib.utils import get_repo_slug
 from pylons_app.model import meta
 from pylons_app.model.db import User, Repo2Perm, Repository, Permission
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound
 import crypt
 import logging
 log = logging.getLogger(__name__)
 def get_crypt_password(password):
     """
     Cryptographic function used for password hashing
     @param password: password to hash
     """
     return crypt.crypt(password, '6a')
 @cache_region('super_short_term', 'cached_user')
 def get_user_cached(username):
     sa = meta.Session
     try:
     user = sa.query(User).filter(User.username == username).one()
     finally:
         meta.Session.remove()
     return user
 def authfunc(environ, username, password):
     password_crypt = get_crypt_password(password)
     try:
         user = get_user_cached(username)
     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
     """
     def __init__(self):
         self.username = 'None'
         self.user_id = None
         self.is_authenticated = False
         self.is_admin = False
         self.permissions = {}
 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:
     """
     log.info('getting information about all available permissions')
     try:
     sa = meta.Session
     all_perms = sa.query(Permission).all()
     finally:
         meta.Session.remove()
     config['available_permissions'] = [x.permission_name for x in all_perms]
 def set_base_path(config):
     config['base_path'] = config['pylons.app_globals'].base_path
 def fill_perms(user):
     sa = meta.Session
     user.permissions['repositories'] = {}
     #first fetch default permissions
     default_perms = sa.query(Repo2Perm, Repository, Permission)\
         .join((Repository, Repo2Perm.repository == Repository.repo_name))\
         .join((Permission, Repo2Perm.permission_id == Permission.permission_id))\
         .filter(Repo2Perm.user_id == sa.query(User).filter(User.username ==
                                             'default').one().user_id).all()
     if user.is_admin:
         user.permissions['global'] = set(['hg.admin'])
         #admin have all rights full
         for perm in default_perms:
             p = 'repository.admin'
             user.permissions['repositories'][perm.Repo2Perm.repository] = p
     else:
         user.permissions['global'] = set()
         for perm in default_perms:
             if perm.Repository.private:
                 #disable defaults for private repos,
                 p = 'repository.none'
             elif perm.Repository.user_id == user.user_id:
                 #set admin if owner
                 p = 'repository.admin'
             else:
                 p = perm.Permission.permission_name
             user.permissions['repositories'][perm.Repo2Perm.repository] = p
         user_perms = sa.query(Repo2Perm, Permission, Repository)\
             .join((Repository, Repo2Perm.repository == Repository.repo_name))\
             .join((Permission, Repo2Perm.permission_id == Permission.permission_id))\
             .filter(Repo2Perm.user_id == user.user_id).all()
         #overwrite userpermissions with defaults
         for perm in user_perms:
             #set write if owner
             if perm.Repository.user_id == user.user_id:
                 p = 'repository.write'
             else:
                 p = perm.Permission.permission_name
             user.permissions['repositories'][perm.Repo2Perm.repository] = p
     meta.Session.remove()
     return user
 def get_user(session):
     """
     Gets user from session, and wraps permissions into user
     @param session:
     """
     user = session.get('hg_app_user', AuthUser())
     if user.is_authenticated:
         user = fill_perms(user)
     session['hg_app_user'] = user
     session.save()
     return user
 #===============================================================================
 # CHECK DECORATORS
 #===============================================================================
 class LoginRequired(object):
     """
     Must be logged in to execute this function else redirect to login page
     """
     def __call__(self, func):
         @wraps(func)
         def _wrapper(*fargs, **fkwargs):
             user = session.get('hg_app_user', AuthUser())
             log.debug('Checking login required for user:%s', user.username)
             if user.is_authenticated:
                 log.debug('user %s is authenticated', user.username)
                 func(*fargs)
             else:
                 log.warn('user %s not authenticated', user.username)
                 log.debug('redirecting to login page')
                 return redirect(url('login_home'))
         return _wrapper
 class PermsDecorator(object):
     """
     Base class for decorators
     """
     def __init__(self, *required_perms):
         available_perms = config['available_permissions']
         for perm in required_perms:
             if perm not in available_perms:

pylons_app/lib/db_manage.py

➞

Show inline comments

 #!/usr/bin/env python
 # encoding: utf-8
 # database managment for hg app
 # 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 10, 2010
 database managment and creation for hg app
 @author: marcink
 """
 from os.path import dirname as dn, join as jn
 import os
 import sys
 import uuid
 ROOT = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(ROOT)
 from pylons_app.lib.auth import get_crypt_password
 from pylons_app.model import init_model
 from pylons_app.model.db import User, Permission, HgAppUi, HgAppSettings
-from pylons_app.model.meta import Session, Base
+from pylons_app.model import meta
 from sqlalchemy.engine import create_engine
 import logging
 log = logging.getLogger('db manage')
 log.setLevel(logging.DEBUG)
 console_handler = logging.StreamHandler()
 console_handler.setFormatter(logging.Formatter("%(asctime)s.%(msecs)03d"
                                   " %(levelname)-5.5s [%(name)s] %(message)s"))
 log.addHandler(console_handler)
 class DbManage(object):
     def __init__(self, log_sql):
         self.dbname = 'hg_app.db'
         dburi = 'sqlite:////%s' % jn(ROOT, self.dbname)
         engine = create_engine(dburi, echo=log_sql)
         init_model(engine)
-        self.sa = Session()
+        self.sa = meta.Session
         self.db_exists = False
     def check_for_db(self, override):
         log.info('checking for exisiting db')
         if os.path.isfile(jn(ROOT, self.dbname)):
             self.db_exists = True
             log.info('database exisist')
             if not override:
                 raise Exception('database already exists')
     def create_tables(self, override=False):
         """
         Create a auth database
         """
         self.check_for_db(override)
         if override:
             log.info("database exisist and it's going to be destroyed")
             if self.db_exists:
                 os.remove(jn(ROOT, self.dbname))
         Base.metadata.create_all(checkfirst=override)
+        meta.Base.metadata.create_all(checkfirst=override)
         log.info('Created tables for %s', self.dbname)
     def admin_prompt(self):
         import getpass
         username = raw_input('Specify admin username:')
         password = getpass.getpass('Specify admin password:')
         self.create_user(username, password, True)
     def config_prompt(self):
         log.info('Setting up repositories config')
         path = raw_input('Specify valid full path to your repositories'
                         ' you can change this later application settings:')
         if not os.path.isdir(path):
             log.error('You entered wrong path')
             sys.exit()
         hooks = HgAppUi()
         hooks.ui_section = 'hooks'
         hooks.ui_key = 'changegroup'
         hooks.ui_value = 'hg update >&2'
         web1 = HgAppUi()
         web1.ui_section = 'web'
         web1.ui_key = 'push_ssl'
         web1.ui_value = 'false'
         web2 = HgAppUi()
         web2.ui_section = 'web'
         web2.ui_key = 'allow_archive'
         web2.ui_value = 'gz zip bz2'
         web3 = HgAppUi()
         web3.ui_section = 'web'
         web3.ui_key = 'allow_push'
         web3.ui_value = '*'
         web4 = HgAppUi()
         web4.ui_section = 'web'
         web4.ui_key = 'baseurl'
         web4.ui_value = '/'
         paths = HgAppUi()
         paths.ui_section = 'paths'
         paths.ui_key = '/'
         paths.ui_value = os.path.join(path, '*')

pylons_app/lib/middleware/simplehg.py

➞

Show inline comments

@@ @@ -7,108 +7,108 @@ @@
 # 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 2010-04-28
 @author: marcink
 SimpleHG middleware for handling mercurial protocol request (push/clone etc.)
 It's implemented with basic auth function
 """
 from datetime import datetime
 from itertools import chain
 from mercurial.error import RepoError
 from mercurial.hgweb import hgweb
 from mercurial.hgweb.request import wsgiapplication
 from paste.auth.basic import AuthBasicAuthenticator
 from paste.httpheaders import REMOTE_USER, AUTH_TYPE
 from pylons_app.lib.auth import authfunc, HasPermissionAnyMiddleware, \
     get_user_cached
 from pylons_app.lib.utils import is_mercurial, make_ui, invalidate_cache, \
     check_repo_fast
 from pylons_app.model import meta
 from pylons_app.model.db import UserLog, User
 from webob.exc import HTTPNotFound, HTTPForbidden, HTTPInternalServerError
 import logging
 import os
 import pylons_app.lib.helpers as h
 import traceback
 log = logging.getLogger(__name__)
 class SimpleHg(object):
     def __init__(self, application, config):
         self.application = application
         self.config = config
         #authenticate this mercurial request using
         realm = self.config['hg_app_auth_realm']
         self.authenticate = AuthBasicAuthenticator(realm, authfunc)
         self.authenticate = AuthBasicAuthenticator('', authfunc)
     def __call__(self, environ, start_response):
         if not is_mercurial(environ):
             return self.application(environ, start_response)
         #===================================================================
         # AUTHENTICATE THIS MERCURIAL REQUEST
         #===================================================================
         username = REMOTE_USER(environ)
         if not username:
             self.authenticate.realm = self.config['hg_app_auth_realm']
             result = self.authenticate(environ)
             if isinstance(result, str):
                 AUTH_TYPE.update(environ, 'basic')
                 REMOTE_USER.update(environ, result)
             else:
                 return result.wsgi_application(environ, start_response)
         try:
             repo_name = '/'.join(environ['PATH_INFO'].split('/')[1:])
         except:
             log.error(traceback.format_exc())
             return HTTPInternalServerError()(environ, start_response)
         #===================================================================
         # CHECK PERMISSIONS FOR THIS REQUEST
         #===================================================================
         action = self.__get_action(environ)
         if action:
             username = self.__get_environ_user(environ)
             try:
                 user = self.__get_user(username)
             except:
                 log.error(traceback.format_exc())
                 return HTTPInternalServerError()(environ, start_response)
             #check permissions for this repository
             if action == 'pull':
                 if not HasPermissionAnyMiddleware('repository.read',
                                                   'repository.write',
                                                   'repository.admin')\
                                                     (user, repo_name):
                     return HTTPForbidden()(environ, start_response)
             if action == 'push':
                 if not HasPermissionAnyMiddleware('repository.write',
                                                   'repository.admin')\
                                                     (user, repo_name):
                     return HTTPForbidden()(environ, start_response)
             #log action
             proxy_key = 'HTTP_X_REAL_IP'
             def_key = 'REMOTE_ADDR'
             ipaddr = environ.get(proxy_key, environ.get(def_key, '0.0.0.0'))
             self.__log_user_action(user, action, repo_name, ipaddr)
         #===================================================================
         # MERCURIAL REQUEST HANDLING
         #===================================================================
         environ['PATH_INFO'] = '/'#since we wrap into hgweb, reset the path
         self.baseui = make_ui('db')
@@ @@ -163,69 +163,71 @@ class SimpleHg(object): @@
     def __get_user(self, username):
         return get_user_cached(username)
     def __get_size(self, repo_path, content_size):
         size = int(content_size)
         for path, dirs, files in os.walk(repo_path):
             if path.find('.hg') == -1:
                 for f in files:
                     size += os.path.getsize(os.path.join(path, f))
         return size
         return h.format_byte_size(size)
     def __get_action(self, environ):
         """
         Maps mercurial request commands into a pull or push command.
         @param environ:
         """
         mapping = {'changegroup': 'pull',
                    'changegroupsubset': 'pull',
                    'stream_out': 'pull',
                    'listkeys': 'pull',
                    'unbundle': 'push',
                    'pushkey': 'push', }
         for qry in environ['QUERY_STRING'].split('&'):
             if qry.startswith('cmd'):
                 cmd = qry.split('=')[-1]
                 if mapping.has_key(cmd):
                     return mapping[cmd]
     def __log_user_action(self, user, action, repo, ipaddr):
         sa = meta.Session
         try:
             user_log = UserLog()
             user_log.user_id = user.user_id
             user_log.action = action
             user_log.repository = repo.replace('/', '')
             user_log.action_date = datetime.now()
             user_log.user_ip = ipaddr
             sa.add(user_log)
             sa.commit()
             log.info('Adding user %s, action %s on %s',
                                             user.username, action, repo)
         except Exception as e:
             sa.rollback()
             log.error('could not log user action:%s', str(e))
         finally:
             meta.Session.remove()
     def __invalidate_cache(self, repo_name):
         """we know that some change was made to repositories and we should
         invalidate the cache to see the changes right away but only for
         push requests"""
         invalidate_cache('cached_repo_list')
         invalidate_cache('full_changelog', repo_name)
     def __load_web_settings(self, hgserve):
         #set the global ui for hgserve
         hgserve.repo.ui = self.baseui
         hgrc = os.path.join(self.repo_path, '.hg', 'hgrc')
         repoui = make_ui('file', hgrc, False)
         if repoui:
             #set the repository based config
             hgserve.repo.ui = repoui
         return hgserve

pylons_app/lib/utils.py

➞

Show inline comments

 #!/usr/bin/env python
 # encoding: utf-8
 # Utilities for hg app
 # 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.
 from beaker.cache import cache_region
 """
 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, HgAppUi, HgAppSettings
-from pylons_app.model.meta import Session
+from pylons_app.model import meta
 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])
 def check_repo_fast(repo_name, base_path):
     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)
     try:
         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)
         return False
     except RepoError:
         #it means that there is no valid repo there...
         log.info('%s repo is free for creation', repo_name)
         return True
 @cache_region('super_short_term', 'cached_hg_ui')
 def get_hg_ui_cached():
     sa = Session()
     return sa.query(HgAppUi).all()
     try:
         sa = meta.Session
         ret = sa.query(HgAppUi).all()
     finally:
         meta.Session.remove()
     return ret
 def get_hg_settings():
     sa = Session()
     try:
         sa = meta.Session
     ret = sa.query(HgAppSettings).scalar()
     finally:
         meta.Session.remove()
     if not ret:
         raise Exception('Could not get application settings !')
     return ret
 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'
     """
     #propagated from mercurial documentation
     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', ]
     baseui = ui.ui()
     if read_from == 'file':
         if not os.path.isfile(path):
             log.warning('Unable to read config file %s' % path)
             return False
         cfg = config.config()
         cfg.read(path)
         for section in sections:
             for k, v in cfg.items(section):
                 baseui.setconfig(section, k, v)
         if checkpaths:check_repo_dir(cfg.items('paths'))
     elif read_from == 'db':
         hg_ui = get_hg_ui_cached()
         for ui_ in hg_ui:
             baseui.setconfig(ui_.ui_section, ui_.ui_key, ui_.ui_value)
     return baseui
@@ @@ -138,73 +147,75 @@ def make_ui(read_from='file', path=None, @@
 def set_hg_app_config(config):
     hgsettings = get_hg_settings()
     config['hg_app_auth_realm'] = hgsettings.app_auth_realm
     config['hg_app_name'] = hgsettings.app_title
 def invalidate_cache(name, *args):
     """Invalidates given name cache"""
     from beaker.cache import region_invalidate
     log.info('INVALIDATING CACHE FOR %s', name)
     """propagate our arguments to make sure invalidation works. First
     argument has to be the name of cached func name give to cache decorator
     without that the invalidation would not work"""
     tmp = [name]
     tmp.extend(args)
     args = tuple(tmp)
     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
     message = ''
     @LazyProperty
     def raw_id(self):
         """
         Returns raw string identifing this changeset, useful for web
         representation.
         """
         return '0' * 12
 def repo2db_mapper(initial_repo_list, remove_obsolete=False):
     """
     maps all found repositories into db
     """
     from pylons_app.model.repo_model import RepoModel
-    sa = Session()
+    sa = meta.Session
     user = sa.query(User).filter(User.admin == True).first()
     rm = RepoModel()
     for name, repo in initial_repo_list.items():
         if not sa.query(Repository).get(name):
             log.info('repository %s not found creating default', name)
             form_data = {
                          'repo_name':name,
                          'description':repo.description if repo.description != 'unknown' else \
                                         'auto description for %s' % name,
                          'private':False
+                         }
             rm.create(form_data, user, just_db=True)
     if remove_obsolete:
         #remove from database those repositories that are not in the filesystem
         for repo in sa.query(Repository).all():
             if repo.repo_name not in initial_repo_list.keys():
                 sa.delete(repo)
                 sa.commit()
     meta.Session.remove()

pylons_app/model/forms.py

➞

Show inline comments

@@ @@ -222,50 +222,59 @@ class LoginForm(formencode.Schema): @@
     #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)
             admin = StringBoolean(if_missing=False)
         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 = False
         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)
         chained_validators = [ValidPerms]
     return _RepoForm
 def RepoSettingsForm(edit=False):
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         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)
         chained_validators = [ValidPerms, ValidSettings]
     return _RepoForm
 def ApplicationSettingsForm():
     class _ApplicationSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         app_title = UnicodeString(strip=True, min=3, not_empty=True)
         app_auth_realm = UnicodeString(strip=True, min=3, not_empty=True)
     return _ApplicationSettingsForm

pylons_app/model/hg_model.py

➞

Show inline comments

 #!/usr/bin/env python
 # encoding: utf-8
 # Model for hg app
 # 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 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 vcs.exceptions import RepositoryError, VCSError
-from pylons_app.model.meta import Session
+from pylons_app.model import meta
 from pylons_app.model.db import Repository
 from sqlalchemy.orm import joinedload
 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
     """
     log.info('getting all repositories list')
     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
         """
     @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()
+        sa = meta.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].dbrepo = dbrepo
                         repos_list[name].description = dbrepo.description
                         repos_list[name].contact = dbrepo.user.full_contact
             except OSError:
                 continue
         meta.Session.remove()
         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())
             tmp_d['last_msg'] = tip.message
             tmp_d['repo'] = repo
             yield tmp_d
     def get_repo(self, repo_name):
         return _get_repos_cached()[repo_name]

pylons_app/templates/admin/settings/settings.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="/base/base.html"/>
 <%def name="title()">
     ${_('Settings administration')}
 </%def>
 <%def name="breadcrumbs()">
 	${h.link_to(u'Admin',h.url('admin_home'))}
+	 /
 	 ${_('Settings')}
 </%def>
 <%def name="page_nav()">
 	${self.menu('admin')}
 	${self.submenu('settings')}
 </%def>
 <%def name="main()">
 	<div>
 	<h2>${_('Settings administration')}</h2>
      ${h.form(url('admin_setting', id='mapping'),method='put')}
        <table class="table_disp">
            <tr class="header">
-            <td colspan="2">${_('Remap andv rescan repositories')}</td>
+            <td colspan="2">${_('Remap and rescan repositories')}</td>
            </tr>
            <tr align="right">
             <td><span class="tooltip" tooltip_title="${h.tooltip(_('In case a repository was deleted from filesystem and there are leftovers in the database check this option to scan obsolete data in database and remove it.'))}">
             ${_('destroy old data')}</span> ${h.checkbox('destroy',True)}</td>
             <td>${h.submit('rescan','rescan repositories')}</td>
        </table>
      ${h.end_form()}
      <br/>
      ${h.form(url('admin_setting', id='global'),method='put')}
 	   <table class="table_disp">
 	       <tr class="header">
 	           <td colspan="3">${_('Global application settings')}</td>
 	       </tr>
 	       <tr>
 	           <td>${_('Application name')}</td>
 	           <td>${h.text('app_title')}</td>
+	           <td>${h.text('app_title',size=30)}${self.get_form_error('app_title')}</td>
 	       </tr>
            <tr>
                <td>${_('Realm text')}</td>
                <td>${h.text('app_auth_realm')}</td>
+               <td>${h.text('app_auth_realm',size=30)}${self.get_form_error('app_auth_realm')}</td>
            </tr>
            <tr>
 	           <td></td>
 	           <td>${h.submit('save','save settings')}</td>
            </tr>
 	   </table>
      ${h.end_form()}
     </div>
 </%def>

pylons_app/templates/index.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="base/base.html"/>
 <%def name="title()">
-    ${c.repos_prefix} Mercurial Repositories
     ${c.repos_prefix}
 </%def>
 <%def name="breadcrumbs()">
-	${c.repos_prefix} Mercurial Repositories
 	${c.repos_prefix}
 </%def>
 <%def name="page_nav()">
 	${self.menu('home')}
 </%def>
 <%def name="main()">
 	<%def name="get_sort(name)">
 		<%name_slug = name.lower().replace(' ','_') %>
 		%if name_slug == c.cs_slug:
 			<span style="font-weight: bold;text-decoration: underline;">${name}</span>
 		%else:
 			<span style="font-weight: bold">${name}</span>
 		%endif
 		<a style="color:#FFF" href="?sort=${name_slug}">&darr;</a>
 		<a style="color:#FFF" href="?sort=-${name_slug}">&uarr;</a>
 	</%def>
 	<table class="table_disp">
 	  <tr class="header">
 	    <td>${get_sort(_('Name'))}</td>
 	    <td>${get_sort(_('Description'))}</td>
 	    <td>${get_sort(_('Last change'))}</td>
 	    <td>${get_sort(_('Tip'))}</td>
 	    <td>${get_sort(_('Contact'))}</td>
 	    <td>${_('RSS')}</td>
 	    <td>${_('Atom')}</td>
 	  </tr>
 	%for cnt,repo in enumerate(c.repos_list):
 		%if h.HasRepoPermissionAny('repository.write','repository.read','repository.admin')(repo['name'],'main page check'):
  		<tr class="parity${cnt%2}">
 		    <td>
  			 %if repo['repo'].dbrepo.private:
 				<img alt="${_('private')}" src="/images/icons/lock.png">
 			 %else:
 			 	<img alt="${_('public')}" src="/images/icons/lock_open.png">
 			 %endif
 		    ${h.link_to(repo['name'],
 		    	h.url('summary_home',repo_name=repo['name']))}</td>
 		    <td title="${repo['description']}">${h.truncate(repo['description'],60)}</td>
 	        <td>${h.age(repo['last_change'])}</td>
 	        <td>${h.link_to_if(repo['rev']>=0,'r%s:%s' % (repo['rev'],repo['tip']),
 		        h.url('changeset_home',repo_name=repo['name'],revision=repo['tip']),
 	        	class_="tooltip",
 		    	tooltip_title=h.tooltip(repo['last_msg']))}</td>
 	        <td title="${repo['contact']}">${h.person(repo['contact'])}</td>
 			<td>
 				<a title="${_('Subscribe to %s rss feed')%repo['name']}" class="rss_logo"  href="${h.url('rss_feed_home',repo_name=repo['name'])}"></a>
 			</td>
 			<td>

pylons_app/templates/login.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 <%inherit file="base/base.html"/>
 <%def name="title()">
-    ${c.repos_prefix} Mercurial Repositories
     ${c.repos_prefix}
 </%def>
 <%def name="breadcrumbs()">
-	${c.repos_prefix} Mercurial Repositories
 	${c.repos_prefix}
 </%def>
 <%def name="page_nav()">
 	&nbsp;
 	</div>
 </%def>
 <%def name="main()">
         <div>
         <br />
         <h2>${_('Login to hg app')}</h2>
         ${h.form(h.url.current())}
         <table>
             <tr>
                 <td>${_('Username')}</td>
                 <td>${h.text('username')}</td>
                 <td>${self.get_form_error('username')}</td>
             </tr>
             <tr>
                 <td>${_('Password')}</td>
                 <td>${h.password('password')}</td>
                 <td>${self.get_form_error('password')}</td>
             </tr>
             <tr>
                 <td></td>
                 <td>${h.submit('login','login')}</td>
             </tr>
         </table>
         ${h.end_form()}
         </div>
 </%def>

0 comments (0 inline, 0 general)