<s> """<STR_LIT>""" <EOL> import logging <EOL> from bravado_core . docstring import create_operation_docstring <EOL> from bravado_core . exception import SwaggerMappingError <EOL> from bravado_core . formatter import SwaggerFormat <EOL> from bravado_core . param import marshal_param <EOL> from bravado_core . spec import Spec <EOL> from six import iteritems , itervalues <EOL> from bravado . docstring_property import docstring_property <EOL> from bravado . requests_client import RequestsClient <EOL> from bravado . swagger_model import Loader <EOL> from bravado . warning import warn_for_deprecated_op <EOL> log = logging . getLogger ( __name__ ) <EOL> CONFIG_DEFAULTS = { <EOL> '<STR_LIT>' : False , <EOL> } <EOL> REQUEST_OPTIONS_DEFAULTS = { <EOL> '<STR_LIT>' : [ ] , <EOL> } <EOL> class SwaggerClient ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , swagger_spec ) : <EOL> self . swagger_spec = swagger_spec <EOL> @ classmethod <EOL> def from_url ( cls , spec_url , http_client = None , request_headers = None , <EOL> config = None ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( u"<STR_LIT>" % spec_url ) <EOL> http_client = http_client or RequestsClient ( ) <EOL> loader = Loader ( http_client , request_headers = request_headers ) <EOL> spec_dict = loader . load_spec ( spec_url ) <EOL> if request_headers is not None : <EOL> http_client . request = inject_headers_for_remote_refs ( <EOL> http_client . request , request_headers ) <EOL> return cls . from_spec ( spec_dict , spec_url , http_client , config ) <EOL> @ classmethod <EOL> def from_spec ( cls , spec_dict , origin_url = None , http_client = None , <EOL> config = None ) : <EOL> """<STR_LIT>""" <EOL> http_client = http_client or RequestsClient ( ) <EOL> config = dict ( CONFIG_DEFAULTS , ** ( config or { } ) ) <EOL> swagger_spec = Spec . from_dict ( <EOL> spec_dict , origin_url , http_client , config ) <EOL> return cls ( swagger_spec ) <EOL> def get_model ( self , model_name ) : <EOL> return self . swagger_spec . definitions [ model_name ] <EOL> def __repr__ ( self ) : <EOL> return u"<STR_LIT>" % ( self . __class__ . __name__ , self . swagger_spec . api_url ) <EOL> def __getattr__ ( self , item ) : <EOL> """<STR_LIT>""" <EOL> resource = self . swagger_spec . resources . get ( item ) <EOL> if not resource : <EOL> raise AttributeError ( <EOL> '<STR_LIT>' <EOL> . format ( item , '<STR_LIT:U+002CU+0020>' . join ( dir ( self ) ) ) ) <EOL> return ResourceDecorator ( resource ) <EOL> def __dir__ ( self ) : <EOL> return self . swagger_spec . resources . keys ( ) <EOL> def inject_headers_for_remote_refs ( request_callable , request_headers ) : <EOL> """<STR_LIT>""" <EOL> def request_wrapper ( request_params , * args , ** kwargs ) : <EOL> def is_remote_ref_request ( request_kwargs ) : <EOL> return request_kwargs . get ( '<STR_LIT>' ) is None <EOL> if is_remote_ref_request ( kwargs ) : <EOL> request_params [ '<STR_LIT>' ] = request_headers <EOL> return request_callable ( request_params , * args , ** kwargs ) <EOL> return request_wrapper <EOL> class ResourceDecorator ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , resource ) : <EOL> """<STR_LIT>""" <EOL> self . resource = resource <EOL> def __getattr__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return CallableOperation ( getattr ( self . resource , name ) ) <EOL> def __dir__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . resource . __dir__ ( ) <EOL> class CallableOperation ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , operation ) : <EOL> self . operation = operation <EOL> @ docstring_property ( __doc__ ) <EOL> def __doc__ ( self ) : <EOL> return create_operation_docstring ( self . operation ) <EOL> def __getattr__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> return getattr ( self . operation , name ) <EOL> def __call__ ( self , ** op_kwargs ) : <EOL> """<STR_LIT>""" <EOL> log . debug ( u"<STR_LIT>" % ( self . operation . operation_id , op_kwargs ) ) <EOL> warn_for_deprecated_op ( self . operation ) <EOL> request_options = dict ( <EOL> REQUEST_OPTIONS_DEFAULTS , <EOL> ** ( op_kwargs . pop ( '<STR_LIT>' , { } ) ) ) <EOL> request_params = construct_request ( <EOL> self . operation , request_options , ** op_kwargs ) <EOL> config = self . operation . swagger_spec . config <EOL> http_client = self . operation . swagger_spec . http_client <EOL> also_return_response = request_options . get ( <EOL> '<STR_LIT>' , <EOL> config [ '<STR_LIT>' ] ) <EOL> return http_client . request ( <EOL> request_params , <EOL> operation = self . operation , <EOL> response_callbacks = request_options [ '<STR_LIT>' ] , <EOL> also_return_response = also_return_response ) <EOL> def construct_request ( operation , request_options , ** op_kwargs ) : <EOL> """<STR_LIT>""" <EOL> url = operation . swagger_spec . api_url . rstrip ( '<STR_LIT:/>' ) + operation . path_name <EOL> request = { <EOL> '<STR_LIT>' : operation . http_method . upper ( ) , <EOL> '<STR_LIT:url>' : url , <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : request_options . get ( '<STR_LIT>' , { } ) , <EOL> } <EOL> for request_option in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> if request_option in request_options : <EOL> request [ request_option ] = request_options [ request_option ] <EOL> construct_params ( operation , request , op_kwargs ) <EOL> return request <EOL> def construct_params ( operation , request , op_kwargs ) : <EOL> """<STR_LIT>""" <EOL> current_params = operation . params . copy ( ) <EOL> for param_name , param_value in iteritems ( op_kwargs ) : <EOL> param = current_params . pop ( param_name , None ) <EOL> if param is None : <EOL> raise SwaggerMappingError ( <EOL> "<STR_LIT>" <EOL> . format ( operation . operation_id , param_name ) ) <EOL> marshal_param ( param , param_value , request ) <EOL> for remaining_param in itervalues ( current_params ) : <EOL> if remaining_param . required : <EOL> raise SwaggerMappingError ( <EOL> '<STR_LIT>' . format ( remaining_param . name ) ) <EOL> if not remaining_param . required and remaining_param . has_default ( ) : <EOL> marshal_param ( remaining_param , None , request ) </s>
