whylabs/whylogs-python

View on GitHub
python/examples/integrations/flask_streaming/app.py

Summary

Maintainability
A
0 mins
Test Coverage
# -*- coding: utf-8 -*-
"""The app module, containing the app factory function."""

import atexit
import logging
import os

import pandas as pd
from dotenv import load_dotenv  # type: ignore
from extensions import init_swagger  # type: ignore
from flask import Flask, Response, jsonify
from flask_cors import CORS  # type: ignore
from joblib import load  # type: ignore
from utils import MessageException, message_exception_handler  # type: ignore

from api.views import blueprint  # type: ignore

# Load environment variables
load_dotenv()

# Initialize Dataset
df = pd.read_csv(os.environ["DATASET_URL"])

# Load model with joblib
model = load(os.environ["MODEL_PATH"])
why_logger = None


def create_app(config_object: str = "settings") -> Flask:
    """Create application factory, as explained here: http://flask.pocoo.org/docs/patterns/appfactories/.

    :param config_object: The configuration object to use.
    """

    app = Flask(__name__.split(".")[0])

    # Adding CORS
    CORS(app)

    # Adding Normal Logging
    logging.basicConfig(level=logging.DEBUG)

    app.config.from_object(config_object)

    register_extensions(app)
    register_blueprints(app)
    register_error_handlers(app)
    atexit.register(close_logger_at_exit, app)
    return app


def register_extensions(app: Flask) -> None:
    """Register Flask extensions."""
    init_swagger(app)
    return None


def register_blueprints(app: Flask) -> None:
    """Register Flask blueprints."""
    # blueprints

    app.register_blueprint(blueprint)
    return None


def register_error_handlers(app: Flask) -> None:
    """Register error handlers."""

    app.register_error_handler(MessageException, message_exception_handler)

    def render_error(error: MessageException) -> Response:
        response = jsonify(error.to_dict())
        response.status_code = error.status_code
        return response


def close_logger_at_exit(app: Flask) -> None:
    app.why_logger.close()  # type: ignore