-
Notifications
You must be signed in to change notification settings - Fork 0
/
console.py
132 lines (103 loc) · 3.82 KB
/
console.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import traceback
from flask.ext.script import Manager, prompt_bool
from sqlalchemy.orm.exc import NoResultFound
from xiaoli import create_app
from xiaoli.config import setting
from xiaoli import models
from xiaoli.models import Account
from xiaoli.models.base import Base, engine
from xiaoli.models.session import db_session_cm
__author__ = 'zouyingjun'
app = create_app(setting)
manager = Manager(app)
@manager.shell
def make_shell_context():
return dict(app=app, db=engine, models=models)
@manager.command
def show_config():
u"""展示所有配置信息"""
for key in sorted(dir(setting)):
if key.isupper():
print key, '=', getattr(setting, key)
@manager.command
def create_db(default_data=True, sample_data=False):
"Creates database from sqlalchemy models"
connect = engine.connect()
connect.execute("CREATE DATABASE %s;" % setting.DB_META["db_name"])
connect.close()
@manager.command
def drop_tables():
"Drops database tables"
if prompt_bool("Are you sure you want to lose all your data"):
Base.metadata.drop_all()
@manager.command
def create_tables(default_data=True, sample_data=False):
"Creates database tables from sqlalchemy models"
Base.metadata.create_all()
@manager.command
def recreate(default_data=True, sample_data=False):
"Recreates database tables (same as issuing 'drop' and then 'create')"
drop_tables()
create_tables(default_data, sample_data)
@manager.command
def build_sample_db():
u"""初始化数据"""
user_data = [
{"cellphone": 18600000000, "nickname": "admin", "email": "admin@admin.com", "password": "admin", "type": Account.TYPE_ADMIN}
]
with db_session_cm() as session:
# init uses
for ud in user_data:
user = Account(ud.get("cellphone"), ud.get("password"))
user.nickname = ud.get("nickname")
user.email = ud.get("email")
session.add(user)
session.commit()
@manager.command
def set_admin_user(phone=None):
u"""
设置用户为管理员
用法(在项目根目录执行下列命令):
线下:python console.py set_admin_user -p=手机号
线上:XIAOLI_ENV=production PYTHONPATH=. python console.py set_admin_user -p=手机号
"""
if not phone:
print "No phone given! Exit"
return
with db_session_cm() as session:
try:
user = session.query(Account).filter(Account.cellphone == phone).one()
if user.is_admin:
print "phone %s is admin now!" % phone
return
user.type = Account.TYPE_ADMIN
session.add(user)
session.commit()
print "Set user[%s] to admin" % phone
except NoResultFound as e:
print "phone %s not exists!" % phone
except Exception as e:
print traceback.format_exc(e)
@manager.command
def migrate_account_friend_rel_to_account_friends():
u"""迁移 account_friends_rel_table 数据到 account_friends 表"""
from xiaoli.models.relationships import account_friends_rel_table
from xiaoli.models.account import AccountFriend
with db_session_cm() as session:
afr_q = session.query(account_friends_rel_table)
for afr in afr_q:
from_account_id = afr[0]
to_account_id = afr[1]
exists_af = session.query(AccountFriend).filter(AccountFriend.from_account_id == from_account_id).\
filter(AccountFriend.to_account_id == to_account_id).first()
if not exists_af:
print afr
af = AccountFriend()
af.from_account_id = from_account_id
af.to_account_id = to_account_id
session.add(af)
session.commit()
if __name__ == '__main__':
manager.run()