Enable duplicate code detection for Python code
1 | """ |
||
2 | byceps.application |
||
3 | ~~~~~~~~~~~~~~~~~~ |
||
4 | |||
5 | :Copyright: 2006-2019 Jochen Kupperschmidt |
||
6 | :License: Modified BSD, see LICENSE for details. |
||
7 | """ |
||
8 | |||
9 | from importlib import import_module |
||
10 | from pathlib import Path |
||
11 | from typing import Any, Dict, Iterator, Optional, Tuple, Union |
||
12 | |||
13 | from flask import Flask, redirect |
||
14 | import jinja2 |
||
15 | |||
16 | from .blueprints.snippet.init import add_routes_for_snippets |
||
17 | from . import config, config_defaults |
||
18 | from .database import db |
||
19 | from . import email |
||
20 | from .redis import redis |
||
21 | from .util.framework.blueprint import register_blueprint |
||
22 | from .util.l10n import set_locale |
||
23 | from .util import templatefilters |
||
24 | from .util.templating import SiteTemplateOverridesLoader |
||
25 | |||
26 | |||
27 | BlueprintReg = Tuple[str, Optional[str]] |
||
28 | |||
29 | |||
30 | def create_app( |
||
1 ignored issue
–
show
Duplication
introduced
by
Loading history...
|
|||
31 | config_filename: Union[Path, str], |
||
32 | config_overrides: Optional[Dict[str, Any]] = None, |
||
33 | ) -> Flask: |
||
34 | """Create the actual Flask application.""" |
||
35 | app = Flask(__name__) |
||
36 | |||
37 | app.config.from_object(config_defaults) |
||
38 | app.config.from_pyfile(str(config_filename)) |
||
39 | if config_overrides is not None: |
||
40 | app.config.from_mapping(config_overrides) |
||
41 | |||
42 | # Throw an exception when an undefined name is referenced in a template. |
||
43 | app.jinja_env.undefined = jinja2.StrictUndefined |
||
44 | |||
45 | # Set the locale. |
||
46 | set_locale(app.config['LOCALE']) # Fail if not configured. |
||
47 | |||
48 | # Initialize database. |
||
49 | db.init_app(app) |
||
50 | |||
51 | # Initialize Redis connection. |
||
52 | redis.init_app(app) |
||
53 | |||
54 | email.init_app(app) |
||
55 | |||
56 | config.init_app(app) |
||
57 | |||
58 | _register_blueprints(app) |
||
59 | |||
60 | templatefilters.register(app) |
||
61 | |||
62 | _add_static_file_url_rules(app) |
||
63 | |||
64 | return app |
||
65 | |||
66 | |||
67 | def _register_blueprints(app: Flask) -> None: |
||
68 | """Register blueprints depending on the configuration.""" |
||
69 | for name, url_prefix in _get_blueprints(app): |
||
70 | register_blueprint(app, name, url_prefix) |
||
71 | |||
72 | |||
73 | def _get_blueprints(app: Flask) -> Iterator[BlueprintReg]: |
||
74 | """Yield blueprints to register on the application.""" |
||
75 | yield from _get_blueprints_common() |
||
76 | |||
77 | current_mode = config.get_site_mode(app) |
||
78 | if current_mode.is_public(): |
||
79 | yield from _get_blueprints_site() |
||
80 | elif current_mode.is_admin(): |
||
81 | yield from _get_blueprints_admin() |
||
82 | |||
83 | yield from _get_blueprints_api() |
||
84 | |||
85 | yield from _get_blueprints_health() |
||
86 | |||
87 | if app.config['METRICS_ENABLED']: |
||
88 | yield from _get_blueprints_metrics() |
||
89 | |||
90 | if app.debug: |
||
91 | yield from _get_blueprints_debug() |
||
92 | |||
93 | |||
94 | def _get_blueprints_common() -> Iterator[BlueprintReg]: |
||
95 | yield from [ |
||
96 | ('authentication', '/authentication' ), |
||
97 | ('authorization', None ), |
||
98 | ('core', '/core' ), |
||
99 | ('user', None ), |
||
100 | ('user.avatar', '/users' ), |
||
101 | ('user.creation', '/users' ), |
||
102 | ('user.current', '/users' ), |
||
103 | ('user.email_address', '/users/email_address' ), |
||
104 | ] |
||
105 | |||
106 | |||
107 | def _get_blueprints_site() -> Iterator[BlueprintReg]: |
||
108 | yield from [ |
||
109 | ('attendance', '/attendance' ), |
||
110 | ('board', '/board' ), |
||
111 | ('consent', '/consent' ), |
||
112 | ('news', '/news' ), |
||
113 | ('newsletter', '/newsletter' ), |
||
114 | ('orga_team', '/orgas' ), |
||
115 | ('party', None ), |
||
116 | ('seating', '/seating' ), |
||
117 | ('shop.order', '/shop' ), |
||
118 | ('shop.orders', '/shop/orders' ), |
||
119 | ('snippet', '/snippets' ), |
||
120 | ('terms', '/terms' ), |
||
121 | ('ticketing', '/tickets' ), |
||
122 | ('user.profile', '/users' ), |
||
123 | ('user_badge', '/user_badges' ), |
||
124 | ('user_group', '/user_groups' ), |
||
125 | ('user_message', '/user_messages' ), |
||
126 | ] |
||
127 | |||
128 | |||
129 | def _get_blueprints_admin() -> Iterator[BlueprintReg]: |
||
130 | yield from [ |
||
131 | ('admin.attendance', '/admin/attendance' ), |
||
132 | ('admin.authorization', '/admin/authorization' ), |
||
133 | ('admin.board', '/admin/board' ), |
||
134 | ('admin.brand', '/admin/brands' ), |
||
135 | ('admin.consent', '/admin/consent' ), |
||
136 | ('admin.core', None ), |
||
137 | ('admin.dashboard', '/admin/dashboard' ), |
||
138 | ('admin.email', '/admin/email' ), |
||
139 | ('admin.news', '/admin/news' ), |
||
140 | ('admin.newsletter', '/admin/newsletter' ), |
||
141 | ('admin.jobs', '/admin/jobs' ), |
||
142 | ('admin.orga', '/admin/orgas' ), |
||
143 | ('admin.orga_presence', '/admin/presence' ), |
||
144 | ('admin.orga_team', '/admin/orga_teams' ), |
||
145 | ('admin.party', '/admin/parties' ), |
||
146 | ('admin.seating', '/admin/seating' ), |
||
147 | ('admin.shop', None ), |
||
148 | ('admin.shop.article', '/admin/shop/articles' ), |
||
149 | ('admin.shop.email', '/admin/shop/email' ), |
||
150 | ('admin.shop.order', '/admin/shop/orders' ), |
||
151 | ('admin.shop.shipping', '/admin/shop/shipping' ), |
||
152 | ('admin.shop.shop', '/admin/shop/shop' ), |
||
153 | ('admin.site', '/admin/sites' ), |
||
154 | ('admin.snippet', '/admin/snippets' ), |
||
155 | ('admin.terms', '/admin/terms' ), |
||
156 | ('admin.ticketing', '/admin/ticketing' ), |
||
157 | ('admin.ticketing.checkin', '/admin/ticketing/checkin' ), |
||
158 | ('admin.tourney', '/admin/tourney' ), |
||
159 | ('admin.user', '/admin/users' ), |
||
160 | ('admin.user_badge', '/admin/user_badges' ), |
||
161 | ] |
||
162 | |||
163 | |||
164 | def _get_blueprints_api() -> Iterator[BlueprintReg]: |
||
165 | yield from [ |
||
166 | ('api.attendance', '/api/attendances' ), |
||
167 | ('api.snippet', '/api/snippets' ), |
||
168 | ('api.tourney.avatar', '/api/tourney/avatars' ), |
||
169 | ('api.tourney.match.comments', '/api/tourney' ), |
||
170 | ('api.user', '/api/users' ), |
||
171 | ('api.user_badge', '/api/user_badges' ), |
||
172 | ] |
||
173 | |||
174 | |||
175 | def _get_blueprints_health() -> Iterator[BlueprintReg]: |
||
176 | yield from [ |
||
177 | ('healthcheck', '/health' ), |
||
178 | ] |
||
179 | |||
180 | |||
181 | def _get_blueprints_metrics() -> Iterator[BlueprintReg]: |
||
182 | yield from [ |
||
183 | ('metrics', '/metrics' ), |
||
184 | ] |
||
185 | |||
186 | |||
187 | def _get_blueprints_debug() -> Iterator[BlueprintReg]: |
||
188 | yield from [ |
||
189 | ('style_guide', '/style_guide' ), |
||
190 | ] |
||
191 | |||
192 | |||
193 | def _add_static_file_url_rules(app: Flask) -> None: |
||
194 | """Add URL rules to for static files.""" |
||
195 | for rule_prefix, endpoint in [ |
||
196 | ('/global', 'global_file'), |
||
197 | ('/brand', 'brand_file'), |
||
198 | ('/party', 'party_file'), |
||
199 | ('/site', 'site_file'), |
||
200 | ]: |
||
201 | rule = rule_prefix + '/<path:filename>' |
||
202 | app.add_url_rule( |
||
203 | rule, endpoint=endpoint, methods=['GET'], build_only=True |
||
204 | ) |
||
205 | |||
206 | |||
207 | def init_app(app: Flask) -> None: |
||
208 | """Initialize the application after is has been created.""" |
||
209 | with app.app_context(): |
||
210 | _set_url_root_path(app) |
||
211 | |||
212 | site_mode = config.get_site_mode() |
||
213 | if site_mode.is_public(): |
||
214 | # Incorporate site-specific template overrides. |
||
215 | app.jinja_loader = SiteTemplateOverridesLoader() |
||
216 | |||
217 | site_id = config.get_current_site_id() |
||
218 | |||
219 | # Mount snippets. |
||
220 | add_routes_for_snippets(site_id) |
||
221 | |||
222 | # Import site-specific code. |
||
223 | _load_site_extension(app, site_id) |
||
224 | elif site_mode.is_admin() and app.config['RQ_DASHBOARD_ENABLED']: |
||
225 | import rq_dashboard |
||
226 | |||
227 | app.register_blueprint( |
||
228 | rq_dashboard.blueprint, url_prefix='/admin/rq' |
||
229 | ) |
||
230 | |||
231 | |||
232 | def _set_url_root_path(app: Flask) -> None: |
||
233 | """Set an optional URL path to redirect to from the root URL path (`/`). |
||
234 | |||
235 | Important: Don't specify the target with a leading slash unless you |
||
236 | really mean the root of the host. |
||
237 | """ |
||
238 | target_url = app.config['ROOT_REDIRECT_TARGET'] |
||
239 | if target_url is None: |
||
240 | return |
||
241 | |||
242 | status_code = app.config['ROOT_REDIRECT_STATUS_CODE'] |
||
243 | |||
244 | def _redirect(): |
||
245 | return redirect(target_url, status_code) |
||
246 | |||
247 | app.add_url_rule('/', endpoint='root', view_func=_redirect) |
||
248 | |||
249 | |||
250 | def _load_site_extension(app: Flask, site_id: str) -> None: |
||
251 | """Import and call custom extension code from the site's package, if |
||
252 | available. |
||
253 | |||
254 | If a site package contains a module named `extension` and it |
||
255 | contains a top-level callable named `template_context_processor`, |
||
256 | then that callable is registered with the application as a template |
||
257 | context processor. |
||
258 | """ |
||
259 | module_name = f'sites.{site_id}.extension' |
||
260 | try: |
||
261 | module = import_module(module_name) |
||
262 | except ModuleNotFoundError: |
||
263 | # No extension module found in site package. |
||
264 | return |
||
265 | |||
266 | context_processor = getattr(module, 'template_context_processor', None) |
||
267 | if context_processor is None: |
||
268 | # Context processor not found in module. |
||
269 | return |
||
270 | |||
271 | if not callable(context_processor): |
||
272 | # Context processor object is not callable. |
||
273 | return |
||
274 | |||
275 | app.context_processor(context_processor) |
||
276 |