Source code for press.config

import os.path
import warnings

from pyramid.config import Configurator
from pyramid.config.settings import asbool
from sqlalchemy.exc import SAWarning

from .auth import RootFactory
from .exceptions import AppStartUpWarning


def discover_set(settings, setting_name, env_var, default=None,
                 modifier=None):
    """Discover a setting from environment variables and add it
    to the settings dictionary. A ``default`` value can be supplied
    and used when the environment variable can not be found.

    :param settings: the settings dictionary to modify
    :type settings: dict
    :param setting_name: the name of the setting to set
    :type: setting_name: str
    :param env_var: the environment variable name
    :type env_var: str
    :param default: default value to give the setting when the environment
                    variable is not set and the setting is already empty
    :param modifier: a callable used to modify the value (e.g. type cast)
    :type modifier: callable
    :returns: None

    """
    if env_var in os.environ:
        value = os.environ[env_var]
        if callable(modifier):
            value = modifier(value)
        settings[setting_name] = value
    elif default is not None:
        settings.setdefault(setting_name, default)


def initialize_sentry_integration():  # pragma: no cover
    """\
    Used to optionally initialize the Sentry service with this app.
    See https://docs.sentry.io/platforms/python/pyramid/

    """
    # This function is not under coverage because it is boilerplate
    # from the Sentry documentation.
    try:
        import sentry_sdk
        from sentry_sdk.integrations.pyramid import PyramidIntegration
        from sentry_sdk.integrations.celery import CeleryIntegration
    except ImportError:
        # This is an optional feature to be used when the dependency
        # and configuration are available.
        return  # bail out early

    try:
        dsn = os.environ['SENTRY_DSN']
    except KeyError:
        warnings.warn(
            "Sentry is not configured because SENTRY_DSN "
            "was not supplied.",
            AppStartUpWarning,
        )
    else:
        sentry_sdk.init(
            dsn=dsn,
            integrations=[PyramidIntegration(), CeleryIntegration()],
        )


def configure(settings=None):
    """Configure the :mod:`pyramid.configure.Configurator` object"""
    if settings is None:
        settings = {}

    # Discover & check settings
    discover_set(settings, 'shared_directory', 'SHARED_DIR')
    assert os.path.exists(settings['shared_directory'])  # required
    # TODO check permissions for write access

    initialize_sentry_integration()
    discover_set(settings, 'celery.broker', 'AMQP_URL')

    discover_set(settings, 'debug', 'DEBUG', False, asbool)
    settings['logging.level'] = settings['debug'] and 'DEBUG' or 'INFO'

    # Create the configuration object
    config = Configurator(settings=settings, root_factory=RootFactory)
    config.include('.logging')
    config.include('.subscribers')
    config.include('.views')
    config.include('.tasks')
    config.include('.auth')
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", category=SAWarning)
        config.include('cnxdb.contrib.pyramid')

    config.scan(ignore=['press.celery'])
    return config


__all__ = ('configure',)