python-client/coreapi/utils.py at master · core-api/python-client · GitHub

This repository was archived by the owner on Mar 18, 2019. It is now read-only.

History

336 lines (280 loc) · 10.8 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

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

from coreapi import exceptions

from coreapi.compat import string_types, text_type, urlparse, _TemporaryFileWrapper

from collections import namedtuple

import os

import pkg_resources

import tempfile

def domain_matches(request, domain):

"""

Domain string matching against an outgoing request.

Patterns starting with '*' indicate a wildcard domain.

"""

if (domain is None) or (domain == '*'):

return True

host = urlparse.urlparse(request.url).hostname

if domain.startswith('*'):

return host.endswith(domain[1:])

return host == domain

def get_installed_codecs():

packages = [

(package, package.load()) for package in

pkg_resources.iter_entry_points(group='coreapi.codecs')

]

return {

package.name: cls() for (package, cls) in packages

}

# File utilities for upload and download support.

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 guess_extension(content_type):

"""

Python's `mimetypes.guess_extension` is no use because it simply returns

the first of an unordered set. We use the same set of media types here,

but take a reasonable preference on what extension to map to.

"""

return {

'application/javascript': '.js',

'application/msword': '.doc',

'application/octet-stream': '.bin',

'application/oda': '.oda',

'application/pdf': '.pdf',

'application/pkcs7-mime': '.p7c',

'application/postscript': '.ps',

'application/vnd.apple.mpegurl': '.m3u',

'application/vnd.ms-excel': '.xls',

'application/vnd.ms-powerpoint': '.ppt',

'application/x-bcpio': '.bcpio',

'application/x-cpio': '.cpio',

'application/x-csh': '.csh',

'application/x-dvi': '.dvi',

'application/x-gtar': '.gtar',

'application/x-hdf': '.hdf',

'application/x-latex': '.latex',

'application/x-mif': '.mif',

'application/x-netcdf': '.nc',

'application/x-pkcs12': '.p12',

'application/x-pn-realaudio': '.ram',

'application/x-python-code': '.pyc',

'application/x-sh': '.sh',

'application/x-shar': '.shar',

'application/x-shockwave-flash': '.swf',

'application/x-sv4cpio': '.sv4cpio',

'application/x-sv4crc': '.sv4crc',

'application/x-tar': '.tar',

'application/x-tcl': '.tcl',

'application/x-tex': '.tex',

'application/x-texinfo': '.texinfo',

'application/x-troff': '.tr',

'application/x-troff-man': '.man',

'application/x-troff-me': '.me',

'application/x-troff-ms': '.ms',

'application/x-ustar': '.ustar',

'application/x-wais-source': '.src',

'application/xml': '.xml',

'application/zip': '.zip',

'audio/basic': '.au',

'audio/mpeg': '.mp3',

'audio/x-aiff': '.aif',

'audio/x-pn-realaudio': '.ra',

'audio/x-wav': '.wav',

'image/gif': '.gif',

'image/ief': '.ief',

'image/jpeg': '.jpe',

'image/png': '.png',

'image/svg+xml': '.svg',

'image/tiff': '.tiff',

'image/vnd.microsoft.icon': '.ico',

'image/x-cmu-raster': '.ras',

'image/x-ms-bmp': '.bmp',

'image/x-portable-anymap': '.pnm',

'image/x-portable-bitmap': '.pbm',

'image/x-portable-graymap': '.pgm',

'image/x-portable-pixmap': '.ppm',

'image/x-rgb': '.rgb',

'image/x-xbitmap': '.xbm',

'image/x-xpixmap': '.xpm',

'image/x-xwindowdump': '.xwd',

'message/rfc822': '.eml',

'text/css': '.css',

'text/csv': '.csv',

'text/html': '.html',

'text/plain': '.txt',

'text/richtext': '.rtx',

'text/tab-separated-values': '.tsv',

'text/x-python': '.py',

'text/x-setext': '.etx',

'text/x-sgml': '.sgml',

'text/x-vcard': '.vcf',

'text/xml': '.xml',

'video/mp4': '.mp4',

'video/mpeg': '.mpeg',

'video/quicktime': '.mov',

'video/webm': '.webm',

'video/x-msvideo': '.avi',

'video/x-sgi-movie': '.movie'

}.get(content_type, '')

if _TemporaryFileWrapper:

# Ideally we subclass this so that we can present a custom representation.

class DownloadedFile(_TemporaryFileWrapper):

basename = None

def __repr__(self):

state = "closed" if self.closed else "open"

mode = "" if self.closed else " '%s'" % self.file.mode

return "<DownloadedFile '%s', %s%s>" % (self.name, state, mode)

def __str__(self):

return self.__repr__()

else:

# On some platforms (eg GAE) the private _TemporaryFileWrapper may not be

# available, just use the standard `NamedTemporaryFile` function

# in this case.

DownloadedFile = tempfile.NamedTemporaryFile

# Negotiation utilities. USed to determine which codec or transport class

# should be used, given a list of supported instances.

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.NetworkError("URL missing scheme '%s'." % url)

if not netloc:

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

for transport in transports:

if scheme in transport.schemes:

return transport

raise exceptions.NetworkError("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:

for media_type in codec.get_media_types():

if media_type in (content_type, main_type, wildcard_type):

return codec

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

raise exceptions.NoCodecAvailable(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:

for media_type in codec.get_media_types():

if media_type in acceptable:

return codec

for codec in encoders:

for media_type in codec.get_media_types():

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.NoCodecAvailable(msg)

# Validation utilities. Used to ensure that we get consistent validation

# exceptions when invalid types are passed as a parameter, rather than

# an exception occuring when the request is made.

def validate_path_param(value):

value = _validate_form_field(value, allow_list=False)

if not value:

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

raise exceptions.ParameterError(msg)

return value

def validate_query_param(value):

return _validate_form_field(value)

def validate_body_param(value, encoding):

if encoding == 'application/json':

return _validate_json_data(value)

elif encoding == 'multipart/form-data':

return _validate_form_object(value, allow_files=True)

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

return _validate_form_object(value)

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

if not is_file(value):

msg = 'Must be an file upload.'

raise exceptions.ParameterError(msg)

return value

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

raise exceptions.NetworkError(msg % encoding)

def validate_form_param(value, encoding):

if encoding == 'application/json':

return _validate_json_data(value)

elif encoding == 'multipart/form-data':

return _validate_form_field(value, allow_files=True)

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

return _validate_form_field(value)

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

raise exceptions.NetworkError(msg % encoding)

def _validate_form_object(value, allow_files=False):

"""

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

"""

if not isinstance(value, dict):

msg = 'Must be an object.'

raise exceptions.ParameterError(msg)

return {

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

for item_key, item_val in value.items()

}

def _validate_form_field(value, 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, allow_files=False, allow_list=False)

for item in value

]

elif allow_files and is_file(value):

return value

msg = 'Must be a primitive type.'

raise exceptions.ParameterError(msg)

def _validate_json_data(value):

"""

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) for item in value]

elif isinstance(value, dict):

return {

text_type(item_key): _validate_json_data(item_val)

for item_key, item_val in value.items()

}

msg = 'Must be a JSON primitive.'

raise exceptions.ParameterError(msg)

python-client/coreapi/utils.py at master · core-api/python-client · GitHub

Uh oh!

Expand file tree

Breadcrumbs

utils.py

Latest commit

History

Breadcrumbs

utils.py

File metadata and controls