rewritten db manage script to use sqlalchemy. Fixed sqlalchemy models to more generic.

pylons_app/lib/db_manage.py

@@ -1,72 +1,63 b''
 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()  
 
 

pylons_app/model/db.py

@@ -1,30 +1,36 b''
-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'
     __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)
-    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)