forked from meolu/walle-web
-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
210 lines (170 loc) · 7.16 KB
/
app.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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
# -*- coding: utf-8 -*-
"""The app module, containing the app factory function."""
import gevent.monkey
gevent.monkey.patch_all()
import logging
import sys
import os
import threading
from flask import Flask, render_template, current_app
from flask_restful import Api
from walle import commands
from walle.api import access as AccessAPI
from walle.api import api as BaseAPI
from walle.api import deploy as DeployAPI
from walle.api import environment as EnvironmentAPI
from walle.api import general as GeneralAPI
from walle.api import group as GroupAPI
from walle.api import passport as PassportAPI
from walle.api import project as ProjectAPI
from walle.api import repo as RepoApi
from walle.api import role as RoleAPI
from walle.api import server as ServerAPI
from walle.api import space as SpaceAPI
from walle.api import task as TaskAPI
from walle.api import user as UserAPI
from walle.config.settings_prod import ProdConfig
from walle.model.user import UserModel, AnonymousUser
from walle.service.code import Code
from walle.service.error import WalleError
from walle.service.extensions import bcrypt, csrf_protect, db, migrate
from walle.service.extensions import login_manager, mail, permission, socketio
from walle.service.websocket import WalleSocketIO
def create_app(config_object=ProdConfig):
"""An application factory, as explained here: http://flask.pocoo.org/docs/patterns/appfactories/.
:param config_object: The configuration object to use.
"""
app = Flask(__name__.split('.')[0])
app.config.from_object(config_object)
register_extensions(app)
register_blueprints(app)
register_errorhandlers(app)
register_shellcontext(app)
register_commands(app)
register_logging(app)
@app.before_request
def before_request():
# TODO
pass
@app.teardown_request
def shutdown_session(exception=None):
# TODO
from walle.model.database import db
db.session.remove()
@app.route('/api/websocket')
def index():
return render_template('socketio.html')
# 单元测试不用开启 websocket
if app.config.get('ENV') != 'test':
register_socketio(app)
try:
reload(sys)
sys.setdefaultencoding('utf-8')
except NameError:
pass
return app
def register_extensions(app):
"""Register Flask extensions."""
bcrypt.init_app(app)
db.init_app(app)
csrf_protect.init_app(app)
login_manager.session_protection = 'strong'
login_manager.anonymous_user = AnonymousUser
@login_manager.user_loader
def load_user(user_id):
current_app.logger.info(user_id)
return UserModel.query.get(user_id)
@login_manager.unauthorized_handler
def unauthorized():
# TODO log
return BaseAPI.ApiResource.json(code=Code.unlogin)
login_manager.init_app(app)
migrate.init_app(app, db)
mail.init_app(app)
permission.init_app(app)
return app
def register_blueprints(app):
"""Register Flask blueprints."""
api = Api(app)
api.add_resource(BaseAPI.Base, '/', endpoint='root')
api.add_resource(GeneralAPI.GeneralAPI, '/api/general/<string:action>', endpoint='general')
api.add_resource(SpaceAPI.SpaceAPI, '/api/space/', '/api/space/<int:space_id>', '/api/space/<int:space_id>/<string:action>', endpoint='space')
api.add_resource(DeployAPI.DeployAPI, '/api/deploy/', '/api/deploy/<int:task_id>', endpoint='deploy')
api.add_resource(AccessAPI.AccessAPI, '/api/access/', '/api/access/<int:access_id>', endpoint='access')
api.add_resource(RoleAPI.RoleAPI, '/api/role/', endpoint='role')
api.add_resource(GroupAPI.GroupAPI, '/api/group/', '/api/group/<int:group_id>', endpoint='group')
api.add_resource(PassportAPI.PassportAPI, '/api/passport/', '/api/passport/<string:action>', endpoint='passport')
api.add_resource(UserAPI.UserAPI, '/api/user/', '/api/user/<int:user_id>/<string:action>', '/api/user/<string:action>', '/api/user/<int:user_id>', endpoint='user')
api.add_resource(ServerAPI.ServerAPI, '/api/server/', '/api/server/<int:id>', endpoint='server')
api.add_resource(ProjectAPI.ProjectAPI, '/api/project/', '/api/project/<int:project_id>', '/api/project/<int:project_id>/<string:action>', endpoint='project')
api.add_resource(RepoApi.RepoAPI, '/api/repo/<string:action>/', endpoint='repo')
api.add_resource(TaskAPI.TaskAPI, '/api/task/', '/api/task/<int:task_id>', '/api/task/<int:task_id>/<string:action>', endpoint='task')
api.add_resource(EnvironmentAPI.EnvironmentAPI, '/api/environment/', '/api/environment/<int:env_id>', endpoint='environment')
return None
def register_errorhandlers(app):
"""Register error handlers."""
@app.errorhandler(WalleError)
def render_error(error):
# response 的 json 内容为自定义错误代码和错误信息
app.logger.error(error, exc_info=1)
return error.render_error()
def register_shellcontext(app):
"""Register shell context objects."""
def shell_context():
"""Shell context objects."""
return {
'db': db,
'User': UserModel,
}
app.shell_context_processor(shell_context)
def register_commands(app):
"""Register Click commands."""
app.cli.add_command(commands.test)
app.cli.add_command(commands.lint)
app.cli.add_command(commands.clean)
app.cli.add_command(commands.urls)
def register_logging(app):
# TODO https://blog.csdn.net/zwxiaoliu/article/details/80890136
# email errors to the administrators
import logging
from logging.handlers import RotatingFileHandler
# Formatter
formatter = logging.Formatter(
'%(asctime)s %(levelname)s %(pathname)s %(lineno)s %(module)s.%(funcName)s %(message)s')
# log dir
if not os.path.exists(app.config['LOG_PATH']):
os.makedirs(app.config['LOG_PATH'])
# FileHandler Info
file_handler_info = RotatingFileHandler(filename=app.config['LOG_PATH_INFO'])
file_handler_info.setFormatter(formatter)
file_handler_info.setLevel(logging.INFO)
info_filter = InfoFilter()
file_handler_info.addFilter(info_filter)
app.logger.addHandler(file_handler_info)
# FileHandler Error
file_handler_error = RotatingFileHandler(filename=app.config['LOG_PATH_ERROR'])
file_handler_error.setFormatter(formatter)
file_handler_error.setLevel(logging.ERROR)
app.logger.addHandler(file_handler_error)
def register_socketio(app):
if len(sys.argv) > 1 and sys.argv[1] == 'db':
return app
socketio.init_app(app, async_mode='gevent')
socketio.on_namespace(WalleSocketIO(namespace='/walle'))
socket_args = {"debug": app.config.get('DEBUG'), "host": app.config.get('HOST'), "port": app.config.get('PORT')}
socket_thread = threading.Thread(target=socketio.run, name="socket_thread", args=(app, ), kwargs=socket_args)
socket_thread.start()
return app
class InfoFilter(logging.Filter):
def filter(self, record):
"""only use INFO
筛选, 只需要 INFO 级别的log
:param record:
:return:
"""
if logging.INFO <= record.levelno < logging.ERROR:
# 已经是INFO级别了
# 然后利用父类, 返回 1
return 1
else:
return 0