96 lines
2.4 KiB
Python
96 lines
2.4 KiB
Python
from contextlib import nullcontext
|
|
from datetime import datetime
|
|
|
|
from flask import Flask, g, has_app_context, request
|
|
from .extensions import default_config
|
|
|
|
|
|
def register_extensions(app):
|
|
from .extensions import db
|
|
db.init_app(app)
|
|
|
|
|
|
def register_jinja2_filters(app):
|
|
import orjson
|
|
with app.app_context():
|
|
@app.template_filter('to_json')
|
|
def to_json(value):
|
|
return orjson.dumps(value).decode('utf-8')
|
|
|
|
|
|
def register_blueprints(app):
|
|
with app.app_context():
|
|
# Register blueprints
|
|
from .views import bp as views_bp
|
|
app.register_blueprint(views_bp, url_prefix='/')
|
|
|
|
|
|
def setup_logging(app):
|
|
# Logging
|
|
if app.debug:
|
|
@app.before_request
|
|
def log_before_request():
|
|
g.log_datetime = datetime.now()
|
|
|
|
@app.after_request
|
|
def log_after_request(response):
|
|
now = datetime.now()
|
|
delta = (now - g.log_datetime).total_seconds() * 1_000
|
|
dt = now.isoformat().replace('T', ' ').split('.')[0]
|
|
print(f'[{dt}] {request.method} {request.path} -> {response.status} ({delta:.2f}ms)')
|
|
return response
|
|
|
|
|
|
def register_shell_context(app):
|
|
import inspect
|
|
from pprint import pprint
|
|
|
|
from flask_sqlalchemy import get_debug_queries
|
|
|
|
from demweb import models
|
|
from demweb.extensions import db
|
|
|
|
@app.shell_context_processor
|
|
def make_shell_context():
|
|
return {
|
|
'app': app,
|
|
'db': db,
|
|
'query': get_debug_queries,
|
|
'print': pprint,
|
|
**dict(inspect.getmembers(models, inspect.isclass))
|
|
}
|
|
|
|
|
|
def fix_sqlalchemy_uwsgi_multiprocess_bug(app):
|
|
from demweb.extensions import db
|
|
|
|
def _dispose_db_pool():
|
|
print('uWSGI+SQLAlchemy: Disposing forked() db pool!')
|
|
with app.app_context() if not has_app_context() else nullcontext():
|
|
db.engine.dispose()
|
|
|
|
try:
|
|
from uwsgidecorators import postfork
|
|
postfork(_dispose_db_pool)
|
|
except ImportError:
|
|
pass
|
|
|
|
|
|
def create_app(test_config=None):
|
|
# Create flask application object
|
|
app = Flask(__name__)
|
|
|
|
if test_config is None:
|
|
app.config.update(default_config)
|
|
else:
|
|
app.config.update(test_config)
|
|
|
|
register_extensions(app)
|
|
register_jinja2_filters(app)
|
|
register_blueprints(app)
|
|
register_shell_context(app)
|
|
setup_logging(app)
|
|
fix_sqlalchemy_uwsgi_multiprocess_bug(app)
|
|
|
|
return app
|