Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, exception): # coverage: disable
self.exception = exception
_, _, tb = sys.exc_info()
self.tb = Traceback(tb)
def suppress_warpper(*args, **kwargs):
try:
func(*args, **kwargs)
except:
et, ev, tb = sys.exc_info()
if notify is not None:
notify(except_type=et, except_value=ev, traceback=tb)
if verbose == 1 or isinstance(ev, CriticalException):
reraise(et, ev, Traceback(tb).as_traceback())
def __init__(self, e_type, e_value, traceback):
self.e_type = dill.dumps(e_type)
self.e_value = dill.dumps(e_value)
self.e_traceback = Traceback(traceback)
exc=st.message
))
return make_response(simplejson.dumps(st.to_dict()), st.status_code)
except ServiceNotFound as snf:
self.logger.error('Service Not Found! => {status} \n{exc}'.format(
status=snf.status_code,
exc=snf.message
))
return make_response(simplejson.dumps(snf.to_dict()), snf.status_code)
# Generic way to return Exceptions we don't know how to handle
# But we can do a tiny bit better if it s a PyrosException
except Exception as exc_value:
exc_type, exc_value, tb = sys.exc_info()
tb = tblib.Traceback(tb)
exc_dict = {
'exc_type': str(exc_type),
# TODO : it would be nice if pyros exception wouldnt need such a check...
'exc_value': str(exc_value.message) if isinstance(exc_value, PyrosException) else str(exc_value),
'traceback': tb.to_dict()
}
self.logger.error('An exception occurred! => 500 \n{exc}'.format(exc=exc_dict))
return make_response(simplejson.dumps(exc_dict), 500)
# return make_response(e, 500)
def _unpack_picked_error(self, pickled_error):
"""Unpacks `pickled_error` into and error `message` and `tb` string."""
exc = pickle.loads(pickled_error)
message = traceback.format_exception(*exc)[-1]
tb = json.dumps(tblib.Traceback(exc[2]).to_dict())
return message, tb
self.logger.debug('setting \n%s param %s', input_data, param.get('name', None))
self.node_client.param_set(rosname, input_data)
response = make_response('{}', 200)
response.mimetype = 'application/json'
return response
# converting pyros exceptions to proper rostful exceptions
# except (InvalidMessageException, NonexistentFieldException, FieldTypeMismatchException) as exc_value:
# raise WrongMessageFormat(
# message=str(exc_value.message),
# traceback=tblib.Traceback(sys.exc_info()[2]).to_dict()
# )
except PyrosServiceTimeout as exc_value:
raise ServiceTimeout(
message=str(exc_value.message),
traceback=tblib.Traceback(sys.exc_info()[2]).to_dict()
)
except PyrosServiceNotFound as exc_value:
raise ServiceNotFound(
message=str(exc_value.message),
traceback=tblib.Traceback(sys.exc_info()[2]).to_dict()
)
# returning local exceptions
except WrongMessageFormat as wmf:
self.logger.error('Wrong message format! => {status} \n{exc}'.format(
status=wmf.status_code,
exc=wmf.message
))
return make_response(simplejson.dumps(wmf.to_dict()), wmf.status_code)
except ServiceTimeout as st:
def exception_to_safe_exception(e):
"""Convert the passed exception to a "safe" exception - this is one
that can be copied because it does not hold references to any
local data
"""
if not issubclass(e.__class__, Exception):
return TypeError(str(e))
import tblib as _tblib
tb = _tblib.Traceback(e.__traceback__)
e.__traceback__ = tb.as_traceback()
return e
'level': record.levelno,
'pathname': record.pathname,
'lineno': record.lineno,
'msg': record.msg,
'args': args,
'exc_info': record.exc_info,
'func': record.funcName,
'sinfo': record.stack_info,
'type': 'log'}
exc_info = info['exc_info']
if exc_info is not None:
# FIXME: It's hard to transmit an exception because the Error type
# might be defined in a library that is not installed on the
# receiving side. Not sure what the best way to do this is.
info['exc_info'] = Traceback(exc_info[2]).to_dict()
"""
new_exc_info = []
first = exc_info[0]
if isinstance(first, Exception):
try:
first = first.__name__
except AttributeError:
first = first.__class__.__name__
print(first, dir(first), isinstance(first, Exception))
new_exc_info.append(first)
new_exc_info.append([str(x) for x in exc_info[1].args])
new_exc_info.append(Traceback(exc_info[2]).to_dict())
info['exc_info'] = new_exc_info
"""