Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)
"""
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)
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)
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
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)
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))
def setup_webapp(config):
app = connexion.App(__name__, config.port, specification_dir='swagger/')
app.add_api('birdwatch-api.yaml')
return app
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)
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])
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.""")