Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def detail(worker: "sy.workers.AbstractWorker", error_tuple: Tuple[str, str, dict]):
"""
Detail and re-raise an Exception forwarded by another worker
"""
error_name, traceback_str, attributes = error_tuple
error_name, traceback_str = error_name.decode("utf-8"), traceback_str.decode("utf-8")
attributes = sy.serde._detail(worker, attributes)
# De-serialize the traceback
tb = Traceback.from_string(traceback_str)
# Check that the error belongs to a valid set of Exceptions
if error_name in dir(sy.exceptions):
error_type = getattr(sy.exceptions, error_name)
error = error_type()
# Include special attributes if any
for attr_name, attr in attributes.items():
setattr(error, attr_name, attr)
reraise(error_type, error, tb.as_traceback())
else:
raise ValueError(f"Invalid Exception returned:\n{traceback_str}")
def raise_for_status(self, allow_redirects=True):
"""Raises stored :class:`HTTPError` or :class:`URLError`, if one occurred."""
if self.status_code == 304:
return
elif self.error:
if self.traceback:
six.reraise(Exception, Exception(self.error), Traceback.from_string(self.traceback).as_traceback())
http_error = HTTPError(self.error)
elif (self.status_code >= 300) and (self.status_code < 400) and not allow_redirects:
http_error = HTTPError('%s Redirection' % (self.status_code))
elif (self.status_code >= 400) and (self.status_code < 500):
http_error = HTTPError('%s Client Error' % (self.status_code))
elif (self.status_code >= 500) and (self.status_code < 600):
http_error = HTTPError('%s Server Error' % (self.status_code))
else:
return
http_error.response = self
raise http_error
def detail(worker: "sy.workers.AbstractWorker", error_tuple: Tuple[str, str, dict]):
"""
Detail and re-raise an Exception forwarded by another worker
"""
error_name, traceback_str, attributes = error_tuple
error_name, traceback_str = error_name.decode("utf-8"), traceback_str.decode("utf-8")
attributes = sy.serde._detail(worker, attributes)
# De-serialize the traceback
tb = Traceback.from_string(traceback_str)
# Check that the error belongs to a valid set of Exceptions
if error_name in dir(sy.exceptions):
error_type = getattr(sy.exceptions, error_name)
error = error_type()
# Include special attributes if any
for attr_name, attr in attributes.items():
setattr(error, attr_name, attr)
reraise(error_type, error, tb.as_traceback())
else:
raise ValueError(f"Invalid Exception returned:\n{traceback_str}")
def _to_remote_traceback(self, tb):
if tb and tblib is not None and self.app.conf.task_remote_tracebacks:
return tblib.Traceback.from_string(tb).as_traceback()
def raise_for_status(self, allow_redirects=True):
"""Raises stored :class:`HTTPError` or :class:`URLError`, if one occurred."""
if self.status_code == 304:
return
elif self.error:
if self.traceback:
six.reraise(Exception, Exception(self.error), Traceback.from_string(self.traceback).as_traceback())
http_error = HTTPError(self.error)
elif (self.status_code >= 300) and (self.status_code < 400) and not allow_redirects:
http_error = HTTPError('%s Redirection' % (self.status_code))
elif (self.status_code >= 400) and (self.status_code < 500):
http_error = HTTPError('%s Client Error' % (self.status_code))
elif (self.status_code >= 500) and (self.status_code < 600):
http_error = HTTPError('%s Server Error' % (self.status_code))
else:
return
http_error.response = self
raise http_error
def deserialize_exception(details) -> Exception:
"""
TODO: Support built-in types like Exception
"""
exception: Exception = None
details_dict = json.loads(details)
source = details_dict.get("source")
exception_cls_name: str = details_dict.get("class")
if source == THIS_SOURCE and exception_cls_name:
try:
klass = import_class_from_string(exception_cls_name)
exception = klass(*details_dict["args"])
t = tblib.Traceback.from_string(details_dict["traceback"])
exception.with_traceback(t.as_traceback())
except Exception as e:
exception = None
logger.error("Failed to deserialize exception (details=%s) cause=%r", details_dict, e)
if not exception:
return ExternalException(details_dict)
else:
return exception