kallithea Changeset - a0116e944da1

Changeset - a0116e944da1

Parent rev.

Child rev.

[Not reviewed]

default

0 7 0

Marcin Kuzminski - 15 years ago 2010-05-27 21:31:30
marcin@python-works.com

changed naming convention for db modules.

7 files changed with 30 insertions and 27 deletions:

pylons_app/controllers/admin.py

pylons_app/controllers/users.py

pylons_app/lib/auth.py

pylons_app/lib/db_manage.py

pylons_app/lib/middleware/simplehg.py

pylons_app/model/db.py

pylons_app/model/forms.py

0 comments (0 inline, 0 general)

pylons_app/controllers/admin.py

➞

Show inline comments

 import logging
 import os
 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.base import BaseController, render
 from pylons_app.model import meta
-from pylons_app.model.db import UserLogs
+from pylons_app.model.db import UserLog
 from webhelpers.paginate import Page
 from pylons_app.lib.auth import LoginRequired
 log = logging.getLogger(__name__)
 class AdminController(BaseController):
     @LoginRequired()
     def __before__(self):
         user = session['hg_app_user']
         c.admin_user = user.is_admin
         c.admin_username = user.username
         super(AdminController, self).__before__()
     def index(self):
         sa = meta.Session
         users_log = sa.query(UserLogs)\
             .order_by(UserLogs.action_date.desc())
         users_log = 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/users.py

➞

Show inline comments

 import logging
 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.base import BaseController, render
 from formencode import htmlfill
-from pylons_app.model.db import Users, UserLogs
+from pylons_app.model.db import User, UserLog
 import crypt
 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')
     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(Users).all()
+        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')
         params = dict(request.params)
         try:
-            new_user = Users()
+            new_user = User()
             new_user.active = params.get('active', False)
             new_user.username = params.get('username')
             new_user.password = crypt.crypt(params.get('password'), '6a')
             new_user.admin = False
             self.sa.add(new_user)
             self.sa.commit()
         except:
             self.sa.rollback()
             raise
         return redirect(url('users'))
@@ @@ -54,57 +54,57 @@ class UsersController(BaseController): @@
     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)
         params = dict(request.params)
         try:
-            new_user = self.sa.query(Users).get(id)
+            new_user = self.sa.query(User).get(id)
             new_user.active = params.get('active', False)
             new_user.username = params.get('username')
             if params.get('new_password'):
                 new_user.password = crypt.crypt(params.get('new_password'), '6a')
             self.sa.add(new_user)
             self.sa.commit()
         except:
             self.sa.rollback()
             raise
         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)
         try:
-            self.sa.delete(self.sa.query(Users).get(id))
+            self.sa.delete(self.sa.query(User).get(id))
             self.sa.commit()
         except:
             self.sa.rollback()
             raise
         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(Users).get(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

 from functools import wraps
 from pylons import session, url
 from pylons.controllers.util import abort, redirect
 from pylons_app.model import meta
-from pylons_app.model.db import Users
+from pylons_app.model.db import User
 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')
 def authfunc(environ, username, password):
     sa = meta.Session
     password_crypt = get_crypt_password(password)
     try:
-        user = sa.query(Users).filter(Users.username == username).one()
+        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)

pylons_app/lib/db_manage.py

➞

Show inline comments

 import logging
 from os.path import dirname as dn
 from os.path import join as jn
 from sqlalchemy.engine import create_engine
 import os
 import sys
 ROOT = dn(dn(dn(os.path.realpath(__file__))))
 sys.path.append(ROOT)
-from pylons_app.model.db import Users
+from pylons_app.model.db import User
 from pylons_app.model.meta import Session, Base
 from pylons_app.lib.auth import get_crypt_password
 from pylons_app.model import init_model
 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' % os.path.join(ROOT, self.dbname)
+        dburi = 'sqlite:////%s' % jn(ROOT, self.dbname)
         engine = create_engine(dburi, echo=log_sql)
         init_model(engine)
         self.sa = Session()
     def check_for_db(self, override):
         log.info('checking for exisiting db')
