Mercurial > kallithea

changeset 226:c6526b7531e9

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
rewritten db manage script to use sqlalchemy. Fixed sqlalchemy models to more generic.
author Marcin Kuzminski <marcin@python-works.com>
date Wed, 26 May 2010 23:55:20 +0200
parents 710e7a75bb6b
children 351013049742
files pylons_app/lib/db_manage.py pylons_app/model/db.py
diffstat 2 files changed, 69 insertions(+), 72 deletions(-) [+]
line wrap: on
line diff
--- a/pylons_app/lib/db_manage.py	Wed May 26 00:20:03 2010 +0200
+++ b/pylons_app/lib/db_manage.py	Wed May 26 23:55:20 2010 +0200
@@ -1,72 +1,63 @@
 import logging
-import sqlite3 
-
+from os.path import dirname as dn
+from sqlalchemy.engine import create_engine
 import os
-import crypt
-from os.path import dirname as dn
-ROOT = dn(dn(dn(os.path.realpath(__file__))))
-logging.basicConfig(level=logging.DEBUG)
+from pylons_app.model.db import Users
+from pylons_app.model.meta import Session
 
-def get_sqlite_conn_cur():
-    conn = sqlite3.connect(os.path.join(ROOT, 'hg_app.db'))
-    cur = conn.cursor()
-    return conn, cur
+from pylons_app.lib.auth import get_crypt_password
+from pylons_app.model import init_model
 
-def check_for_db(override):
-    if not override:
-        if os.path.isfile(os.path.join(ROOT, 'hg_app.db')):
-            raise Exception('database already exists')
+ROOT = dn(dn(dn(os.path.realpath(__file__))))
+logging.basicConfig(level=logging.DEBUG, format='%(asctime)s.%(msecs)03d %(levelname)-5.5s [%(name)s] %(message)s')
+from pylons_app.model.meta import Base
 
-def create_tables(override=False):
-    """
-    Create a auth database
-    """
-    check_for_db(override)
-    conn, cur = get_sqlite_conn_cur()
-    try:
-        logging.info('creating table %s', 'users')
-        cur.execute("""DROP TABLE IF EXISTS users """)
-        cur.execute("""CREATE TABLE users
-                        (user_id INTEGER PRIMARY KEY AUTOINCREMENT, 
-                         username TEXT, 
-                         password TEXT,
-                         active INTEGER,
-                         admin INTEGER)""")
-        logging.info('creating table %s', 'user_logs')
-        cur.execute("""DROP TABLE IF EXISTS user_logs """)
-        cur.execute("""CREATE TABLE user_logs
-                        (id INTEGER PRIMARY KEY AUTOINCREMENT,
-                            user_id INTEGER,
-                            repository TEXT,
-                            action TEXT, 
-                            action_date DATETIME)""")
-        conn.commit()
-    except:
-        conn.rollback()
-        raise
+class DbManage(object):
+    def __init__(self):
+        dburi = 'sqlite:////%s' % os.path.join(ROOT, 'hg_app.db')
+        engine = create_engine(dburi) 
+        init_model(engine)
+        self.sa = Session()
+    
+    def check_for_db(self, override):
+        if not override:
+            if os.path.isfile(os.path.join(ROOT, 'hg_app.db')):
+                raise Exception('database already exists')
+    
+    def create_tables(self, override=False):
+        """
+        Create a auth database
+        """
+        self.check_for_db(override)
+                
+        Base.metadata.create_all(checkfirst=override)
+        logging.info('Created tables')
     
-    cur.close()
-
-def admin_prompt():
-    import getpass
-    username = raw_input('give username:')
-    password = getpass.getpass('Specify admin password:')
-    create_user(username, password, True)
-    
-def create_user(username, password, admin=False):
-    conn, cur = get_sqlite_conn_cur()    
-    password_crypt = crypt.crypt(password, '6a')
-    logging.info('creating user %s', username)
-    try:
-        cur.execute("""INSERT INTO users values (?,?,?,?,?) """,
-                    (None, username, password_crypt, 1, admin))     
-        conn.commit()
-    except:
-        conn.rollback()
-        raise
+    def admin_prompt(self):
+        import getpass
+        username = raw_input('give admin username:')
+        password = getpass.getpass('Specify admin password:')
+        self.create_user(username, password, True)
+        
+    def create_user(self, username, password, admin=False):
+        logging.info('creating user %s', username)
+        
+        new_user = Users()
+        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
     
 if __name__ == '__main__':
-    create_tables(True)
-    admin_prompt()  
+    dbmanage = DbManage()
+    dbmanage.create_tables(override=True)
+    dbmanage.admin_prompt()  
 
 
--- a/pylons_app/model/db.py	Wed May 26 00:20:03 2010 +0200
+++ b/pylons_app/model/db.py	Wed May 26 23:55:20 2010 +0200
@@ -1,13 +1,6 @@
-import sqlalchemy
 from pylons_app.model.meta import Base
-from sqlalchemy import ForeignKey, Column
 from sqlalchemy.orm import relation, backref
-
-if sqlalchemy.__version__ == '0.6.0':
-    from sqlalchemy.dialects.sqlite import *
-else:
-    from sqlalchemy.databases.sqlite import SLBoolean as BOOLEAN, \
-    SLInteger as INTEGER, SLText as TEXT, SLDateTime as DATETIME
+from sqlalchemy import *
 
 class Users(Base): 
     __tablename__ = 'users'
@@ -17,14 +10,27 @@
     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)
-    action_log = relation('UserLogs')
+    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')
       
 class UserLogs(Base): 
     __tablename__ = 'user_logs'
     __table_args__ = {'useexisting':True}
-    id = Column("id", INTEGER(), nullable=False, unique=True, default=None, primary_key=1)
+    user_log_id = Column("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')
+
+
+class Permissions(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)