kallithea Changeset - c6526b7531e9

Changeset - c6526b7531e9

Parent rev.

Child rev.

[Not reviewed]

default

0 2 0

Marcin Kuzminski - 15 years ago 2010-05-26 23:55:20
marcin@python-works.com

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

2 files changed with 69 insertions and 72 deletions:

pylons_app/lib/db_manage.py

pylons_app/model/db.py

0 comments (0 inline, 0 general)

pylons_app/lib/db_manage.py

➞

Show inline comments

 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

➞

Show inline comments

 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)

0 comments (0 inline, 0 general)