application.py
"""
byceps.application
~~~~~~~~~~~~~~~~~~
:Copyright: 2014-2025 Jochen Kupperschmidt
:License: Revised BSD (see `LICENSE` file for details)
"""
from datetime import timedelta
from typing import Any
from flask_babel import Babel
import jinja2
from redis import Redis
import structlog
from byceps.announce.announce import enable_announcements
from byceps.blueprints.admin import register_admin_blueprints
from byceps.blueprints.api import register_api_blueprints
from byceps.blueprints.site import register_site_blueprints
from byceps.config.converter import convert_config
from byceps.config.errors import ConfigurationError
from byceps.config.integration import parse_value_from_environment
from byceps.config.models import (
AdminAppConfig,
ApiAppConfig,
AppConfig,
AppMode,
BycepsConfig,
CliAppConfig,
SiteAppConfig,
WebAppConfig,
WorkerAppConfig,
)
from byceps.database import db
from byceps.services.jobs.blueprints.admin.views import enable_rq_dashboard
from byceps.services.site.models import SiteID
from byceps.util import templatefilters
from byceps.util.authz import load_permissions
from byceps.util.l10n import get_current_user_locale
from byceps.util.templating import create_site_template_loader
from .byceps_app import BycepsApp
log = structlog.get_logger()
def create_admin_app(
byceps_config: BycepsConfig, app_config: AdminAppConfig
) -> BycepsApp:
app = _create_app(byceps_config, app_config)
_log_app_state(app)
return app
def create_site_app(
byceps_config: BycepsConfig, app_config: SiteAppConfig
) -> BycepsApp:
app = _create_app(byceps_config, app_config)
_init_site_app(app, app_config.site_id)
_log_app_state(app)
return app
def create_api_app(
byceps_config: BycepsConfig, app_config: ApiAppConfig
) -> BycepsApp:
app = _create_app(byceps_config, app_config)
register_api_blueprints(app)
return app
def create_cli_app(byceps_config: BycepsConfig) -> BycepsApp:
app_config = CliAppConfig()
return _create_app(byceps_config, app_config)
def create_worker_app(byceps_config: BycepsConfig) -> BycepsApp:
app_config = WorkerAppConfig()
return _create_app(byceps_config, app_config)
def _create_app(
byceps_config: BycepsConfig, app_config: AppConfig
) -> BycepsApp:
"""Create the actual Flask-based BYCEPS application."""
app_mode = _get_app_mode(app_config)
site_id = _get_site_id(app_config)
app = BycepsApp(app_mode, byceps_config, site_id)
# Avoid connection errors after database becomes temporarily
# unreachable, then becomes reachable again.
app.config['SQLALCHEMY_ENGINE_OPTIONS'] = {'pool_pre_ping': True}
_configure(app, byceps_config, app_config)
app_mode = app.byceps_app_mode
# Throw an exception when an undefined name is referenced in a template.
# NB: Set via `app.jinja_options['undefined'] = ` instead of
# `app.jinja_env.undefined = ` as that would create the Jinja
# environment too early.
app.jinja_options['undefined'] = jinja2.StrictUndefined
app.babel_instance = Babel(app, locale_selector=get_current_user_locale)
# Initialize database.
db.init_app(app)
# Initialize Redis client.
app.redis_client = Redis.from_url(app.config['REDIS_URL'])
load_permissions()
app.byceps_feature_states['debug'] = app.debug
metrics_enabled = (
byceps_config.metrics.enabled and app.byceps_app_mode.is_admin()
)
app.byceps_feature_states['metrics'] = metrics_enabled
style_guide_enabled = byceps_config.development.style_guide_enabled and (
app_mode.is_admin() or app_mode.is_site()
)
app.byceps_feature_states['style_guide'] = style_guide_enabled
if app_mode.is_admin():
register_admin_blueprints(
app,
metrics_enabled=metrics_enabled,
style_guide_enabled=style_guide_enabled,
)
_enable_rq_dashboard(app)
elif app_mode.is_site():
register_site_blueprints(app, style_guide_enabled=style_guide_enabled)
templatefilters.register(app)
_add_static_file_url_rules(app)
enable_announcements()
debug_toolbar_enabled = (
byceps_config.development.toolbar_enabled
and (app_mode.is_admin() or app_mode.is_site())
and app.debug
)
if debug_toolbar_enabled:
_enable_debug_toolbar(app)
app.byceps_feature_states['debug_toolbar'] = debug_toolbar_enabled
return app
def _get_app_mode(app_config: AppConfig) -> AppMode:
"""Derive application mode from application config."""
match app_config:
case AdminAppConfig():
return AppMode.admin
case ApiAppConfig():
return AppMode.api
case CliAppConfig():
return AppMode.cli
case SiteAppConfig():
return AppMode.site
case WorkerAppConfig():
return AppMode.worker
case _:
raise ValueError('Unexpected application configuration type')
def _get_site_id(app_config: AppConfig) -> str | None:
"""Return site ID for site application configurations, `None` otherwise."""
match app_config:
case SiteAppConfig():
return app_config.site_id
case _:
return None
def _configure(
app: BycepsApp, byceps_config: BycepsConfig, app_config: AppConfig
) -> None:
"""Configure application from file, environment variables, and defaults."""
data = _assemble_configuration(byceps_config, app_config)
app.config.from_mapping(data)
def _assemble_configuration(
byceps_config: BycepsConfig, app_config: AppConfig
) -> dict[str, Any]:
"""Assemble configuration."""
data = {
# login sessions
'PERMANENT_SESSION_LIFETIME': timedelta(14),
'SESSION_COOKIE_SAMESITE': 'Lax',
'SESSION_COOKIE_SECURE': True,
# Limit incoming request content.
'MAX_CONTENT_LENGTH': 4000000,
}
data.update(convert_config(byceps_config))
if isinstance(app_config, WebAppConfig):
data['SERVER_NAME'] = app_config.server_name
# Allow configuration values to be overridden by environment variables.
data.update(_get_config_from_environment())
_ensure_required_config_keys(data)
locale = data['LOCALE']
data['BABEL_DEFAULT_LOCALE'] = locale
timezone = data['TIMEZONE']
data['BABEL_DEFAULT_TIMEZONE'] = timezone
data['SHOP_ORDER_EXPORT_TIMEZONE'] = timezone
return data
def _get_config_from_environment() -> dict[str, Any]:
"""Obtain selected config values from environment variables."""
data = {}
for key in (
'REDIS_URL',
'SESSION_COOKIE_SECURE',
'SQLALCHEMY_DATABASE_URI',
):
value = parse_value_from_environment(key)
if value is not None:
data[key] = value
return data
def _ensure_required_config_keys(config: dict[str, Any]) -> None:
"""Ensure the required configuration keys have values."""
for key in (
'LOCALE',
'REDIS_URL',
'SECRET_KEY',
'SQLALCHEMY_DATABASE_URI',
'TIMEZONE',
):
if not config.get(key):
raise ConfigurationError(
f'Missing value for configuration key "{key}".'
)
def _add_static_file_url_rules(app: BycepsApp) -> None:
"""Add URL rules to for static files."""
app.add_url_rule(
'/static_sites/<site_id>/<path:filename>',
endpoint='site_file',
methods=['GET'],
build_only=True,
)
def _init_site_app(app: BycepsApp, site_id: SiteID) -> None:
"""Initialize site application."""
# Incorporate site-specific template overrides.
app.jinja_loader = create_site_template_loader(site_id)
def _enable_debug_toolbar(app: BycepsApp) -> None:
try:
from flask_debugtoolbar import DebugToolbarExtension
except ImportError:
log.warning(
'Could not import Flask-DebugToolbar. '
'`pip install Flask-DebugToolbar` should make it available.'
)
return
app.config['DEBUG_TB_INTERCEPT_REDIRECTS'] = False
DebugToolbarExtension(app)
def _enable_rq_dashboard(app: BycepsApp) -> None:
app.config['RQ_DASHBOARD_REDIS_URL'] = app.config['REDIS_URL']
enable_rq_dashboard(app, '/rq')
app.byceps_feature_states['rq_dashboard'] = True
def _log_app_state(app: BycepsApp) -> None:
event_kw = {'app_mode': app.byceps_app_mode.name}
features = {
name: (enabled and 'enabled' or 'disabled')
for name, enabled in app.byceps_feature_states.items()
}
event_kw.update(features)
match app.byceps_app_mode:
case AppMode.site:
event_kw['site_id'] = app.site_id
log.info('Application created', **event_kw)
