我们从Python开源项目中，提取了以下27个代码示例，用于说明如何使用inspect.trace()。
def threaded(method): @wraps(method) def wrapper(*args, **kwargs): treated_result = ThreadedResult() def _method(): treated_result.start() try: result=method(*args, **kwargs) except Exception as e: trace=inspect.trace() trace=traces(trace, 2) treated_result.fail(trace) else: treated_result.succeed(result) t=Thread(target = _method) t.start() treated_result.set_thread(t) return treated_result return wrapper
def __call__(self,method): @wraps(method) def wrapper(*args, **kwargs): async_result = ThreadedResult() def _method(): async_result.start() try: result=method(*args, **kwargs) except Exception as e: trace=inspect.trace() trace=traces(trace,start=0, end=None) if self.log: logger.error("%s\n %s" % (repr(e), '\n '.join(trace))) async_result.fail(trace) else: async_result.succeed(result) t=Thread(target = _method) t.start() async_result.set_thread(t) return async_result return wrapper
def _log_after_retry(func, trial_number, trial_time_taken): """After call strategy that logs to some logger the finished attempt.""" # Using inspect to get arguments of the relevant call frames = inspect.trace() # Look at frame #2 first because of the internal functions _do_X formated_args = _get_args_from_frame(frames, 2) if not formated_args: formated_args = _get_args_from_frame(frames, 1) if not formated_args: formated_args = "Unknown" LOG.warning("Finished retry of %(func)s for the %(num)s time after " "%(time)0.3f(s) with args: %(args)s", {'func': tenacity_utils.get_callback_name(func), 'num': tenacity_utils.to_ordinal(trial_number), 'time': trial_time_taken, 'args': formated_args})
def exception_error(): error_message = "" for i in range(len(inspect.trace())): error_line = """ File: %s - [%s] Function: %s Statement: %s -------------------------------------------------------------------------------------------""" % ( inspect.trace()[i][1], inspect.trace()[i][2], inspect.trace()[i][3], inspect.trace()[i][4]) error_message = "%s%s" % (error_message, error_line) error_message = """Error! %s %s ======================================== Error Message ====================================%s ======================================== Error Message ======================================================""" % (sys.exc_info()[0], sys.exc_info()[1], error_message) return error_message
def get_requested_page(site, request, path): """Retrieve a page from a site given a request and path. This method uses the standard `wagtail.wagtailcore.Page.route` method to retrieve a page using its path from the given site root. If a requested page exists and is published, the result of `Page.route` can be returned directly. If the page exists but is not yet published, `Page.route` raises an `Http404`, which this method tries to catch and handle. `Page.route` raises `Http404` in two cases: if no page with the given path exists and if a page exists but is unpublished. This method catches both cases, and, if they fall into the latter category, returns the requested page back to the caller despite its draft status. """ path_components = [ component for component in path.split('/') if component ] try: return site.root_page.route(request, path_components) except Http404: exception_source = inspect.trace()[-1] stack_frame = exception_source[0] page = stack_frame.f_locals['self'] path_components = stack_frame.f_locals['path_components'] if path_components: raise return page, [], {}
def __log_error(self, task, stop_on_exception): logutil=module_logger.error if stop_on_exception else module_logger.warning err_exception, pickle_trace=task.result err_trace=pickle.loads(pickle_trace) logutil('Exception in run_action: \n {}'.format(task,)) #) logutil("%s" % (repr(err_exception), )) trace='\n'.join([line.rstrip() for line in err_trace]) if trace: logutil("%s" % (trace, ) )
def __initiate_delay(self, task, previous_task=None): ''' Runs delay function associated with task to register delay in delay table ''' delay=self.__delays[task.step_id] module_logger.debug("Initiating delay: %s (previous=%s)" % (delay.delay_id, repr(previous_task))) active=True activated=datetime.utcnow() if previous_task is not None: prev_delay=self.__previous_delays[task.sequence][task.step_id] active=prev_delay.active activated=prev_delay.activated module_logger.debug("Fetched delay from previous: active: %s, activated: %s" % (active, activated)) delay_func=delay.func try: result=delay_func(activated=activated, active=active, sequence=task.sequence, recovery=task.recovery) except Exception as e: task.status=TaskStatus.failure module_logger.critical('Exception in task execution: \n {}'.format(task,)) #) trace=inspect.trace() trace=traces(trace) module_logger.critical("%s\n %s" % (repr(e), '\n '.join(trace))) module_logger.info("Stopping running processes") self.__state=EventorState.shutdown if result: task.status=TaskStatus.success result=TaskAdminMsg(msg_type=TaskAdminMsgType.result, value=task) adminq=self.__adminq_th adminq.put( result )
def argue(self, a, b, c): try: spam(a, b, c) except: self.ex = sys.exc_info() self.tr = inspect.trace() # line 48
def traces(trace, start=0, end=None): ''' File "/private/var/acrisel/sand/gradior/gradior/gradior/gradior/loop_task.py", line 41, in task_wrapper ''' result=[ "File \"%s\", line %s, in %s\n %s" % (frame.filename, frame.lineno, frame.function, frame.code_context[0].rstrip()) for frame in trace] return result
def run(options): """Run command execute.""" service_id, resource_name, method_name = lib.pad_list(options.api_path.split(".", 2), 3) request_fd = (sys.stdin if options.json_request == "-" else open(options.json_request)) method_options = lib.load_json(request_fd.read()) try: response = do_request(service_id, resource_name, method_name, method_options, options) lib.output(lib.pretty_json(response)) except TypeError as error: frm = inspect.trace()[-1] mod = inspect.getmodule(frm[0]) if mod.__name__ == 'googleapiclient.discovery': config.logger.error("googleapiclient.discovery: {}".format(error)) else: raise
def info(self): """ @rtype: dict @return: Exception name, raiser function and raiser module """ sr = super(ThriveBaseException, self).__repr__() try: callframe = inspect.trace()[-1] origin_funcname = callframe[3] origin_module = os.path.basename(callframe[1]) except IndexError: origin_funcname = origin_module = "" return {"exception": sr, "function": origin_funcname, "module": origin_module}
def create(self, relative_path, *args, **kwargs): try: create_kwargs = self.create_kwargs(relative_path, **kwargs) create_args = self.create_args(relative_path, *args) return super(FactoryFromUrlRegex, self).create(*create_args, **create_kwargs) except TypeError as ex: if len(inspect.trace()) == 1: # Note: we modify the args, and then just raise, because we want the original stack trace ex.args = (ex.args[0]+' (from regex "%s")' % self.regex_path.regex,) raise
def from_exception(cls, fault): trace = inspect.trace()[-1] module = inspect.getmodule(trace[0]) module_name = module.__name__ if module else 'unknown' return cls( function_name=trace[3], module_name=module_name, exception=fault.__class__.__name__, exception_message=six.text_type(fault))
def with_exception_logging(func): """Decorates a function to log unhandled exceptions""" def _decorator(*args, **kwargs): try: return func(*args, **kwargs) except Exception: stacktrace = inspect.trace()[1:] logger = logging.getLogger(__name__) logger.exception("An unhandled exception occurred") log_last_django_query(logger) log_stacktrace(logging.getLogger('nav.topology.stacktrace'), stacktrace) raise return wraps(func)(_decorator)
def exception_error(): error_message = "" for i in range(len(inspect.trace())): error_line = u""" File: %s - [%s] Function: %s Statement: %s -------------------------------------------------------------------------------------------""" \ % ( inspect.trace()[i][1], inspect.trace()[i][2], inspect.trace()[i][3], inspect.trace()[i][4]) error_message = "%s%s" % (error_message, error_line) error_message = u"""Error! %s %s ======================================== Error Message ====================================%s ======================================== Error Message ======================================================""" \ % ( sys.exc_info()[0], sys.exc_info()[1], error_message) return error_message
def get_variable_from_exception(exception, variable_name): """ Grab the variable from closest frame in the stack """ for frame in reversed(trace()): try: # From http://stackoverflow.com/a/9059407/6461688 frame_variable = frame[0].f_locals[variable_name] except KeyError: pass else: return frame_variable else: raise KeyError("Variable '%s' not in any stack frames", variable_name)
def task_wrapper(run_id=None, task=None, step=None, adminq=None, use_process=True, logger_info=None): ''' Args: func: object with action method with the following signature: action(self, action, unit, group, sequencer) action: object with taskid, unit, group: id of the unit to pass sqid: sequencer id to pass to action ''' global module_logger if use_process: module_logger=MpLogger.get_logger(logger_info=logger_info, name='') # name="%s.%s_%s" %(logger_info['name'], step.name, task.sequence)) task.pid=os.getpid() os.environ['EVENTOR_STEP_SEQUENCE']=str(task.sequence) os.environ['EVENTOR_STEP_RECOVERY']=str(task.recovery) os.environ['EVENTOR_STEP_NAME']=str(step.name) if setproctitle is not None and use_process: run_id_s = "%s." % run_id if run_id else '' setproctitle("eventor: %s%s.%s(%s)" % (run_id_s, step.name, task.id_, task.sequence)) # Update task with PID update=TaskAdminMsg(msg_type=TaskAdminMsgType.update, value=task) adminq.put( update ) module_logger.info('[ Step {}/{} ] Trying to run'.format(step.name, task.sequence)) try: # todo: need to pass task resources. result=step(seq_path=task.sequence, ) except Exception as e: trace=inspect.trace() trace=traces(trace) #[2:] task.result=(e, pickle.dumps(trace)) task.status=TaskStatus.failure else: task.result=result task.status=TaskStatus.success result=TaskAdminMsg(msg_type=TaskAdminMsgType.result, value=task) module_logger.info('[ Step {}/{} ] Completed, status: {}'.format(step.name, task.sequence, str(task.status), )) adminq.put( result ) return True
def debug_exception(logger=None): """ Diagnostic helper context manager It controls execution within its context and writes extended diagnostic info to the Kodi log if an unhandled exception happens within the context. The info includes the following items: - Module path. - Code fragment where the exception has happened. - Global variables. - Local variables. After logging the diagnostic info the exception is re-raised. Example:: with debug_exception(): # Some risky code raise RuntimeError('Fatal error!') :param logger: logger function which must accept a single argument which is a log message. By default it is :func:`xbmc.log` with ``ERROR`` level. """ try: yield except: if logger is None: logger = lambda msg: xbmc.log(msg, xbmc.LOGERROR) logger('Unhandled exception detected!') logger('*** Start diagnostic info ***') frame_info = inspect.trace(5)[-1] logger('File: {0}'.format(frame_info[1])) context = '' for i, line in enumerate(frame_info[4], frame_info[2] - frame_info[5]): if i == frame_info[2]: context += '{0}:>{1}'.format(str(i).rjust(5), line) else: context += '{0}: {1}'.format(str(i).rjust(5), line) logger('Code context:\n' + context) logger('Global variables:\n' + _format_vars(frame_info[0].f_globals)) logger('Local variables:\n' + _format_vars(frame_info[0].f_locals)) logger('**** End diagnostic info ****') raise
