我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用types.TracebackType()。
def __init__(self, capture_exc_info=False): # type: (bool) -> None """ :param capture_exc_info: Whether to capture `sys.exc_info` when an handling an exception. This is turned off by default to reduce memory usage, but it is useful in certain cases (e.g., if you want to send exceptions to a logger that expect exc_info). Regardless, you can still check ``self.has_exceptions`` to see if an exception occurred. """ super(MemoryHandler, self).__init__() self.messages = OrderedDict() # type: Union[OrderedDict, Dict[Text, List[FilterMessage]]] self.has_exceptions = False self.capture_exc_info = capture_exc_info self.exc_info = [] # type: List[Tuple[type, Exception, TracebackType]]
def test_job_wrapper_fills_pipe_with_exception_info_bubble_up(self): module_name = self.test_jobs_module.constants.JOB_FAILED_EXCEPTION_JOB_NAME config = {} to_job, to_self = Pipe() job_wrapper(to_self, module_name, config, MockStatusUpdater(), entry_point_group_name="hoplite.test_jobs") exec_info = to_job.recv() exec_info = exec_info.get('previous_exception', None) try: self.assertEqual(exec_info['address'], "10.2.1.1") self.assertEqual(exec_info['uuid'], "5") self.assertIsInstance(exec_info['traceback'], types.TracebackType) # Get to the very bottom level of the exception information exec_info = exec_info.get('previous_exception', None) self.assertEqual(exec_info['message'], "Test Message") self.assertEqual(exec_info['type'], "Test Type String") self.assertEqual(exec_info['exception_object'], "pickled_string") except Exception, e: raise e finally: to_job.close() to_self.close()
def istraceback(object): """Return true if the object is a traceback. Traceback objects provide these attributes: tb_frame frame object at this level tb_lasti index of last attempted instruction in bytecode tb_lineno current line number in Python source code tb_next next inner traceback object (called by this level)""" return isinstance(object, types.TracebackType)
def iter_parallel(func, # type: Callable args_lists, # type: Sequence[CallArgs] ccmode=CC_PROCESSES): # type: (...) -> Iterator[Any] if not args_lists: return if ccmode != CC_OFF: args_lists = [((func, args, kwargs), {}) for args, kwargs in args_lists] wrappedfunc = tracebackwrapper else: wrappedfunc = func for result in iter_parallel_report(wrappedfunc, args_lists, ccmode=ccmode): if ccmode == CC_OFF: yield result else: tbtext = None try: if isinstance(result, ExcInfo): t, v, tb = result.exc_info if not isinstance(tb, types.TracebackType): tbtext = tb tb = None reraise(t, v, tb) else: yield result except Exception: if tbtext is not None: raise Exception(tbtext) else: traceback.print_exc() raise # ---------------------------------------------------------------------- # The data types option and style.
def is_internal_attribute(obj, attr): """Test if the attribute given is an internal python attribute. For example this function returns `True` for the `func_code` attribute of python objects. This is useful if the environment method :meth:`~SandboxedEnvironment.is_safe_attribute` is overridden. >>> from jinja2.sandbox import is_internal_attribute >>> is_internal_attribute(str, "mro") True >>> is_internal_attribute(str, "upper") False """ if isinstance(obj, types.FunctionType): if attr in UNSAFE_FUNCTION_ATTRIBUTES: return True elif isinstance(obj, types.MethodType): if attr in UNSAFE_FUNCTION_ATTRIBUTES or \ attr in UNSAFE_METHOD_ATTRIBUTES: return True elif isinstance(obj, type): if attr == 'mro': return True elif isinstance(obj, (types.CodeType, types.TracebackType, types.FrameType)): return True elif isinstance(obj, types.GeneratorType): if attr in UNSAFE_GENERATOR_ATTRIBUTES: return True return attr.startswith('__')
def make_frame_proxy(frame): proxy = TracebackFrameProxy(frame) if tproxy is None: return proxy def operation_handler(operation, *args, **kwargs): if operation in ('__getattribute__', '__getattr__'): return getattr(proxy, args[0]) elif operation == '__setattr__': proxy.__setattr__(*args, **kwargs) else: return getattr(proxy, operation)(*args, **kwargs) return tproxy(TracebackType, operation_handler)
def standard_exc_info(self): """Standard python exc_info for re-raising""" tb = self.frames[0] # the frame will be an actual traceback (or transparent proxy) if # we are on pypy or a python implementation with support for tproxy if type(tb) is not TracebackType: tb = tb.tb return self.exc_type, self.exc_value, tb
def __exit__(self, exc_type: Optional[Type[BaseException]], exc_value: Optional[Exception], traceback: Optional[TracebackType]) \ -> None: """Terminate the server and join the thread on exit.""" self.server.terminate() self.server.join()
def is_internal_attribute(obj, attr): """Test if the attribute given is an internal python attribute. For example this function returns `True` for the `func_code` attribute of python objects. This is useful if the environment method :meth:`~SandboxedEnvironment.is_safe_attribute` is overridden. >>> from jinja2.sandbox import is_internal_attribute >>> is_internal_attribute(str, "mro") True >>> is_internal_attribute(str, "upper") False """ if isinstance(obj, types.FunctionType): if attr in UNSAFE_FUNCTION_ATTRIBUTES: return True elif isinstance(obj, types.MethodType): if attr in UNSAFE_FUNCTION_ATTRIBUTES or \ attr in UNSAFE_METHOD_ATTRIBUTES: return True elif isinstance(obj, type): if attr == 'mro': return True elif isinstance(obj, (types.CodeType, types.TracebackType, types.FrameType)): return True elif isinstance(obj, types.GeneratorType): if attr in UNSAFE_GENERATOR_ATTRIBUTES: return True elif hasattr(types, 'CoroutineType') and isinstance(obj, types.CoroutineType): if attr in UNSAFE_COROUTINE_ATTRIBUTES: return True elif hasattr(types, 'AsyncGeneratorType') and isinstance(obj, types.AsyncGeneratorType): if attr in UNSAFE_ASYNC_GENERATOR_ATTRIBUTES: return True return attr.startswith('__')
def exc_info(self): # type: () -> List[Tuple[type, Exception, TracebackType]] """ Returns tracebacks from any exceptions that were captured. """ self.full_clean() return self._handler.exc_info
def as_traceback(self): if tproxy: return tproxy(TracebackType, self.__tproxy_handler) elif tb_set_next: f_code = self.tb_frame.f_code code = compile('\n' * (self.tb_lineno - 1) + 'raise __traceback_maker', self.tb_frame.f_code.co_filename, 'exec') if PY3: code = CodeType( 0, 0, f_code.co_nlocals, f_code.co_stacksize, f_code.co_flags, code.co_code, code.co_consts, code.co_names, code.co_varnames, f_code.co_filename, f_code.co_name, code.co_firstlineno, b"", (), () ) else: code = CodeType( 0, f_code.co_nlocals, f_code.co_stacksize, f_code.co_flags, code.co_code, code.co_consts, code.co_names, code.co_varnames, f_code.co_filename.encode(), f_code.co_name.encode(), code.co_firstlineno, b"", (), () ) try: exec(code, self.tb_frame.f_globals, {}) except: tb = sys.exc_info()[2].tb_next tb_set_next(tb, self.tb_next and self.tb_next.as_traceback()) try: return tb finally: del tb else: raise RuntimeError("Cannot re-create traceback !")
def install(): try: import copy_reg except ImportError: import copyreg as copy_reg copy_reg.pickle(TracebackType, pickle_traceback) # Added by gevent # We have to defer the initialization, and especially the import of platform, # until runtime. If we're monkey patched, we need to be sure to use # the original __import__ to avoid switching through the hub due to # import locks on Python 2. See also builtins.py for details.
def as_traceback(self): if tproxy: return tproxy(TracebackType, self.__tproxy_handler) elif tb_set_next: f_code = self.tb_frame.f_code code = compile('\n' * (self.tb_lineno - 1) + 'raise __traceback_maker', self.tb_frame.f_code.co_filename, 'exec') if PY3: code = CodeType( 0, code.co_kwonlyargcount, code.co_nlocals, code.co_stacksize, code.co_flags, code.co_code, code.co_consts, code.co_names, code.co_varnames, f_code.co_filename, f_code.co_name, code.co_firstlineno, code.co_lnotab, (), () ) else: code = CodeType( 0, code.co_nlocals, code.co_stacksize, code.co_flags, code.co_code, code.co_consts, code.co_names, code.co_varnames, f_code.co_filename.encode(), f_code.co_name.encode(), code.co_firstlineno, code.co_lnotab, (), () ) # noinspection PyBroadException try: exec(code, self.tb_frame.f_globals, {}) except: tb = sys.exc_info()[2].tb_next tb_set_next(tb, self.tb_next and self.tb_next.as_traceback()) try: return tb finally: # gevent: don't leak the traceback objects, this # makes our leaktests fail del tb else: raise RuntimeError("Cannot re-create traceback !") # noinspection SpellCheckingInspection