"""Application. The API application is a `flask` application. It provides simple features such as registering a url for a specific handlers. """ from abacus_common_logic.connectors.database import db from abacus_common_logic.marshalling.custom_fields import ma from abacus_common_logic.utils.users import set_flask_user_details_from_headers from flask import Flask from owslogger import flask_logger from owsrequest import flask_request from owsresponse import response from owsresponse.adaptors.flask import flaskify from werkzeug.exceptions import HTTPException, InternalServerError def create_app(config): """Create an app.""" app = Flask(config.SERVICE_NAME) app.config.from_object(config) app.url_map.strict_slashes = False setup_application_log(app, config) setup_owsrequest(app, config) setup_db(app, config) setup_marshmallow(app) register_blueprints(app) register_error_handlers(app) register_request_handler(app) return app def setup_owsrequest(app, config): """Set up request framework.""" try: import uwsgi # noqa uwsgiRunning = True except ImportError: uwsgiRunning = False flask_request.set_rules_validator(app, 'abacus_account/access_rules.yml') flask_request.setup( application=app, environment=config.ENVIRONMENT, add_request_context=True, label_profile=True, verify_access=False, rules_file=None, access_log_only=config.ONLY_LOG_ACCESS_ERRORS, exclude_paths=[config.HEALTH_CHECK], uwsgi_cache_enabled=uwsgiRunning ) def setup_application_log(app, config): """Set up the application log.""" flask_logger.setup( app, config.ENVIRONMENT, config.LOGGER_NAME, config.LOGGER_LEVEL, config.SERVICE_NAME, config.SERVICE_VERSION, exclude_paths=[config.HEALTH_CHECK]) def setup_db(app, config): """Set up the database.""" app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False app.config['SQLALCHEMY_DATABASE_URI'] = \ 'mysql+pymysql://{}:{}@{}:{}/{}?charset=UTF8MB4'.format( config.MYSQL_DB_USER, config.MYSQL_DB_PASS, config.MYSQL_DB_HOST, config.MYSQL_DB_PORT, config.MYSQL_DB_NAME) db.init_app(app) def setup_marshmallow(app): """Set up marshmallow.""" ma.init_app(app) def register_blueprints(app): """Register blueprints.""" from abacus_account.blueprints.account import account_api from abacus_account.blueprints.account_contract_vat_detail \ import account_contract_vat_detail_api from abacus_account.blueprints.account_payee import account_payee_api from abacus_account.blueprints.account_payee_history import \ account_payee_history_api from abacus_account.blueprints.account_payment_term import account_payment_term_api from abacus_account.blueprints.account_payment_term_template import \ account_payment_term_template_api from abacus_account.blueprints.account_tax_info import account_tax_info_api from abacus_account.blueprints.account_tax_info_history import \ account_tax_info_history_api from abacus_account.blueprints.base import base_api from abacus_account.blueprints.payment_entity_payoneer_program \ import payment_entity_payoneer_program_api from abacus_account.blueprints.payment_group import payment_group_api from abacus_account.blueprints.payment_hold import payment_hold_api from abacus_account.blueprints.payment_hold_history import payment_hold_history_api from abacus_account.blueprints.reference_agreement_type import \ reference_agreement_type_api from abacus_account.blueprints.reference_payoneer_program import \ reference_payoneer_program_api from abacus_account.blueprints.spec import spec_api app.register_blueprint(base_api) app.register_blueprint(account_api) app.register_blueprint(account_payee_api) app.register_blueprint(account_payee_history_api) app.register_blueprint(account_payment_term_api) app.register_blueprint(account_payment_term_template_api) app.register_blueprint(account_tax_info_api) app.register_blueprint(account_tax_info_history_api) app.register_blueprint(account_contract_vat_detail_api) app.register_blueprint(payment_entity_payoneer_program_api) app.register_blueprint(payment_group_api) app.register_blueprint(payment_hold_api) app.register_blueprint(payment_hold_history_api) app.register_blueprint(reference_agreement_type_api) app.register_blueprint(reference_payoneer_program_api) app.register_blueprint(spec_api) def register_error_handlers(app): """Register error handlers.""" def http_error_handler(error): """Handle http errors.""" return flaskify(response.create_error_response( message=error.description, status=error.code, code='error')) app.register_error_handler(HTTPException, http_error_handler) def exception_handler(_): """Handle exceptions.""" message = ( 'The server encountered an internal error ' 'and was unable to complete your request.') return flaskify(response.create_fatal_response(message)) app.register_error_handler(InternalServerError, exception_handler) def register_request_handler(app): """Register function to preprocess request data.""" @app.before_request def before_request_start(): set_flask_user_details_from_headers()