byceps.application._get_site_template_context() - Code Metrics - Inspection of "Dispatch to site-specific template context process..." - byceps/byceps - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 956510...210001 )

by Jochen

created 2020-02-08 14:46 UTC

byceps.application._get_site_template_context() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines	6
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	4
nop	0
dl	0
loc	6
rs	10
c	0
b	0
f	0

"""
byceps.application
~~~~~~~~~~~~~~~~~~

:Copyright: 2006-2020 Jochen Kupperschmidt
:License: Modified BSD, see LICENSE for details.
"""

from importlib import import_module
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)

    # 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_site_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.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.attendance',              '/api/attendances'          ),
        ('api.snippet',                 '/api/snippets'             ),
        ('api.tourney.avatar',          '/api/tourney/avatars'      ),
        ('api.tourney.match.comments',  '/api/tourney'              ),
        ('api.user',                    '/api/users'                ),
        ('api.user_badge',              '/api/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."""
    for rule_prefix, endpoint in [
        ('/party', 'party_file'),
        ('/site', 'site_file'),
    ]:
        rule = rule_prefix + '/<path:filename>'
        app.add_url_rule(
            rule, endpoint=endpoint, 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)

        site_mode = config.get_site_mode()
        if site_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)
        elif site_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
    )
    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


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

byceps / byceps

Push — master ( 956510...210001 )

byceps.application._get_site_template_context() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like