-        if os.path.isfile(os.path.join(ROOT, self.dbname)):
+        if os.path.isfile(jn(ROOT, self.dbname)):
             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")
             os.remove(jn(ROOT, self.dbname))
         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 create_user(self, username, password, admin=False):
         log.info('creating administrator user %s', username)
-        new_user = Users()
+        new_user = User()
         new_user.username = username
         new_user.password = get_crypt_password(password)
         new_user.admin = admin
         new_user.active = True
         try:
             self.sa.add(new_user)
             self.sa.commit()
         except:
             self.sa.rollback()
             raise

pylons_app/lib/middleware/simplehg.py

➞

Show inline comments

@@ @@ -9,25 +9,25 @@ 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 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
 from pylons_app.lib.utils import is_mercurial, make_ui, invalidate_cache
 from pylons_app.model import meta
-from pylons_app.model.db import UserLogs, Users
+from pylons_app.model.db import UserLog, User
 from webob.exc import HTTPNotFound
 import logging
 import os
 log = logging.getLogger(__name__)
 class SimpleHg(object):
     def __init__(self, application, config):
         self.application = application
         self.config = config
         #authenticate this mercurial request using
         realm = '%s %s' % (config['hg_app_name'], 'mercurial repository')
@@ @@ -93,27 +93,26 @@ class SimpleHg(object): @@
             'unbundle': 'push',
             'stream_out': 'pull',
+        }
         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, username, action, repo):
         sa = meta.Session
         try:
             user = sa.query(Users)\
                     .filter(Users.username == username).one()
             user_log = UserLogs()
             user = sa.query(User).filter(User.username == username).one()
             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()
             sa.add(user_log)
             sa.commit()
             log.info('Adding user %s, action %s on %s',
                                             username, action, repo)
         except Exception as e:
             sa.rollback()
             log.error('could not log user action:%s', str(e))

pylons_app/model/db.py

➞

Show inline comments

 from pylons_app.model.meta import Base
 from sqlalchemy.orm import relation, backref
 from sqlalchemy import *
-class Users(Base):
+class User(Base):
     __tablename__ = 'users'
     __table_args__ = {'useexisting':True}
     user_id = Column("user_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=1)
     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('UserLogs')
+    user_log = relation('UserLog')
     def __repr__(self):
         return "<User('%s:%s')>" % (self.user_id, self.username)
-class UserLogs(Base):
+class UserLog(Base):
     __tablename__ = 'user_logs'
     __table_args__ = {'useexisting':True}
-    user_log_id = Column("id", INTEGER(), nullable=False, unique=True, default=None, primary_key=1)
+    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)
     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('Users')
+    user = relation('User')
 class Repository(Base):
     __tablename__ = 'repositories'
     repo_id = Column("repo_id", INTEGER(), nullable=False, unique=True, default=None, primary_key=1)
-class Permissions(Base):
+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)
     def __repr__(self):
         return "<Permission('%s:%s')>" % (self.permission_id, self.permission_name)

pylons_app/model/forms.py

➞

Show inline comments

@@ @@ -15,25 +15,25 @@ ignore_key_missing False If Tru @@
 <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.validators import UnicodeString, OneOf, Int, Number, Regex
 from pylons import session
 from pylons.i18n.translation import _
 from pylons_app.lib.auth import get_crypt_password
 from pylons_app.model import meta
-from pylons_app.model.db import Users
+from pylons_app.model.db import User
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound
 from webhelpers.pylonslib.secure_form import authentication_token
 import formencode
 import logging
 log = logging.getLogger(__name__)
 #this is needed to translate the messages using _() in validators
 class State_obj(object):
     _ = staticmethod(_)
@@ @@ -57,25 +57,25 @@ class ValidAuth(formencode.validators.Fa @@
+            }
     #error mapping
     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(Users).filter(Users.username == username).one()
+            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 == 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
                     session['hg_app_user'] = auth_user

0 comments (0 inline, 0 general)