demgirlz/demweb/app.py
2021-05-23 16:55:57 +02:00

111 lines
2.9 KiB
Python

from contextlib import nullcontext
from datetime import datetime, timedelta
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')
@app.template_filter('to_duration')
def to_duration(seconds):
out = ''
(days, remainder) = divmod(seconds, 86400)
(hours, remainder) = divmod(remainder, 3600)
(minutes, seconds) = divmod(remainder, 60)
if days:
out += f'{days:.0f}d '
if hours:
out += f'{hours:.0f}h '
if minutes:
out += f'{minutes:.0f}m '
out += f'{seconds:.0f}s'
return out
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