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 sqlalchemy as sa
import rq_dashboard
from flask import Flask
from whitenoise import WhiteNoise
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 . 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

    if not app.debug:
        import logging

        # Send ERROR via mail
        from logging.handlers import SMTPHandler

        mail_handler = SMTPHandler(
            app.config["MAIL_SERVER"],
            fromaddr=app.config["EMAIL_SENDER"],
            toaddrs=app.config["ADMIN_EMAILS"],
            subject="CSEntry: ERROR raised",
            credentials=app.config["MAIL_CREDENTIALS"],
        )
        mail_handler.setFormatter(
            logging.Formatter(
                """
            Message type:       %(levelname)s
            Location:           %(pathname)s:%(lineno)d
            Module:             %(module)s
            Function:           %(funcName)s
            Time:               %(asctime)s

            Message:

            %(message)s
        """
            )
        )
        mail_handler.setLevel(logging.ERROR)
        app.logger.addHandler(mail_handler)
        # 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 = [
        f"{key}: {value}"
        for key, value in app.config.items()
        if key
        not in (
            "SECRET_KEY",
            "LDAP_BIND_USER_PASSWORD",
            "MAIL_CREDENTIALS",
            "SQLALCHEMY_DATABASE_URI",
        )
    ]
    # The repr() of make_url hides the password
    settings_to_display.append(
        f'SQLALCHEMY_DATABASE_URI: {sa.engine.url.make_url(app.config["SQLALCHEMY_DATABASE_URI"])!r}'
    )
    settings_string = "\n".join(settings_to_display)
    app.logger.info(f"Settings:\n{settings_string}")

    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)

    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(AdminModelView(models.Tag, 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="static/")
    app.wsgi_app.add_files(
        root="/opt/conda/envs/csentry/lib/python3.6/site-packages/flask_bootstrap/static/",
        prefix="bootstrap/",
    )

    register_cli(app)

    return app