/
application.py
322 lines (249 loc) · 11.4 KB
/
application.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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
"""
byceps.application
~~~~~~~~~~~~~~~~~~
:Copyright: 2006-2020 Jochen Kupperschmidt
:License: Modified BSD, see LICENSE for details.
"""
from importlib import import_module
import os
from pathlib import Path
from typing import Any, Callable, Dict, Iterator, Optional, Tuple, Union
from flask import current_app, Flask, g, redirect
import jinja2
from . import config, config_defaults
from .database import db
from . import email
from .redis import redis
from .util.framework.blueprint import register_blueprint
from .util.l10n import set_locale
from .util import templatefilters
from .util.templating import SiteTemplateOverridesLoader
BlueprintReg = Tuple[str, Optional[str]]
def create_app(
config_filename: Union[Path, str],
config_overrides: Optional[Dict[str, Any]] = None,
) -> Flask:
"""Create the actual Flask application."""
app = Flask(__name__)
app.config.from_object(config_defaults)
app.config.from_pyfile(str(config_filename))
if config_overrides is not None:
app.config.from_mapping(config_overrides)
# Allow database URI to be overriden via environment variable.
sqlalchemy_database_uri = os.environ.get('SQLALCHEMY_DATABASE_URI')
if sqlalchemy_database_uri:
app.config['SQLALCHEMY_DATABASE_URI'] = sqlalchemy_database_uri
# Throw an exception when an undefined name is referenced in a template.
app.jinja_env.undefined = jinja2.StrictUndefined
# Set the locale.
set_locale(app.config['LOCALE']) # Fail if not configured.
# Initialize database.
db.init_app(app)
# Initialize Redis connection.
redis.init_app(app)
email.init_app(app)
config.init_app(app)
_register_blueprints(app)
templatefilters.register(app)
_add_static_file_url_rules(app)
return app
def _register_blueprints(app: Flask) -> None:
"""Register blueprints depending on the configuration."""
for name, url_prefix in _get_blueprints(app):
register_blueprint(app, name, url_prefix)
def _get_blueprints(app: Flask) -> Iterator[BlueprintReg]:
"""Yield blueprints to register on the application."""
yield from _get_blueprints_common()
current_mode = config.get_app_mode(app)
if current_mode.is_public():
yield from _get_blueprints_site()
elif current_mode.is_admin():
yield from _get_blueprints_admin()
yield from _get_blueprints_api()
yield from _get_blueprints_health()
if app.config['METRICS_ENABLED']:
yield from _get_blueprints_metrics()
if app.debug:
yield from _get_blueprints_debug()
def _get_blueprints_common() -> Iterator[BlueprintReg]:
yield from [
('authentication', '/authentication' ),
('authorization', None ),
('core', '/core' ),
('user', None ),
('user.avatar', '/users' ),
('user.creation', '/users' ),
('user.current', '/users' ),
('user.email_address', '/users/email_address' ),
]
def _get_blueprints_site() -> Iterator[BlueprintReg]:
yield from [
('attendance', '/attendance' ),
('board', '/board' ),
('consent', '/consent' ),
('news', '/news' ),
('newsletter', '/newsletter' ),
('orga_team', '/orgas' ),
('party', None ),
('seating', '/seating' ),
('shop.order', '/shop' ),
('shop.orders', '/shop/orders' ),
('snippet', None ),
('terms', '/terms' ),
('ticketing', '/tickets' ),
('user.profile', '/users' ),
('user_badge', '/user_badges' ),
('user_group', '/user_groups' ),
('user_message', '/user_messages' ),
]
def _get_blueprints_admin() -> Iterator[BlueprintReg]:
yield from [
('admin.attendance', '/admin/attendance' ),
('admin.authorization', '/admin/authorization' ),
('admin.board', '/admin/board' ),
('admin.brand', '/admin/brands' ),
('admin.consent', '/admin/consent' ),
('admin.core', None ),
('admin.dashboard', '/admin/dashboard' ),
('admin.email', '/admin/email' ),
('admin.news', '/admin/news' ),
('admin.newsletter', '/admin/newsletter' ),
('admin.jobs', '/admin/jobs' ),
('admin.orga', '/admin/orgas' ),
('admin.orga_presence', '/admin/presence' ),
('admin.orga_team', '/admin/orga_teams' ),
('admin.party', '/admin/parties' ),
('admin.seating', '/admin/seating' ),
('admin.shop', None ),
('admin.shop.article', '/admin/shop/articles' ),
('admin.shop.email', '/admin/shop/email' ),
('admin.shop.order', '/admin/shop/orders' ),
('admin.shop.shipping', '/admin/shop/shipping' ),
('admin.shop.shop', '/admin/shop/shop' ),
('admin.shop.storefront', '/admin/shop/storefronts' ),
('admin.site', '/admin/sites' ),
('admin.snippet', '/admin/snippets' ),
('admin.terms', '/admin/terms' ),
('admin.ticketing', '/admin/ticketing' ),
('admin.ticketing.checkin', '/admin/ticketing/checkin' ),
('admin.tourney', '/admin/tourney' ),
('admin.user', '/admin/users' ),
('admin.user_badge', '/admin/user_badges' ),
]
def _get_blueprints_api() -> Iterator[BlueprintReg]:
yield from [
('api.v1.attendance', '/api/v1/attendances' ),
('api.v1.snippet', '/api/v1/snippets' ),
('api.v1.tourney.avatar', '/api/v1/tourney/avatars' ),
('api.v1.tourney.match.comments', '/api/v1/tourney' ),
('api.v1.user', '/api/v1/users' ),
('api.v1.user_avatar', '/api/v1/user_avatars' ),
('api.v1.user_badge', '/api/v1/user_badges' ),
]
def _get_blueprints_health() -> Iterator[BlueprintReg]:
yield from [
('healthcheck', '/health' ),
]
def _get_blueprints_metrics() -> Iterator[BlueprintReg]:
yield from [
('metrics', '/metrics' ),
]
def _get_blueprints_debug() -> Iterator[BlueprintReg]:
yield from [
('style_guide', '/style_guide' ),
]
def _add_static_file_url_rules(app: Flask) -> None:
"""Add URL rules to for static files."""
app.add_url_rule(
'/sites/<site_id>/<path:filename>',
endpoint='site_file',
methods=['GET'],
build_only=True,
)
def init_app(app: Flask) -> None:
"""Initialize the application after is has been created."""
with app.app_context():
_set_url_root_path(app)
app_mode = config.get_app_mode()
if app_mode.is_public():
# Incorporate site-specific template overrides.
app.jinja_loader = SiteTemplateOverridesLoader()
# Set up site-aware template context processor.
app._site_context_processors = {}
app.context_processor(_get_site_template_context)
_load_announce_signal_handlers()
if app_mode.is_admin() and app.config['RQ_DASHBOARD_ENABLED']:
import rq_dashboard
app.register_blueprint(
rq_dashboard.blueprint, url_prefix='/admin/rq'
)
def _set_url_root_path(app: Flask) -> None:
"""Set an optional URL path to redirect to from the root URL path (`/`).
Important: Don't specify the target with a leading slash unless you
really mean the root of the host.
"""
target_url = app.config['ROOT_REDIRECT_TARGET']
if target_url is None:
return
status_code = app.config['ROOT_REDIRECT_STATUS_CODE']
def _redirect():
return redirect(target_url, status_code)
app.add_url_rule('/', endpoint='root', view_func=_redirect)
def _get_site_template_context() -> Dict[str, Any]:
"""Return the site-specific additions to the template context."""
site_context_processor = _find_site_template_context_processor_cached(
g.site_id
)
if not site_context_processor:
return {}
return site_context_processor()
def _find_site_template_context_processor_cached(
site_id: str
) -> Optional[Callable[[], Dict[str, Any]]]:
"""Return the template context processor for the site.
A processor will be cached after it has been obtained for the first
time.
"""
# `None` is a valid value for a site that does not specify a
# template context processor.
if site_id in current_app._site_context_processors:
return current_app._site_context_processors.get(site_id)
else:
context_processor = _find_site_template_context_processor(site_id)
current_app._site_context_processors[site_id] = context_processor
return context_processor
def _find_site_template_context_processor(
site_id: str
) -> Optional[Callable[[], Dict[str, Any]]]:
"""Import a template context processor from the site's package.
If a site package contains a module named `extension` and that
contains a top-level callable named `template_context_processor`,
then that callable is imported and returned.
"""
module_name = f'sites.{site_id}.extension'
try:
module = import_module(module_name)
except ModuleNotFoundError:
# No extension module found in site package.
return None
context_processor = getattr(module, 'template_context_processor', None)
if context_processor is None:
# Context processor not found in module.
return None
if not callable(context_processor):
# Context processor object is not callable.
return None
return context_processor
def _load_announce_signal_handlers() -> None:
"""Import modules containing handlers so they connect to the
corresponding signals.
"""
from .announce import discord
from .announce.irc import (
board,
news,
shop_order,
snippet,
user,
user_badge,
)