<s> """<STR_LIT>""" <EOL> import copy <EOL> import json <EOL> import logging <EOL> import pathlib <EOL> import sys <EOL> import flask <EOL> import jinja2 <EOL> import six <EOL> import werkzeug . exceptions <EOL> import yaml <EOL> from swagger_spec_validator . validator20 import validate_spec <EOL> from . import resolver , utils <EOL> from . handlers import AuthErrorHandler <EOL> from . operation import Operation <EOL> MODULE_PATH = pathlib . Path ( __file__ ) . absolute ( ) . parent <EOL> SWAGGER_UI_PATH = MODULE_PATH / '<STR_LIT>' / '<STR_LIT>' <EOL> SWAGGER_UI_URL = '<STR_LIT>' <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> def compatibility_layer ( spec ) : <EOL> """<STR_LIT>""" <EOL> for path_name , methods_available in spec . get ( '<STR_LIT>' , { } ) . items ( ) : <EOL> for method_name , method_def in methods_available . items ( ) : <EOL> if method_name == '<STR_LIT>' : <EOL> continue <EOL> response_definitions = { } <EOL> for response_code , response_def in method_def . get ( '<STR_LIT>' , { } ) . items ( ) : <EOL> response_definitions [ str ( response_code ) ] = response_def <EOL> method_def [ '<STR_LIT>' ] = response_definitions <EOL> return spec <EOL> class Api ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , swagger_yaml_path , base_url = None , arguments = None , <EOL> swagger_json = None , swagger_ui = None , swagger_path = None , swagger_url = None , <EOL> validate_responses = False , resolver = resolver . Resolver ( ) , <EOL> auth_all_paths = False , debug = False ) : <EOL> """<STR_LIT>""" <EOL> self . debug = debug <EOL> self . swagger_yaml_path = pathlib . Path ( swagger_yaml_path ) <EOL> logger . debug ( '<STR_LIT>' , swagger_yaml_path , <EOL> extra = { '<STR_LIT>' : swagger_yaml_path , <EOL> '<STR_LIT>' : base_url , <EOL> '<STR_LIT>' : arguments , <EOL> '<STR_LIT>' : swagger_ui , <EOL> '<STR_LIT>' : swagger_path , <EOL> '<STR_LIT>' : swagger_url , <EOL> '<STR_LIT>' : auth_all_paths } ) <EOL> arguments = arguments or { } <EOL> with swagger_yaml_path . open ( ) as swagger_yaml : <EOL> swagger_template = swagger_yaml . read ( ) <EOL> swagger_string = jinja2 . Template ( swagger_template ) . render ( ** arguments ) <EOL> self . specification = yaml . load ( swagger_string ) <EOL> logger . debug ( '<STR_LIT>' , extra = self . specification ) <EOL> self . specification = compatibility_layer ( self . specification ) <EOL> spec = copy . deepcopy ( self . specification ) <EOL> validate_spec ( spec ) <EOL> if base_url is None : <EOL> self . base_url = self . specification . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> else : <EOL> self . base_url = base_url <EOL> self . specification [ '<STR_LIT>' ] = base_url <EOL> self . produces = self . specification . get ( '<STR_LIT:produces>' , list ( ) ) <EOL> self . security = self . specification . get ( '<STR_LIT>' ) <EOL> self . security_definitions = self . specification . get ( '<STR_LIT>' , dict ( ) ) <EOL> logger . debug ( '<STR_LIT>' , self . security_definitions ) <EOL> self . definitions = self . specification . get ( '<STR_LIT>' , { } ) <EOL> self . parameter_definitions = self . specification . get ( '<STR_LIT>' , { } ) <EOL> self . response_definitions = self . specification . get ( '<STR_LIT>' , { } ) <EOL> self . swagger_path = swagger_path or SWAGGER_UI_PATH <EOL> self . swagger_url = swagger_url or SWAGGER_UI_URL <EOL> self . resolver = resolver <EOL> logger . debug ( '<STR_LIT>' , str ( validate_responses ) ) <EOL> self . validate_responses = validate_responses <EOL> self . blueprint = self . create_blueprint ( ) <EOL> if swagger_json : <EOL> self . add_swagger_json ( ) <EOL> if swagger_ui : <EOL> self . add_swagger_ui ( ) <EOL> self . add_paths ( ) <EOL> if auth_all_paths : <EOL> self . add_auth_on_not_found ( ) <EOL> def add_operation ( self , method , path , swagger_operation , path_parameters ) : <EOL> """<STR_LIT>""" <EOL> operation = Operation ( method = method , <EOL> path = path , <EOL> path_parameters = path_parameters , <EOL> operation = swagger_operation , <EOL> app_produces = self . produces , <EOL> app_security = self . security , <EOL> security_definitions = self . security_definitions , <EOL> definitions = self . definitions , <EOL> parameter_definitions = self . parameter_definitions , <EOL> response_definitions = self . response_definitions , <EOL> validate_responses = self . validate_responses , <EOL> resolver = self . resolver ) <EOL> operation_id = operation . operation_id <EOL> logger . debug ( '<STR_LIT>' , method . upper ( ) , operation_id , <EOL> extra = vars ( operation ) ) <EOL> flask_path = utils . flaskify_path ( path , operation . get_path_parameter_types ( ) ) <EOL> self . blueprint . add_url_rule ( flask_path , operation . endpoint_name , operation . function , methods = [ method ] ) <EOL> def add_paths ( self , paths = None ) : <EOL> """<STR_LIT>""" <EOL> paths = paths or self . specification . get ( '<STR_LIT>' , dict ( ) ) <EOL> for path , methods in paths . items ( ) : <EOL> logger . debug ( '<STR_LIT>' , self . base_url , path ) <EOL> path_parameters = methods . get ( '<STR_LIT>' , [ ] ) <EOL> for method , endpoint in methods . items ( ) : <EOL> if method == '<STR_LIT>' : <EOL> continue <EOL> try : <EOL> self . add_operation ( method , path , endpoint , path_parameters ) <EOL> except Exception : <EOL> url = '<STR_LIT>' . format ( base_url = self . base_url , <EOL> path = path ) <EOL> error_msg = '<STR_LIT>' . format ( <EOL> method = method . upper ( ) , <EOL> url = url ) <EOL> if self . debug : <EOL> logger . exception ( error_msg ) <EOL> else : <EOL> logger . error ( error_msg ) <EOL> six . reraise ( * sys . exc_info ( ) ) <EOL> def add_auth_on_not_found ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' ) <EOL> not_found_error = AuthErrorHandler ( werkzeug . exceptions . NotFound ( ) , security = self . security , <EOL> security_definitions = self . security_definitions ) <EOL> endpoint_name = "<STR_LIT>" . format ( name = self . blueprint . name ) <EOL> self . blueprint . add_url_rule ( '<STR_LIT>' , endpoint_name , not_found_error . function ) <EOL> def add_swagger_json ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' , self . base_url ) <EOL> endpoint_name = "<STR_LIT>" . format ( name = self . blueprint . name ) <EOL> self . blueprint . add_url_rule ( '<STR_LIT>' , endpoint_name , lambda : json . dumps ( self . specification ) ) <EOL> def add_swagger_ui ( self ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' , self . base_url , self . swagger_url ) <EOL> static_endpoint_name = "<STR_LIT>" . format ( name = self . blueprint . name ) <EOL> self . blueprint . add_url_rule ( '<STR_LIT>' . format ( swagger_url = self . swagger_url ) , <EOL> static_endpoint_name , self . swagger_ui_static ) <EOL> index_endpoint_name = "<STR_LIT>" . format ( name = self . blueprint . name ) <EOL> self . blueprint . add_url_rule ( '<STR_LIT>' . format ( swagger_url = self . swagger_url ) , <EOL> index_endpoint_name , self . swagger_ui_index ) <EOL> def create_blueprint ( self , base_url = None ) : <EOL> """<STR_LIT>""" <EOL> base_url = base_url or self . base_url <EOL> logger . debug ( '<STR_LIT>' , base_url ) <EOL> endpoint = utils . flaskify_endpoint ( base_url ) <EOL> blueprint = flask . Blueprint ( endpoint , __name__ , url_prefix = base_url , <EOL> template_folder = str ( self . swagger_path ) ) <EOL> return blueprint <EOL> def swagger_ui_index ( self ) : <EOL> return flask . render_template ( '<STR_LIT>' , api_url = self . base_url ) <EOL> def swagger_ui_static ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> return flask . send_from_directory ( str ( self . swagger_path ) , filename ) </s>
