<s> from __future__ import unicode_literals <EOL> from collections import OrderedDict <EOL> from coreapi . codecs import default_decoders , negotiate_decoder <EOL> from coreapi . compat import is_file , urlparse <EOL> from coreapi . document import Document , Object , Link , Array , Error <EOL> from coreapi . exceptions import ErrorMessage <EOL> from coreapi . transports . base import BaseTransport <EOL> import collections <EOL> import requests <EOL> import itypes <EOL> import mimetypes <EOL> import uritemplate <EOL> Params = collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT:path>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:body>' , '<STR_LIT:data>' , '<STR_LIT>' ] ) <EOL> empty_params = Params ( { } , { } , { } , None , { } , { } ) <EOL> def _get_method ( action ) : <EOL> if not action : <EOL> return '<STR_LIT:GET>' <EOL> return action . upper ( ) <EOL> def _get_params ( method , fields , params = None ) : <EOL> """<STR_LIT>""" <EOL> if params is None : <EOL> return empty_params <EOL> field_map = { field . name : field for field in fields } <EOL> path = { } <EOL> query = { } <EOL> headers = { } <EOL> body = None <EOL> data = { } <EOL> files = { } <EOL> for key , value in params . items ( ) : <EOL> if key not in field_map or not field_map [ key ] . location : <EOL> location = '<STR_LIT>' if method in ( '<STR_LIT:GET>' , '<STR_LIT>' ) else '<STR_LIT>' <EOL> else : <EOL> location = field_map [ key ] . location <EOL> if location == '<STR_LIT:path>' : <EOL> path [ key ] = value <EOL> elif location == '<STR_LIT>' : <EOL> query [ key ] = value <EOL> elif location == '<STR_LIT>' : <EOL> headers [ key ] = value <EOL> elif location == '<STR_LIT:body>' : <EOL> body = value <EOL> elif location == '<STR_LIT>' : <EOL> if is_file ( value ) : <EOL> files [ key ] = value <EOL> else : <EOL> data [ key ] = value <EOL> return Params ( path , query , headers , body , data , files ) <EOL> def _get_encoding ( encoding , params ) : <EOL> if encoding : <EOL> return encoding <EOL> if params . body is not None : <EOL> if is_file ( params . body ) : <EOL> return '<STR_LIT>' <EOL> return '<STR_LIT:application/json>' <EOL> elif params . files : <EOL> return '<STR_LIT>' <EOL> elif params . data : <EOL> return '<STR_LIT:application/json>' <EOL> return '<STR_LIT>' <EOL> def _get_url ( url , path_params ) : <EOL> """<STR_LIT>""" <EOL> if path_params : <EOL> return uritemplate . expand ( url , path_params ) <EOL> return url <EOL> def _get_headers ( url , decoders = None , credentials = None ) : <EOL> """<STR_LIT>""" <EOL> if decoders is None : <EOL> decoders = default_decoders <EOL> accept = '<STR_LIT>' % decoders [ <NUM_LIT:0> ] . media_type <EOL> headers = { <EOL> '<STR_LIT>' : accept , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> if credentials : <EOL> url_components = urlparse . urlparse ( url ) <EOL> host = url_components . hostname <EOL> if host in credentials : <EOL> headers [ '<STR_LIT>' ] = credentials [ host ] <EOL> return headers <EOL> def _get_content_type ( file_obj ) : <EOL> """<STR_LIT>""" <EOL> name = getattr ( file_obj , '<STR_LIT:name>' , None ) <EOL> if name is not None : <EOL> content_type , encoding = mimetypes . guess_type ( name ) <EOL> else : <EOL> content_type = None <EOL> return content_type <EOL> def _build_http_request ( url , method , headers = None , encoding = None , params = empty_params ) : <EOL> """<STR_LIT>""" <EOL> opts = { <EOL> "<STR_LIT>" : headers or { } <EOL> } <EOL> if params . query : <EOL> opts [ '<STR_LIT>' ] = params . query <EOL> if ( params . body is not None ) or params . data or params . files : <EOL> if encoding == '<STR_LIT:application/json>' : <EOL> if params . body is not None : <EOL> opts [ '<STR_LIT>' ] = params . body <EOL> else : <EOL> opts [ '<STR_LIT>' ] = params . data <EOL> elif encoding == '<STR_LIT>' : <EOL> opts [ '<STR_LIT:data>' ] = params . data <EOL> opts [ '<STR_LIT>' ] = params . files <EOL> elif encoding == '<STR_LIT>' : <EOL> opts [ '<STR_LIT:data>' ] = params . data <EOL> elif encoding == '<STR_LIT>' : <EOL> opts [ '<STR_LIT:data>' ] = params . body <EOL> content_type = _get_content_type ( params . body ) <EOL> if content_type : <EOL> opts [ '<STR_LIT>' ] [ '<STR_LIT>' ] = content_type <EOL> request = requests . Request ( method , url , ** opts ) <EOL> request = request . prepare ( ) <EOL> return request <EOL> def _send_http_request ( request ) : <EOL> session = requests . Session ( ) <EOL> response = session . send ( request ) <EOL> return response <EOL> def _coerce_to_error_content ( node ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( node , ( Document , Object ) ) : <EOL> return OrderedDict ( [ <EOL> ( key , _coerce_to_error_content ( value ) ) <EOL> for key , value in node . data . items ( ) <EOL> ] ) <EOL> elif isinstance ( node , Array ) : <EOL> return [ <EOL> _coerce_to_error_content ( item ) <EOL> for item in node <EOL> if not isinstance ( item , Link ) <EOL> ] <EOL> return node <EOL> def _coerce_to_error ( obj , default_title ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( obj , Document ) : <EOL> return Error ( <EOL> title = obj . title or default_title , <EOL> content = _coerce_to_error_content ( obj ) <EOL> ) <EOL> elif isinstance ( obj , dict ) : <EOL> return Error ( title = default_title , content = obj ) <EOL> elif isinstance ( obj , list ) : <EOL> return Error ( title = default_title , content = { '<STR_LIT>' : obj } ) <EOL> elif obj is None : <EOL> return Error ( title = default_title ) <EOL> return Error ( title = default_title , content = { '<STR_LIT:message>' : obj } ) <EOL> def _decode_result ( response , decoders = None ) : <EOL> """<STR_LIT>""" <EOL> if response . content : <EOL> content_type = response . headers . get ( '<STR_LIT>' ) <EOL> codec = negotiate_decoder ( content_type , decoders = decoders ) <EOL> result = codec . load ( response . content , base_url = response . url ) <EOL> else : <EOL> result = None <EOL> is_error = response . status_code >= <NUM_LIT> and response . status_code <= <NUM_LIT> <EOL> if is_error and not isinstance ( result , Error ) : <EOL> result = _coerce_to_error ( result , default_title = response . reason ) <EOL> return result <EOL> def _handle_inplace_replacements ( document , link , link_ancestors ) : <EOL> """<STR_LIT>""" <EOL> if not link . transform : <EOL> if link . action . lower ( ) in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> transform = '<STR_LIT>' <EOL> else : <EOL> transform = '<STR_LIT>' <EOL> else : <EOL> transform = link . transform <EOL> if transform == '<STR_LIT>' : <EOL> root = link_ancestors [ <NUM_LIT:0> ] . document <EOL> keys_to_link_parent = link_ancestors [ - <NUM_LIT:1> ] . keys <EOL> if document is None : <EOL> return root . delete_in ( keys_to_link_parent ) <EOL> return root . set_in ( keys_to_link_parent , document ) <EOL> return document <EOL> class HTTPTransport ( BaseTransport ) : <EOL> schemes = [ '<STR_LIT:http>' , '<STR_LIT>' ] <EOL> def __init__ ( self , credentials = None , headers = None , request_callback = None , response_callback = None ) : <EOL> if headers : <EOL> headers = { key . lower ( ) : value for key , value in headers . items ( ) } <EOL> self . _credentials = itypes . Dict ( credentials or { } ) <EOL> self . _headers = itypes . Dict ( headers or { } ) <EOL> self . _request_callback = request_callback <EOL> self . _response_callback = response_callback <EOL> @ property <EOL> def credentials ( self ) : <EOL> return self . _credentials <EOL> @ property <EOL> def headers ( self ) : <EOL> return self . _headers <EOL> def transition ( self , link , params = None , decoders = None , link_ancestors = None ) : <EOL> method = _get_method ( link . action ) <EOL> params = _get_params ( method , link . fields , params ) <EOL> encoding = _get_encoding ( link . encoding , params ) <EOL> url = _get_url ( link . url , params . path ) <EOL> headers = _get_headers ( url , decoders , self . credentials ) <EOL> headers . update ( self . headers ) <EOL> request = _build_http_request ( url , method , headers , encoding , params ) <EOL> if self . _request_callback : <EOL> self . _request_callback ( request ) <EOL> response = _send_http_request ( request ) <EOL> if self . _response_callback : <EOL> self . _response_callback ( response ) <EOL> result = _decode_result ( response , decoders ) <EOL> if isinstance ( result , Document ) and link_ancestors : <EOL> result = _handle_inplace_replacements ( result , link , link_ancestors ) <EOL> if isinstance ( result , Error ) : <EOL> raise ErrorMessage ( result ) <EOL> return result </s>
