<s> """<STR_LIT>""" <EOL> import logging <EOL> import pathlib <EOL> import flask <EOL> import werkzeug . exceptions <EOL> from connexion . decorators . produces import JSONEncoder as ConnexionJSONEncoder <EOL> from connexion . resolver import Resolver <EOL> from . api import Api <EOL> from . problem import problem <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class App ( object ) : <EOL> def __init__ ( self , import_name , port = None , specification_dir = '<STR_LIT>' , <EOL> server = None , arguments = None , auth_all_paths = False , <EOL> debug = False , swagger_json = True , swagger_ui = True , swagger_path = None , <EOL> swagger_url = None ) : <EOL> """<STR_LIT>""" <EOL> self . app = flask . Flask ( import_name ) <EOL> self . app . json_encoder = ConnexionJSONEncoder <EOL> self . root_path = pathlib . Path ( self . app . root_path ) <EOL> logger . debug ( '<STR_LIT>' , self . root_path ) <EOL> specification_dir = pathlib . Path ( specification_dir ) <EOL> if specification_dir . is_absolute ( ) : <EOL> self . specification_dir = specification_dir <EOL> else : <EOL> self . specification_dir = self . root_path / specification_dir <EOL> logger . debug ( '<STR_LIT>' , self . specification_dir ) <EOL> logger . debug ( '<STR_LIT>' ) <EOL> for error_code in range ( <NUM_LIT> , <NUM_LIT> ) : <EOL> self . add_error_handler ( error_code , self . common_error_handler ) <EOL> self . port = port <EOL> self . server = server or '<STR_LIT>' <EOL> self . debug = debug <EOL> self . import_name = import_name <EOL> self . arguments = arguments or { } <EOL> self . swagger_json = swagger_json <EOL> self . swagger_ui = swagger_ui <EOL> self . swagger_path = swagger_path <EOL> self . swagger_url = swagger_url <EOL> self . auth_all_paths = auth_all_paths <EOL> @ staticmethod <EOL> def common_error_handler ( exception ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( exception , werkzeug . exceptions . HTTPException ) : <EOL> exception = werkzeug . exceptions . InternalServerError ( ) <EOL> return problem ( title = exception . name , detail = exception . description , status = exception . code ) <EOL> def add_api ( self , swagger_file , base_path = None , arguments = None , auth_all_paths = None , swagger_json = None , <EOL> swagger_ui = None , swagger_path = None , swagger_url = None , validate_responses = False , <EOL> resolver = Resolver ( ) ) : <EOL> """<STR_LIT>""" <EOL> resolver = Resolver ( resolver ) if hasattr ( resolver , '<STR_LIT>' ) else resolver <EOL> swagger_json = swagger_json if swagger_json is not None else self . swagger_json <EOL> swagger_ui = swagger_ui if swagger_ui is not None else self . swagger_ui <EOL> swagger_path = swagger_path if swagger_path is not None else self . swagger_path <EOL> swagger_url = swagger_url if swagger_url is not None else self . swagger_url <EOL> auth_all_paths = auth_all_paths if auth_all_paths is not None else self . auth_all_paths <EOL> logger . debug ( '<STR_LIT>' , swagger_file ) <EOL> arguments = arguments or dict ( ) <EOL> arguments = dict ( self . arguments , ** arguments ) <EOL> yaml_path = self . specification_dir / swagger_file <EOL> api = Api ( swagger_yaml_path = yaml_path , <EOL> base_url = base_path , arguments = arguments , <EOL> swagger_json = swagger_json , <EOL> swagger_ui = swagger_ui , <EOL> swagger_path = swagger_path , <EOL> swagger_url = swagger_url , <EOL> resolver = resolver , <EOL> validate_responses = validate_responses , <EOL> auth_all_paths = auth_all_paths , <EOL> debug = self . debug ) <EOL> self . app . register_blueprint ( api . blueprint ) <EOL> return api <EOL> def add_error_handler ( self , error_code , function ) : <EOL> """<STR_LIT>""" <EOL> self . app . error_handler_spec [ None ] [ error_code ] = function <EOL> def add_url_rule ( self , rule , endpoint = None , view_func = None , ** options ) : <EOL> """<STR_LIT>""" <EOL> log_details = { '<STR_LIT>' : endpoint , '<STR_LIT>' : view_func . __name__ } <EOL> log_details . update ( options ) <EOL> logger . debug ( '<STR_LIT>' , rule , extra = log_details ) <EOL> self . app . add_url_rule ( rule , endpoint , view_func , ** options ) <EOL> def route ( self , rule , ** options ) : <EOL> """<STR_LIT>""" <EOL> logger . debug ( '<STR_LIT>' , rule , extra = options ) <EOL> return self . app . route ( rule , ** options ) <EOL> def run ( self , port = None , server = None , debug = None , ** options ) : <EOL> """<STR_LIT>""" <EOL> if port is not None : <EOL> self . port = port <EOL> elif self . port is None : <EOL> self . port = <NUM_LIT> <EOL> if server is not None : <EOL> self . server = server <EOL> if debug is not None : <EOL> self . debug = debug <EOL> logger . debug ( '<STR_LIT>' , self . server , extra = vars ( self ) ) <EOL> if self . server == '<STR_LIT>' : <EOL> self . app . run ( '<STR_LIT>' , port = self . port , debug = self . debug , ** options ) <EOL> elif self . server == '<STR_LIT>' : <EOL> try : <EOL> import tornado . wsgi <EOL> import tornado . httpserver <EOL> import tornado . ioloop <EOL> except : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> wsgi_container = tornado . wsgi . WSGIContainer ( self . app ) <EOL> http_server = tornado . httpserver . HTTPServer ( wsgi_container , ** options ) <EOL> http_server . listen ( self . port ) <EOL> logger . info ( '<STR_LIT>' , self . port ) <EOL> tornado . ioloop . IOLoop . instance ( ) . start ( ) <EOL> elif self . server == '<STR_LIT>' : <EOL> try : <EOL> import gevent . wsgi <EOL> except : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> http_server = gevent . wsgi . WSGIServer ( ( '<STR_LIT>' , self . port ) , self . app , ** options ) <EOL> logger . info ( '<STR_LIT>' , self . port ) <EOL> http_server . serve_forever ( ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' , self . server ) <EOL> def __call__ ( self , environ , start_response ) : <EOL> """<STR_LIT>""" <EOL> return self . app ( environ , start_response ) </s>
