Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
generator
"""
try:
# Python 3 requires all JSON data to be a text string
lines = self._decoder1.decode(data, False).split("\n")
# Add first input line to last buffer line, if applicable, to
# handle cases where the JSON string has been chopped in half
# at the network level due to streaming
if len(self._buffer) > 0 and self._buffer[-1] is not None:
self._buffer[-1] += lines[0]
self._buffer.extend(lines[1:])
else:
self._buffer.extend(lines)
except UnicodeDecodeError as error:
six.raise_from(exceptions.DecodingError('json', error), error)
# Process data buffer
index = 0
try:
# Process each line as separate buffer
#PERF: This way the `.lstrip()` call becomes almost always a NOP
# even if it does return a different string it will only
# have to allocate a new buffer for the currently processed
# line.
while index < len(self._buffer):
while self._buffer[index]:
# Make sure buffer does not start with whitespace
#PERF: `.lstrip()` does not reallocate if the string does
# not actually start with whitespace.
self._buffer[index] = self._buffer[index].lstrip()
~ipfshttpclient.exceptions.EncoderMissingError
Parameters
----------
name : str
Encoding name. Supported options:
* ``"none"``
* ``"json"``
* ``"protobuf"``
* ``"xml"``
"""
try:
return __encodings[name.lower()]()
except KeyError:
six.raise_from(exceptions.EncoderMissingError(name), None)
def _do_request(self, *args, **kwargs):
for name, value in self._kwargs.items():
kwargs.setdefault(name, value)
try:
if self._session:
return self._session.request(*args, **kwargs)
else:
return requests.request(*args, **kwargs)
except (requests.ConnectTimeout, requests.Timeout) as error:
six.raise_from(exceptions.TimeoutError(error), error)
except requests.ConnectionError as error:
six.raise_from(exceptions.ConnectionError(error), error)
except http_client.HTTPException as error:
six.raise_from(exceptions.ProtocolError(error), error)
header = None
try:
def handler(signum, frame):
raise HashNotFoundError()
# if os.name != 'nt': # Bug #310
# signal.signal(signal.SIGALRM, handler)
# signal.alarm(10)
payload = IPFS_API.cat(digests[-1])
header = IPFS_API.cat(digests[-2])
# if os.name != 'nt': # Bug #310
# signal.alarm(0)
except ipfsapi.exceptions.TimeoutError:
print("{0} not found at {1}".format(cdxjParts[0], digests[-1]))
respString = ('{0} not found in IPFS :(' +
' <a href="http://{1}:{2}">Go home</a>').format(
path, IPWBREPLAY_HOST, IPWBREPLAY_PORT)
return Response(respString)
except TypeError as e:
print('A type error occurred')
print(e)
return "A Type Error Occurred", 500
except HTTPError as e:
print("Fetching from the IPFS failed")
print(e)
return "Fetching from IPFS failed", 503
except HashNotFoundError:
if payload is None:
print("Hashes not found:\n\t{0}\n\t{1}".format(
obj : Union[str, list, dict, int]
JSON serializable Python object
Returns
-------
bytes
"""
try:
result = json.dumps(obj, sort_keys=True, indent=None,
separators=(',', ':'), ensure_ascii=False)
if isinstance(result, six.text_type): #PY3
return result.encode("utf-8")
else: #PY2
return result
except (UnicodeEncodeError, TypeError) as error:
six.raise_from(exceptions.EncodingError('json', error), error)