How to use the connexion.FlaskApp function in connexion

Creates a mocked interface for each specified swagger spec and runs a server with a forked process to ensure swagger validates fully.

        :return:
        """

        port = 8088
        for swagger in self.service_swaggers:
            swagger = os.path.join(self.prefix, swagger)
            pid = os.fork()
            if not pid:
                name = swagger.split('/')[2]
                print(('Starting server for: {} at: {}'.format(name, swagger)))
                resolver = MockResolver(mock_all='all')
                api_extra_args ={'resolver': resolver}

                app = connexion.FlaskApp(name,
                                         swagger_json=False,
                                         swagger_ui=False)

                app.add_api(swagger + '/swagger.yaml',
                            resolver_error=True,
                            validate_responses=True,
                            strict_validation=True,
                            **api_extra_args)

                app.run(port=port)

            else:
                try:
                    print('Wait for pinging server')
                    # Let the api initialize
                    time.sleep(2)

def mock_app_run(mock_get_function_from_name):
    test_server = MagicMock(wraps=connexion.FlaskApp(__name__))
    test_server.run = MagicMock(return_value=True)
    test_app = MagicMock(return_value=test_server)
    mock_get_function_from_name.return_value = test_app
    return test_app

def setUpClass(cls):
        super().setUpClass(AccountHandler, AccountClient)
        flask_app = connexion.FlaskApp('remme.rest_api')
        flask_app.add_api(resource_filename('remme.rest_api', 'openapi.yml'),
                          resolver=RestMethodsSwitcherResolver('remme.rest_api'))
        cls.client = flask_app.app.test_client()

def connexion_register_blueprint(app, swagger_file, **kwargs):
    options = {"swagger_ui": True}
    con = connexion.FlaskApp("api/v1", app.instance_path, #/v1/swagger
                             specification_dir='api/v1/swagger',
                             options=options)
    specification = 'main.yaml'
    api = super(connexion.FlaskApp, con).add_api(specification, **kwargs)
    app.register_blueprint(api.blueprint)
    return api

"""Application code."""

import connexion

from models import db

app = connexion.FlaskApp(__name__, specification_dir=".")
app.app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
db.init_app(app.app)
with app.app.app_context():
    db.create_all()
app.add_api("api.yaml")
app.run(port=8080)

import connexion
import torchvision

from flask_cors import CORS


model = torchvision.models.detection.maskrcnn_resnet50_fpn(pretrained=True)
model.eval()

connexion_app = connexion.FlaskApp(__name__, specification_dir='./openapi/')
flask_app = connexion_app.app
flask_app.config['JSON_AS_ASCII'] = False
connexion_app.add_api('openapi.yaml', arguments={'title': 'ObjectCut API'})

CORS(flask_app)


@flask_app.route('/')
def alive_check():
    return 'Welcome to ObjectCut API!', 200

from pytorch_pretrained_bert import BertModel, BertTokenizer

from flask import render_template, redirect, send_from_directory
from flask_cors import CORS
from utils.mask_att import strip_attention
import connexion
import os
import pickle
import utils.path_fixes as pf
import numpy as np

from data.processing.create_faiss import Indexes, ContextIndexes
from data.processing.corpus_embeddings import CorpusEmbeddings, AttentionCorpusEmbeddings
from copy import deepcopy

app = connexion.FlaskApp(__name__, static_folder='client/dist', specification_dir='.')
flask_app = app.app
CORS(flask_app)

parser = argparse.ArgumentParser(
    formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("--debug", action='store_true', help=' Debug mode')
parser.add_argument("--port", default=5555, help="Port to run the app. ")

# NOTE: Connexion runs all global code twice. We need to load the info on the second pass of the app instantiating, not the first.
class FaissLoader:
    def __init__(self):
        self.embedding_faiss = None
        self.context_faiss = None
        self.embedding_corpus = None
        self.context_corpus  = None

from prometheus_client import make_wsgi_app
from werkzeug.wsgi import DispatcherMiddleware

try:
    import ijson.backends.yajl2_cffi as ijson
except ImportError:
    import ijson

from entityservice.logger_setup import setup_logging

# Logging setup
setup_logging()

# Define the WSGI application object
# Note we explicitly do this before importing our own code
con_app = connexion.FlaskApp(__name__, specification_dir='api_def', debug=True)
app = con_app.app

import entityservice.views
from entityservice.tracing import initialize_tracer
from flask_opentracing import FlaskTracer
from entityservice import database as db
from entityservice.serialization import generate_scores
from entityservice.object_store import connect_to_object_store
from entityservice.settings import Config as config
from entityservice.utils import fmt_bytes, iterable_to_stream


# Add prometheus wsgi middleware to route /metrics requests
# Note we have to call "app_dispatch" from gunicorn/wsgi
app_dispatch = DispatcherMiddleware(app, {
    '/metrics': make_wsgi_app()

environ['PATH_INFO'] = path_info[len(script_name):]
        scheme = environ.get('HTTP_X_SCHEME', '') or self.scheme
        if scheme:
            environ['wsgi.url_scheme'] = scheme
        server = environ.get('HTTP_X_FORWARDED_SERVER', '') or self.server
        if server:
            environ['HTTP_HOST'] = server
        return self.app(environ, start_response)


def hello():
    return "hello"


if __name__ == '__main__':
    app = connexion.FlaskApp(__name__)
    app.add_api('openapi.yaml')
    flask_app = app.app
    proxied = ReverseProxied(
        flask_app.wsgi_app,
        script_name='/reverse_proxied/'
    )
    flask_app.wsgi_app = proxied
    app.run(port=8080)