factory.py

# -*- coding: utf-8 -*-
"""
app.factory
~~~~~~~~~~~

Create the WSGI application.

:copyright: (c) 2017 European Spallation Source ERIC
:license: BSD 2-Clause, see LICENSE for more details.

"""
import logging
import sqlalchemy as sa
import rq_dashboard
import sentry_sdk
from flask import Flask
from whitenoise import WhiteNoise
from elasticsearch import Elasticsearch
from sentry_sdk.integrations.flask import FlaskIntegration
from . import settings, models
from .extensions import (
    db,
    migrate,
    login_manager,
    ldap_manager,
    admin,
    mail,
    jwt,
    toolbar,
    session_redis_store,
    fsession,
    cache,
)
from .admin.views import (
    AdminModelView,
    ItemAdmin,
    UserAdmin,
    TokenAdmin,
    NetworkAdmin,
    TaskAdmin,
)
from .main.views import bp as main
from .inventory.views import bp as inventory
from .network.views import bp as network
from .task.views import bp as task
from .user.views import bp as user
from .api.user import bp as user_api
from .api.inventory import bp as inventory_api
from .api.network import bp as network_api
from .commands import register_cli
from ._version import __version__
from . import utils


def create_app(config=None):
    app = Flask(__name__)

    app.config.from_object(rq_dashboard.default_settings)
    app.config.from_object(settings)
    app.config.from_envvar("LOCAL_SETTINGS", silent=True)
    app.config.update(config or {})

    app.jinja_env.filters["datetimeformat"] = utils.format_datetime
    app.jinja_env.filters["toyaml"] = utils.pretty_yaml
    app.jinja_env.globals["__version__"] = __version__

    if app.config["SENTRY_DSN"]:
        sentry_sdk.init(
            dsn=app.config["SENTRY_DSN"],
            environment=app.config["CSENTRY_ENVIRONMENT"],
            send_default_pii=True,
            integrations=[FlaskIntegration()],
        )

    if not app.debug:
        # Log to stderr
        handler = logging.StreamHandler()
        handler.setFormatter(
            logging.Formatter(
                "%(asctime)s %(levelname)s: %(message)s " "[in %(pathname)s:%(lineno)d]"
            )
        )
        # Set app logger level to DEBUG
        # otherwise only WARNING and above are propagated
        app.logger.setLevel(logging.DEBUG)
        handler.setLevel(logging.DEBUG)
        app.logger.addHandler(handler)
    app.logger.info("CSEntry created!")
    # Remove variables that contain a password
    settings_to_display = {
        key: value
        for key, value in app.config.items()
        if key
        not in (
            "SENTRY_DSN",
            "SECRET_KEY",
            "LDAP_BIND_USER_PASSWORD",
            "SQLALCHEMY_DATABASE_URI",
        )
    }
    # The repr() of make_url hides the password
    settings_to_display["SQLALCHEMY_DATABASE_URI"] = repr(
        sa.engine.url.make_url(app.config["SQLALCHEMY_DATABASE_URI"])
    )
    settings_string = "\n".join(
        [f"{key}: {settings_to_display[key]}" for key in sorted(settings_to_display)]
    )
    app.logger.info(f"Settings:\n{settings_string}")

    # Force RQ_DASHBOARD_REDIS_URL to RQ_REDIS_URL
    app.config["RQ_DASHBOARD_REDIS_URL"] = app.config["RQ_REDIS_URL"]

    db.init_app(app)
    migrate.init_app(app)
    login_manager.init_app(app)
    login_manager.login_view = "user.login"
    ldap_manager.init_app(app)
    mail.init_app(app)
    jwt.init_app(app)
    toolbar.init_app(app)
    session_redis_store.init_app(app)
    app.config["SESSION_REDIS"] = session_redis_store
    fsession.init_app(app)
    cache.init_app(app)
    app.elasticsearch = (
        Elasticsearch([app.config["ELASTICSEARCH_URL"]])
        if app.config["ELASTICSEARCH_URL"]
        else None
    )

    admin.init_app(app)
    admin.add_view(UserAdmin(models.User, db.session, endpoint="users"))
    admin.add_view(TokenAdmin(models.Token, db.session))
    admin.add_view(AdminModelView(models.Action, db.session))
    admin.add_view(AdminModelView(models.Manufacturer, db.session))
    admin.add_view(AdminModelView(models.Model, db.session))
    admin.add_view(AdminModelView(models.Location, db.session))
    admin.add_view(AdminModelView(models.Status, db.session))
    admin.add_view(ItemAdmin(models.Item, db.session))
    admin.add_view(AdminModelView(models.ItemComment, db.session))
    admin.add_view(AdminModelView(models.Domain, db.session))
    admin.add_view(AdminModelView(models.NetworkScope, db.session))
    admin.add_view(NetworkAdmin(models.Network, db.session, endpoint="networks"))
    admin.add_view(AdminModelView(models.DeviceType, db.session))
    admin.add_view(AdminModelView(models.AnsibleGroup, db.session))
    admin.add_view(AdminModelView(models.Host, db.session))
    admin.add_view(AdminModelView(models.Interface, db.session))
    admin.add_view(AdminModelView(models.Mac, db.session))
    admin.add_view(AdminModelView(models.Cname, db.session))
    admin.add_view(TaskAdmin(models.Task, db.session, endpoint="tasks"))

    app.register_blueprint(main)
    app.register_blueprint(inventory, url_prefix="/inventory")
    app.register_blueprint(network, url_prefix="/network")
    app.register_blueprint(task, url_prefix="/task")
    app.register_blueprint(user, url_prefix="/user")
    app.register_blueprint(user_api, url_prefix="/api/v1/user")
    app.register_blueprint(inventory_api, url_prefix="/api/v1/inventory")
    app.register_blueprint(network_api, url_prefix="/api/v1/network")
    app.register_blueprint(rq_dashboard.blueprint, url_prefix="/rq")

    app.wsgi_app = WhiteNoise(app.wsgi_app, root=app.config["CSENTRY_STATIC_DIR"])

    register_cli(app)

    return app