|  | 
|  | 1 | +import logging | 
|  | 2 | +from pathlib import Path | 
|  | 3 | + | 
|  | 4 | +from connexion import FlaskApp | 
|  | 5 | +from connexion.lifecycle import ConnexionRequest, ConnexionResponse | 
|  | 6 | +from connexion.problem import problem | 
|  | 7 | + | 
|  | 8 | +# reuse the configured logger for simplicity | 
|  | 9 | +logger = logging.getLogger("uvicorn.error") | 
|  | 10 | + | 
|  | 11 | + | 
|  | 12 | +def handle_error(request: ConnexionRequest, ex: Exception) -> ConnexionResponse: | 
|  | 13 | +    """ | 
|  | 14 | +    Report an error that happened while processing a request. | 
|  | 15 | +    See: https://connexion.readthedocs.io/en/latest/exceptions.html | 
|  | 16 | +
 | 
|  | 17 | +    :param request: Request that failed | 
|  | 18 | +    :parm ex: Exception that was raised | 
|  | 19 | +    :return: ConnexionResponse with RFC7087 problem details | 
|  | 20 | +    """ | 
|  | 21 | +    # log the request URL, exception and stack trace | 
|  | 22 | +    logger.exception( | 
|  | 23 | +        "Connexion caught exception on request to %s", request.url, exc_info=ex | 
|  | 24 | +    ) | 
|  | 25 | +    return problem(title="Error", status=500, detail=repr(ex)) | 
|  | 26 | + | 
|  | 27 | + | 
|  | 28 | +def create_app() -> FlaskApp: | 
|  | 29 | +    """ | 
|  | 30 | +    Create the connexion.FlaskApp, which wraps a Flask app. | 
|  | 31 | +
 | 
|  | 32 | +    :return Newly created connexion.FlaskApp | 
|  | 33 | +    """ | 
|  | 34 | +    app = FlaskApp(__name__, specification_dir="spec/") | 
|  | 35 | +    # hook the functions to the OpenAPI spec | 
|  | 36 | +    title = {"title": "Hello World Plus Example"} | 
|  | 37 | +    app.add_api("openapi.yaml", arguments=title, validate_responses=True) | 
|  | 38 | +    app.add_api("swagger.yaml", arguments=title, validate_responses=True) | 
|  | 39 | +    # hook a function that is invoked on any exception | 
|  | 40 | +    app.add_error_handler(Exception, handle_error) | 
|  | 41 | +    # return the fully initialized connexion.FlaskApp | 
|  | 42 | +    return app | 
|  | 43 | + | 
|  | 44 | + | 
|  | 45 | +def post_greeting(name: str, body: dict) -> tuple: | 
|  | 46 | +    logger.info( | 
|  | 47 | +        "%s: name len %d, body items %d", post_greeting.__name__, len(name), len(body) | 
|  | 48 | +    ) | 
|  | 49 | +    # the body is optional | 
|  | 50 | +    message = body.get("message", None) | 
|  | 51 | +    if "crash" == message: | 
|  | 52 | +        msg = f"Found message {message}, raise ValueError" | 
|  | 53 | +        logger.info("%s", msg) | 
|  | 54 | +        raise ValueError(msg) | 
|  | 55 | +    if "invalid" == message: | 
|  | 56 | +        logger.info("Found message %s, return invalid response", message) | 
|  | 57 | +        return {"bogus": "response"} | 
|  | 58 | +    return {"greeting": f"Hello {name}"}, 200 | 
|  | 59 | + | 
|  | 60 | + | 
|  | 61 | +# define app so loader can find it | 
|  | 62 | +conn_app = create_app() | 
|  | 63 | +if __name__ == "__main__": | 
|  | 64 | +    conn_app.run(f"{Path(__file__).stem}:conn_app", port=8080) | 
0 commit comments