db_manage.py
74 lines
| 2.3 KiB
| text/x-python
|
PythonLexer
|
r59 | import logging | ||
r226 | from os.path import dirname as dn | |||
from sqlalchemy.engine import create_engine | ||||
|
r59 | import os | ||
r228 | import sys | |||
ROOT = dn(dn(dn(os.path.realpath(__file__)))) | ||||
sys.path.append(ROOT) | ||||
r226 | from pylons_app.model.db import Users | |||
r229 | from pylons_app.model.meta import Session, Base | |||
|
r59 | |||
r226 | from pylons_app.lib.auth import get_crypt_password | |||
from pylons_app.model import init_model | ||||
|
r59 | |||
r229 | 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) | ||||
|
r66 | |||
r226 | class DbManage(object): | |||
r229 | def __init__(self, log_sql): | |||
self.dbname = 'hg_app.db' | ||||
dburi = 'sqlite:////%s' % os.path.join(ROOT, self.dbname) | ||||
engine = create_engine(dburi, echo=log_sql) | ||||
r226 | init_model(engine) | |||
self.sa = Session() | ||||
def check_for_db(self, override): | ||||
r229 | log.info('checking for exisiting db') | |||
if os.path.isfile(os.path.join(ROOT, self.dbname)): | ||||
log.info('database exisist') | ||||
if not override: | ||||
r226 | raise Exception('database already exists') | |||
r229 | ||||
r226 | def create_tables(self, override=False): | |||
""" | ||||
Create a auth database | ||||
""" | ||||
self.check_for_db(override) | ||||
r229 | if override: | |||
log.info("database exisist and it's going to be destroyed") | ||||
r226 | Base.metadata.create_all(checkfirst=override) | |||
r229 | log.info('Created tables for %s', self.dbname) | |||
|
r59 | |||
r226 | def admin_prompt(self): | |||
import getpass | ||||
r229 | username = raw_input('Specify admin username:') | |||
r226 | password = getpass.getpass('Specify admin password:') | |||
self.create_user(username, password, True) | ||||
def create_user(self, username, password, admin=False): | ||||
r229 | log.info('creating administrator user %s', username) | |||
r226 | ||||
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 | ||||
|
r66 | |||
if __name__ == '__main__': | ||||
r229 | dbmanage = DbManage(log_sql=True) | |||
r226 | dbmanage.create_tables(override=True) | |||
dbmanage.admin_prompt() | ||||
|
r66 | |||