byceps / byceps

byceps/application.py

"""
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(

This code seems to be duplicated in your project.

    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