python-client/coreapi/utils.py at version-2 · mpeter/python-client · GitHub

History

189 lines (147 loc) · 5.96 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

from coreapi import exceptions

from coreapi.compat import string_types, text_type, urlparse

from collections import namedtuple

import os

File = namedtuple('File', 'name content content_type')

File.__new__.__defaults__ = (None,)

def is_file(obj):

if isinstance(obj, File):

return True

if hasattr(obj, '__iter__') and not isinstance(obj, (string_types, list, tuple, dict)):

# A stream object.

return True

return False

def guess_filename(obj):

name = getattr(obj, 'name', None)

if name and isinstance(name, string_types) and name[0] != '<' and name[-1] != '>':

return os.path.basename(name)

return None

def determine_transport(transports, url):

"""

Given a URL determine the appropriate transport instance.

"""

url_components = urlparse.urlparse(url)

scheme = url_components.scheme.lower()

netloc = url_components.netloc

if not scheme:

raise exceptions.TransportError("URL missing scheme '%s'." % url)

if not netloc:

raise exceptions.TransportError("URL missing hostname '%s'." % url)

for transport in transports:

if scheme in transport.schemes:

return transport

raise exceptions.TransportError("Unsupported URL scheme '%s'." % scheme)

def negotiate_decoder(decoders, content_type=None):

"""

Given the value of a 'Content-Type' header, return the appropriate

codec for decoding the request content.

"""

if content_type is None:

return decoders[0]

content_type = content_type.split(';')[0].strip().lower()

main_type = content_type.split('/')[0] + '/*'

wildcard_type = '*/*'

for codec in decoders:

if codec.media_type in (content_type, main_type, wildcard_type):

return codec

msg = "Unsupported media in Content-Type header '%s'" % content_type

raise exceptions.UnsupportedContentType(msg)

def negotiate_encoder(encoders, accept=None):

"""

Given the value of a 'Accept' header, return the appropriate codec for

encoding the response content.

"""

if accept is None:

return encoders[0]

acceptable = set([

item.split(';')[0].strip().lower()

for item in accept.split(',')

])

for codec in encoders:

if codec.media_type in acceptable:

return codec

for codec in encoders:

if codec.media_type.split('/')[0] + '/*' in acceptable:

return codec

if '*/*' in acceptable:

return encoders[0]

msg = "Unsupported media in Accept header '%s'" % accept

raise exceptions.NotAcceptable(msg)

def validate_path_param(value, name):

value = _validate_form_field(value, name, allow_list=False)

if not value:

msg = 'Parameter %s: May not be empty.'

raise exceptions.ValidationError(msg % name)

return value

def validate_query_param(value, name):

return _validate_form_field(value, name)

def validate_body_param(value, encoding, name):

if encoding == 'application/json':

return _validate_json_data(value, name)

elif encoding == 'multipart/form':

return _validate_form_object(value, name, allow_files=True)

elif encoding == 'application/x-www-form-urlencoded':

return _validate_form_object(value, name)

elif encoding == 'application/octet-stream':

if not is_file(value):

msg = 'Parameter %s: Must be an file upload.'

raise exceptions.ValidationError(msg % name)

msg = 'Unsupported encoding "%s" for outgoing request.'

raise exceptions.TransportError(msg % encoding)

def validate_form_param(value, encoding, name):

if encoding == 'application/json':

return _validate_json_data(value, name)

elif encoding == 'multipart/form':

return _validate_form_field(value, name, allow_files=True)

elif encoding == 'application/x-www-form-urlencoded':

return _validate_form_field(value, name)

msg = 'Unsupported encoding "%s" for outgoing request.'

raise exceptions.TransportError(msg % encoding)

def _validate_form_object(value, name, allow_files=False):

"""

Ensure that `value` can be encoded as form data or as query parameters.

"""

if not isinstance(value, dict):

msg = 'Parameter %s: Must be an object.'

raise exceptions.ValidationError(msg % name)

return {

text_type(item_key): _validate_form_field(item_val, name, allow_files=allow_files)

for item_key, item_val in value.items()

}

def _validate_form_field(value, name, allow_files=False, allow_list=True):

"""

Ensure that `value` can be encoded as a single form data or a query parameter.

Basic types that has a simple string representation are supported.

A list of basic types is also valid.

"""

if isinstance(value, string_types):

return value

elif isinstance(value, bool) or (value is None):

return {True: 'true', False: 'false', None: ''}[value]

elif isinstance(value, (int, float)):

return "%s" % value

elif allow_list and isinstance(value, (list, tuple)) and not is_file(value):

# Only the top-level element may be a list.

return [

_validate_form_field(item, name, allow_files=False, allow_list=False)

for item in value

]

elif allow_files and is_file(value):

return value

msg = 'Parameter %s: Must be a primative type.'

raise exceptions.ValidationError(msg % name)

def _validate_json_data(value, name):

"""

Ensure that `value` can be encoded into JSON.

"""

if (value is None) or isinstance(value, (bool, int, float, string_types)):

return value

elif isinstance(value, (list, tuple)) and not is_file(value):

return [_validate_json_data(item, name) for item in value]

elif isinstance(value, dict):

return {

text_type(item_key): _validate_json_data(item_val, name)

for item_key, item_val in value.items()

}

msg = 'Parameter %s: Must be a JSON primative.'

raise exceptions.ValidationError(msg % name)

python-client/coreapi/utils.py at version-2 · mpeter/python-client · GitHub

Expand file tree

Breadcrumbs

utils.py

Latest commit

History

Breadcrumbs

utils.py

File metadata and controls