/usr/lib/python2.7/dist-packages/coreapi/client.py is in python-coreapi 2.3.3-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 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 | from coreapi import codecs, exceptions, transports
from coreapi.compat import string_types
from coreapi.document import Document, Link
from coreapi.utils import determine_transport, get_installed_codecs
import collections
import itypes
LinkAncestor = collections.namedtuple('LinkAncestor', ['document', 'keys'])
def _lookup_link(document, keys):
"""
Validates that keys looking up a link are correct.
Returns a two-tuple of (link, link_ancestors).
"""
if not isinstance(keys, (list, tuple)):
msg = "'keys' must be a list of strings or ints."
raise TypeError(msg)
if any([
not isinstance(key, string_types) and not isinstance(key, int)
for key in keys
]):
raise TypeError("'keys' must be a list of strings or ints.")
# Determine the link node being acted on, and its parent document.
# 'node' is the link we're calling the action for.
# 'document_keys' is the list of keys to the link's parent document.
node = document
link_ancestors = [LinkAncestor(document=document, keys=[])]
for idx, key in enumerate(keys):
try:
node = node[key]
except (KeyError, IndexError, TypeError):
index_string = ''.join('[%s]' % repr(key).strip('u') for key in keys)
msg = 'Index %s did not reference a link. Key %s was not found.'
raise exceptions.LinkLookupError(msg % (index_string, repr(key).strip('u')))
if isinstance(node, Document):
ancestor = LinkAncestor(document=node, keys=keys[:idx + 1])
link_ancestors.append(ancestor)
# Ensure that we've correctly indexed into a link.
if not isinstance(node, Link):
index_string = ''.join('[%s]' % repr(key).strip('u') for key in keys)
msg = "Can only call 'action' on a Link. Index %s returned type '%s'."
raise exceptions.LinkLookupError(
msg % (index_string, type(node).__name__)
)
return (node, link_ancestors)
def _validate_parameters(link, parameters):
"""
Ensure that parameters passed to the link are correct.
Raises a `ParameterError` if any parameters do not validate.
"""
provided = set(parameters.keys())
required = set([
field.name for field in link.fields if field.required
])
optional = set([
field.name for field in link.fields if not field.required
])
errors = {}
# Determine if any required field names not supplied.
missing = required - provided
for item in missing:
errors[item] = 'This parameter is required.'
# Determine any parameter names supplied that are not valid.
unexpected = provided - (optional | required)
for item in unexpected:
errors[item] = 'Unknown parameter.'
if errors:
raise exceptions.ParameterError(errors)
def get_default_decoders():
return [
codecs.CoreJSONCodec(),
codecs.JSONCodec(),
codecs.TextCodec(),
codecs.DownloadCodec()
]
def get_default_transports(auth=None, session=None):
return [
transports.HTTPTransport(auth=auth, session=session)
]
class Client(itypes.Object):
def __init__(self, decoders=None, transports=None, auth=None, session=None):
assert transports is None or auth is None, (
"Cannot specify both 'auth' and 'transports'. "
"When specifying transport instances explicitly you should set "
"the authentication directly on the transport."
)
if decoders is None:
decoders = get_default_decoders()
if transports is None:
transports = get_default_transports(auth=auth)
self._decoders = itypes.List(decoders)
self._transports = itypes.List(transports)
@property
def decoders(self):
return self._decoders
@property
def transports(self):
return self._transports
def get(self, url, format=None, force_codec=False):
link = Link(url, action='get')
decoders = self.decoders
if format:
force_codec = True
decoders = [decoder for decoder in self.decoders if decoder.format == format]
if not decoders:
installed_codecs = get_installed_codecs()
if format in installed_codecs:
decoders = [installed_codecs[format]]
else:
raise ValueError("No decoder available with format='%s'" % format)
# Perform the action, and return a new document.
transport = determine_transport(self.transports, link.url)
return transport.transition(link, decoders, force_codec=force_codec)
def reload(self, document, format=None, force_codec=False):
# Fallback for v1.x. To be removed in favour of explict `get` style.
return self.get(document.url, format=format, force_codec=force_codec)
def action(self, document, keys, params=None, validate=True, overrides=None,
action=None, encoding=None, transform=None):
if (action is not None) or (encoding is not None) or (transform is not None):
# Fallback for v1.x overrides.
# Will be removed at some point, most likely in a 2.1 release.
if overrides is None:
overrides = {}
if action is not None:
overrides['action'] = action
if encoding is not None:
overrides['encoding'] = encoding
if transform is not None:
overrides['transform'] = transform
if isinstance(keys, string_types):
keys = [keys]
if params is None:
params = {}
# Validate the keys and link parameters.
link, link_ancestors = _lookup_link(document, keys)
if validate:
_validate_parameters(link, params)
if overrides:
# Handle any explicit overrides.
url = overrides.get('url', link.url)
action = overrides.get('action', link.action)
encoding = overrides.get('encoding', link.encoding)
transform = overrides.get('transform', link.transform)
fields = overrides.get('fields', link.fields)
link = Link(url, action=action, encoding=encoding, transform=transform, fields=fields)
# Perform the action, and return a new document.
transport = determine_transport(self.transports, link.url)
return transport.transition(link, self.decoders, params=params, link_ancestors=link_ancestors)
|