How to use the connexion.App function in connexion

To help you get started, we’ve selected a few connexion examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Barski-lab / cwl-airflow / cwl_airflow / wes / server.py View on Github external
def run_wes_server(args):
    app = connexion.App(__name__)
    backend = CWLAirflowBackend()
    def rs(x):
        return getattr(backend, x.split('.')[-1])
    app.add_api('openapi/swagger_configuration.yaml', resolver=Resolver(rs))
    app.run(port=args.port, host=args.host)
github biocommons / biocommons.seqrepo / src / biocommons / seqrepo / refget / __main__.py View on Github external
"""

import logging

from pkg_resources import resource_filename

import connexion
from connexion.resolver import RestyResolver
from flask import Flask, redirect


if __name__ == "__main__":
    #logging.basicConfig(level="DEBUG")

    spec_fn = resource_filename(__name__, "refget-openapi.yaml")
    cxapp = connexion.App(__name__, debug=True)
    cxapp.add_api(spec_fn,
                  validate_responses=True,
                  strict_validation=True)

    @cxapp.route('/')
    def index():
        return redirect("/refget/1/ui")

    cxapp.run(host="0.0.0.0",
              extra_files=[spec_fn],
              processes=1)
github OpenAPITools / openapi-generator / samples / openapi3 / server / petstore / python-flask / openapi_server / __main__.py View on Github external
def main():
    app = connexion.App(__name__, specification_dir='./openapi/')
    app.app.json_encoder = encoder.JSONEncoder
    app.add_api('openapi.yaml',
                arguments={'title': 'OpenAPI Petstore'},
                pythonic_params=True)
    app.run(port=8080)
github emissions-api / emissions-api / emissionsapi / web.py View on Github external
def get_products():
    """Get all products currently available from the API.

    :return: Dictionary describing the available products.
    :rtype: dict
    """
    return [
        {
            'name': name,
            'product_variable': attributes['product']
        } for name, attributes in emissionsapi.db.products.items()
    ]


# Create connexion app
app = connexion.App(__name__)

# Add swagger description to api
app.add_api(os.path.join(os.path.abspath(
    os.path.dirname(__file__)), 'openapi.yml'),
    arguments={'products': list(emissionsapi.db.products.keys())})

# Create app to run with wsgi server
application = app.app


@app.route('/')
def home():
    """Redirect / to the swagger ui

    :return: Redirect response
    :rtype: werkzeug.wrappers.response.Response
github kearch / kearch / packages / sp-gateway / openapi_server / __main__.py View on Github external
def main():
    app = connexion.App(__name__, specification_dir='./openapi/')
    app.app.json_encoder = encoder.JSONEncoder
    app.add_api('openapi.yaml', arguments={'title': 'Kearch specialist search engine gateway API'})
    app.run(port=10080)
github DataBiosphere / job-manager / servers / cromwell / jobs / __main__.py View on Github external
type=str,
                    help='Path prefix, e.g. /api/v1, to serve from',
                    default=os.environ.get('PATH_PREFIX'))

if __name__ == '__main__':
    parser.add_argument('--port',
                        type=int,
                        default=8190,
                        help='The port on which to serve HTTP requests')
    args = parser.parse_args()
else:
    # Allow unknown args if we aren't the main program, these include flags to
    # gunicorn.
    args, _ = parser.parse_known_args()

app = connexion.App(__name__, specification_dir='./swagger/', swagger_ui=False)
DEFAULT_CROMWELL_CREDENTIALS = {'cromwell_user': '', 'cromwell_password': ''}

# Load credentials for cromwell
config_path = os.environ.get('CROMWELL_CREDENTIALS')

# Set path to capabilities config
capabilities_path = os.environ.get('CAPABILITIES_CONFIG')

# Check if the credentials are provided properly
try:
    with open(config_path) as f:
        config = json.load(f)
except (IOError, TypeError):
    logger.warning(
        'Failed to load credentials file, using the default config: {}'.format(
            DEFAULT_CROMWELL_CREDENTIALS))
github zalando-incubator / catwatch / birdwatch / birdwatch.py View on Github external
def setup_webapp(config):
    app = connexion.App(__name__, config.port, specification_dir='swagger/')
    app.add_api('birdwatch-api.yaml')
    return app
github mozilla-releng / balrog / src / auslib / web / public / base.py View on Github external
from sentry_sdk import capture_exception
from specsynthase.specbuilder import SpecBuilder

import auslib.web
from auslib.errors import BadDataError
from auslib.web.admin.views.problem import problem

try:
    import html
except ImportError:  # pragma: no cover
    import cgi as html


log = logging.getLogger(__name__)

connexion_app = connexion.App(__name__, specification_dir=".", options={"swagger_ui": False})
app = connexion_app.app

current_dir = path.dirname(__file__)
web_dir = path.dirname(auslib.web.__file__)
spec = (
    SpecBuilder()
    .add_spec(path.join(current_dir, "swagger/api.yml"))
    .add_spec(path.join(current_dir, "swagger/public_api_spec.yml"))
    .add_spec(path.join(web_dir, "common/swagger/definitions.yml"))
    .add_spec(path.join(web_dir, "common/swagger/parameters.yml"))
    .add_spec(path.join(web_dir, "common/swagger/responses.yml"))
)
# Response validation should be enabled when it actually works
connexion_app.add_api(spec, strict_validation=True)
github OpenNetworkingFoundation / TAPI / RI / flask_server / tapi_server / __main__.py View on Github external
def main():
    app = connexion.App(__name__, specification_dir='./openapi/')
    app.app.json_encoder = encoder.JSONEncoder
    app.add_api('openapi.yaml', arguments={'title': 'tapi-photonic-media API'})
    with app.app.app_context():
        with current_app.open_resource("../../database/"+sys.argv[1]+"/context.json", 'r') as f:
            data = json.load(f)
            database.context = TapiCommonContext.from_dict(data)
    app.run(port=sys.argv[2])
github HendrikStrobelt / Seq2Seq-Vis / server.py View on Github external
from copy import deepcopy

from s2s.lru import LRU
from s2s.project import S2SProject
from index.annoyVectorIndex import AnnoyVectorIndex

__author__ = 'Hendrik Strobelt, Sebastian Gehrmann, Alexander M. Rush'
CONFIG_FILE_NAME = 's2s.yaml'
projects = {}
cache_translate = LRU(50)
# cache_neighbors = LRU(20)
cache_compare = LRU(50)
pre_cached = []

logging.basicConfig(level=logging.INFO)
app = connexion.App(__name__)

parser = argparse.ArgumentParser(
    formatter_class=argparse.ArgumentDefaultsHelpFormatter)
parser.add_argument("--debug", action='store_true', help=' Debug mode')
parser.add_argument("--port", default="8080", help="Port to run the app. ")
# parser.add_argument("--nocache", default=False)
parser.add_argument("--preload", action='store_true', help="Preload indices.")
parser.add_argument("--cache", type=str, default='',
                    help="Preload cache from dir")
parser.add_argument("--dir", type=str,
                    default=os.path.abspath('data'),
                    help='Path to project')

# parser.add_argument('-api', type=str, default='pytorch',
#                     choices=['pytorch', 'lua'],
#                     help="""The API to use.""")