跟踪记录序列化库。
项目描述
异常和跟踪记录的序列化库。
自由软件:BSD许可
它可以让你
Pickle 跟踪记录并在不同进程中引发带有序列化跟踪记录的异常。这在使用多进程运行代码时提供了更好的错误处理(想象一下multiprocessing、billiard、futures、celery等)。
从字符串创建跟踪记录对象(from_string 方法)。不使用序列化。
将跟踪记录序列化为/从普通字典(from_dict 和 to_dict 方法)。不使用序列化。
引发从上述来源创建的跟踪记录。
将异常与其跟踪记录和异常链一起序列化(raise ... from ...)(仅限Python 3)
再次强调,使用pickle支持完全是可选的。如果你决定使用pickle支持,那么安全问题是完全由你自己负责。
安装
pip install tblib
文档
腌制回溯
注意:输出的回溯对象已移除了一些属性(如变量)。但您可以使用这些回溯引发异常或打印它们 - 这应该覆盖99%的使用场景。
>>> from tblib import pickling_support
>>> pickling_support.install()
>>> import pickle, sys
>>> def inner_0():
... raise Exception('fail')
...
>>> def inner_1():
... inner_0()
...
>>> def inner_2():
... inner_1()
...
>>> try:
... inner_2()
... except:
... s1 = pickle.dumps(sys.exc_info())
...
>>> len(s1) > 1
True
>>> try:
... inner_2()
... except:
... s2 = pickle.dumps(sys.exc_info(), protocol=pickle.HIGHEST_PROTOCOL)
...
>>> len(s2) > 1
True
>>> try:
... import cPickle
... except ImportError:
... import pickle as cPickle
>>> try:
... inner_2()
... except:
... s3 = cPickle.dumps(sys.exc_info(), protocol=pickle.HIGHEST_PROTOCOL)
...
>>> len(s3) > 1
True
反腌制回溯
>>> pickle.loads(s1)
(<...Exception'>, Exception('fail'...), <traceback object at ...>)
>>> pickle.loads(s2)
(<...Exception'>, Exception('fail'...), <traceback object at ...>)
>>> pickle.loads(s3)
(<...Exception'>, Exception('fail'...), <traceback object at ...>)
引发
>>> from six import reraise
>>> reraise(*pickle.loads(s1))
Traceback (most recent call last):
...
File "<doctest README.rst[14]>", line 1, in <module>
reraise(*pickle.loads(s2))
File "<doctest README.rst[8]>", line 2, in <module>
inner_2()
File "<doctest README.rst[5]>", line 2, in inner_2
inner_1()
File "<doctest README.rst[4]>", line 2, in inner_1
inner_0()
File "<doctest README.rst[3]>", line 2, in inner_0
raise Exception('fail')
Exception: fail
>>> reraise(*pickle.loads(s2))
Traceback (most recent call last):
...
File "<doctest README.rst[14]>", line 1, in <module>
reraise(*pickle.loads(s2))
File "<doctest README.rst[8]>", line 2, in <module>
inner_2()
File "<doctest README.rst[5]>", line 2, in inner_2
inner_1()
File "<doctest README.rst[4]>", line 2, in inner_1
inner_0()
File "<doctest README.rst[3]>", line 2, in inner_0
raise Exception('fail')
Exception: fail
>>> reraise(*pickle.loads(s3))
Traceback (most recent call last):
...
File "<doctest README.rst[14]>", line 1, in <module>
reraise(*pickle.loads(s2))
File "<doctest README.rst[8]>", line 2, in <module>
inner_2()
File "<doctest README.rst[5]>", line 2, in inner_2
inner_1()
File "<doctest README.rst[4]>", line 2, in inner_1
inner_0()
File "<doctest README.rst[3]>", line 2, in inner_0
raise Exception('fail')
Exception: fail
将异常及其回溯和链(仅限Python 3)一起腌制
>>> try: # doctest: +SKIP
... try:
... 1 / 0
... except Exception as e:
... raise Exception("foo") from e
... except Exception as e:
... s = pickle.dumps(e)
>>> raise pickle.loads(s) # doctest: +SKIP
Traceback (most recent call last):
File "<doctest README.rst[16]>", line 3, in <module>
1 / 0
ZeroDivisionError: division by zero
The above exception was the direct cause of the following exception:
Traceback (most recent call last):
File "<doctest README.rst[17]>", line 1, in <module>
raise pickle.loads(s)
File "<doctest README.rst[16]>", line 5, in <module>
raise Exception("foo") from e
Exception: foo
在调用pickling_support.install()之后定义的BaseException子类将不会保留它们的回溯和异常链的腌制。为了覆盖自定义异常,有三个选项
将@pickling_support.install用作每个自定义异常的装饰器
>>> from tblib import pickling_support >>> # Declare all imports of your package's dependencies >>> import numpy # doctest: +SKIP >>> pickling_support.install() # install for all modules imported so far >>> @pickling_support.install ... class CustomError(Exception): ... passCustomError的最终子类需要再次进行装饰。
在所有模块都已导入且所有异常子类都已声明后调用pickling_support.install()
>>> # Declare all imports of your package's dependencies >>> import numpy # doctest: +SKIP >>> from tblib import pickling_support >>> # Declare your own custom Exceptions >>> class CustomError(Exception): ... pass >>> # Finally, install tblib >>> pickling_support.install()在选择性地在异常实例被腌制之前安装tblib
pickling_support.install(<Exception instance>, [Exception instance], ...)上述操作将为所有列出的异常以及它们的异常链中的任何其他异常安装tblib的腌制
例如,可以编写一个包装器,与ProcessPoolExecutor、Dask.distributed或类似库一起使用
>>> from tblib import pickling_support >>> def wrapper(func, *args, **kwargs): ... try: ... return func(*args, **kwargs) ... except Exception as e: ... pickling_support.install(e) ... raise
如果我们有一个局部堆栈,它会正确显示吗?
是的,它确实如此
>>> exc_info = pickle.loads(s3)
>>> def local_0():
... reraise(*exc_info)
...
>>> def local_1():
... local_0()
...
>>> def local_2():
... local_1()
...
>>> local_2()
Traceback (most recent call last):
File "...doctest.py", line ..., in __run
compileflags, 1) in test.globs
File "<doctest README.rst[24]>", line 1, in <module>
local_2()
File "<doctest README.rst[23]>", line 2, in local_2
local_1()
File "<doctest README.rst[22]>", line 2, in local_1
local_0()
File "<doctest README.rst[21]>", line 2, in local_0
reraise(*exc_info)
File "<doctest README.rst[11]>", line 2, in <module>
inner_2()
File "<doctest README.rst[5]>", line 2, in inner_2
inner_1()
File "<doctest README.rst[4]>", line 2, in inner_1
inner_0()
File "<doctest README.rst[3]>", line 2, in inner_0
raise Exception('fail')
Exception: fail
它还支持更多复杂的情况
像有语法错误的回溯一样
>>> from tblib import Traceback
>>> from examples import bad_syntax
>>> try:
... bad_syntax()
... except:
... et, ev, tb = sys.exc_info()
... tb = Traceback(tb)
...
>>> reraise(et, ev, tb.as_traceback())
Traceback (most recent call last):
...
File "<doctest README.rst[58]>", line 1, in <module>
reraise(et, ev, tb.as_traceback())
File "<doctest README.rst[57]>", line 2, in <module>
bad_syntax()
File "...tests...examples.py", line 18, in bad_syntax
import badsyntax
File "...tests...badsyntax.py", line 5
is very bad
^
SyntaxError: invalid syntax
或其他导入失败
>>> from examples import bad_module
>>> try:
... bad_module()
... except:
... et, ev, tb = sys.exc_info()
... tb = Traceback(tb)
...
>>> reraise(et, ev, tb.as_traceback())
Traceback (most recent call last):
...
File "<doctest README.rst[61]>", line 1, in <module>
reraise(et, ev, tb.as_traceback())
File "<doctest README.rst[60]>", line 2, in <module>
bad_module()
File "...tests...examples.py", line 23, in bad_module
import badmodule
File "...tests...badmodule.py", line 3, in <module>
raise Exception("boom!")
Exception: boom!
或由于超过递归限制而导致的回溯(在这里我们强制类型和值在平台上保持一致性)
>>> def f(): f()
>>> try:
... f()
... except RuntimeError:
... et, ev, tb = sys.exc_info()
... tb = Traceback(tb)
...
>>> reraise(RuntimeError, RuntimeError("maximum recursion depth exceeded"), tb.as_traceback())
Traceback (most recent call last):
...
File "<doctest README.rst[32]>", line 1, in f
def f(): f()
File "<doctest README.rst[32]>", line 1, in f
def f(): f()
File "<doctest README.rst[32]>", line 1, in f
def f(): f()
...
RuntimeError: maximum recursion depth exceeded
参考
tblib.Traceback
它被pickling_support使用。如果您想获得更多灵活性,也可以使用它
>>> from tblib import Traceback
>>> try:
... inner_2()
... except:
... et, ev, tb = sys.exc_info()
... tb = Traceback(tb)
...
>>> reraise(et, ev, tb.as_traceback())
Traceback (most recent call last):
...
File "<doctest README.rst[21]>", line 6, in <module>
reraise(et, ev, tb.as_traceback())
File "<doctest README.rst[21]>", line 2, in <module>
inner_2()
File "<doctest README.rst[5]>", line 2, in inner_2
inner_1()
File "<doctest README.rst[4]>", line 2, in inner_1
inner_0()
File "<doctest README.rst[3]>", line 2, in inner_0
raise Exception('fail')
Exception: fail
tblib.Traceback.to_dict
您可以使用to_dict方法和from_dict类方法将回溯转换为字典并从字典中转换回来,该字典可以由stdlib json.JSONDecoder序列化
>>> import json
>>> from pprint import pprint
>>> try:
... inner_2()
... except:
... et, ev, tb = sys.exc_info()
... tb = Traceback(tb)
... tb_dict = tb.to_dict()
... pprint(tb_dict)
{'tb_frame': {'f_code': {'co_filename': '<doctest README.rst[...]>',
'co_name': '<module>'},
'f_globals': {'__name__': '__main__'},
'f_lineno': 5},
'tb_lineno': 2,
'tb_next': {'tb_frame': {'f_code': {'co_filename': ...,
'co_name': 'inner_2'},
'f_globals': {'__name__': '__main__'},
'f_lineno': 2},
'tb_lineno': 2,
'tb_next': {'tb_frame': {'f_code': {'co_filename': ...,
'co_name': 'inner_1'},
'f_globals': {'__name__': '__main__'},
'f_lineno': 2},
'tb_lineno': 2,
'tb_next': {'tb_frame': {'f_code': {'co_filename': ...,
'co_name': 'inner_0'},
'f_globals': {'__name__': '__main__'},
'f_lineno': 2},
'tb_lineno': 2,
'tb_next': None}}}}
tblib.Traceback.from_dict
基于上一个示例
>>> tb_json = json.dumps(tb_dict)
>>> tb = Traceback.from_dict(json.loads(tb_json))
>>> reraise(et, ev, tb.as_traceback())
Traceback (most recent call last):
...
File "<doctest README.rst[21]>", line 6, in <module>
reraise(et, ev, tb.as_traceback())
File "<doctest README.rst[21]>", line 2, in <module>
inner_2()
File "<doctest README.rst[5]>", line 2, in inner_2
inner_1()
File "<doctest README.rst[4]>", line 2, in inner_1
inner_0()
File "<doctest README.rst[3]>", line 2, in inner_0
raise Exception('fail')
Exception: fail
tblib.Traceback.from_string
>>> tb = Traceback.from_string("""
... File "skipped.py", line 123, in func_123
... Traceback (most recent call last):
... File "tests/examples.py", line 2, in func_a
... func_b()
... File "tests/examples.py", line 6, in func_b
... func_c()
... File "tests/examples.py", line 10, in func_c
... func_d()
... File "tests/examples.py", line 14, in func_d
... Doesn't: matter
... """)
>>> reraise(et, ev, tb.as_traceback())
Traceback (most recent call last):
...
File "<doctest README.rst[42]>", line 6, in <module>
reraise(et, ev, tb.as_traceback())
File "...examples.py", line 2, in func_a
func_b()
File "...examples.py", line 6, in func_b
func_c()
File "...examples.py", line 10, in func_c
func_d()
File "...examples.py", line 14, in func_d
raise Exception("Guessing time !")
Exception: fail
如果使用strict=False选项,则解析会略微宽松一些
>>> tb = Traceback.from_string("""
... File "bogus.py", line 123, in bogus
... Traceback (most recent call last):
... File "tests/examples.py", line 2, in func_a
... func_b()
... File "tests/examples.py", line 6, in func_b
... func_c()
... File "tests/examples.py", line 10, in func_c
... func_d()
... File "tests/examples.py", line 14, in func_d
... Doesn't: matter
... """, strict=False)
>>> reraise(et, ev, tb.as_traceback())
Traceback (most recent call last):
...
File "<doctest README.rst[42]>", line 6, in <module>
reraise(et, ev, tb.as_traceback())
File "bogus.py", line 123, in bogus
File "...examples.py", line 2, in func_a
func_b()
File "...examples.py", line 6, in func_b
func_c()
File "...examples.py", line 10, in func_c
func_d()
File "...examples.py", line 14, in func_d
raise Exception("Guessing time !")
Exception: fail
tblib.decorators.return_error
>>> from tblib.decorators import return_error
>>> inner_2r = return_error(inner_2)
>>> e = inner_2r()
>>> e
<tblib.decorators.Error object at ...>
>>> e.reraise()
Traceback (most recent call last):
...
File "<doctest README.rst[26]>", line 1, in <module>
e.reraise()
File "...tblib...decorators.py", line 19, in reraise
reraise(self.exc_type, self.exc_value, self.traceback)
File "...tblib...decorators.py", line 25, in return_exceptions_wrapper
return func(*args, **kwargs)
File "<doctest README.rst[5]>", line 2, in inner_2
inner_1()
File "<doctest README.rst[4]>", line 2, in inner_1
inner_0()
File "<doctest README.rst[3]>", line 2, in inner_0
raise Exception('fail')
Exception: fail
这有什么用?想象一下您正在这样使用multiprocessing
# Note that Python 3.4 and later will show the remote traceback (but as a string sadly) so we skip testing this.
>>> import traceback
>>> from multiprocessing import Pool
>>> from examples import func_a
>>> pool = Pool() # doctest: +SKIP
>>> try: # doctest: +SKIP
... for i in pool.map(func_a, range(5)):
... print(i)
... except:
... print(traceback.format_exc())
...
Traceback (most recent call last):
File "<doctest README.rst[...]>", line 2, in <module>
for i in pool.map(func_a, range(5)):
File "...multiprocessing...pool.py", line ..., in map
...
File "...multiprocessing...pool.py", line ..., in get
...
Exception: Guessing time !
<BLANKLINE>
>>> pool.terminate() # doctest: +SKIP
这不太有用,对吗?让我们解决这个问题
>>> from tblib.decorators import apply_with_return_error, Error
>>> from itertools import repeat
>>> pool = Pool()
>>> try:
... for i in pool.map(apply_with_return_error, zip(repeat(func_a), range(5))):
... if isinstance(i, Error):
... i.reraise()
... else:
... print(i)
... except:
... print(traceback.format_exc())
...
Traceback (most recent call last):
File "<doctest README.rst[...]>", line 4, in <module>
i.reraise()
File "...tblib...decorators.py", line ..., in reraise
reraise(self.exc_type, self.exc_value, self.traceback)
File "...tblib...decorators.py", line ..., in return_exceptions_wrapper
return func(*args, **kwargs)
File "...tblib...decorators.py", line ..., in apply_with_return_error
return args[0](*args[1:])
File "...examples.py", line 2, in func_a
func_b()
File "...examples.py", line 6, in func_b
func_c()
File "...examples.py", line 10, in func_c
func_d()
File "...examples.py", line 14, in func_d
raise Exception("Guessing time !")
Exception: Guessing time !
<BLANKLINE>
>>> pool.terminate()
好多了!
如果我们有一个局部调用堆栈?
>>> def local_0():
... pool = Pool()
... try:
... for i in pool.map(apply_with_return_error, zip(repeat(func_a), range(5))):
... if isinstance(i, Error):
... i.reraise()
... else:
... print(i)
... finally:
... pool.close()
...
>>> def local_1():
... local_0()
...
>>> def local_2():
... local_1()
...
>>> try:
... local_2()
... except:
... print(traceback.format_exc())
Traceback (most recent call last):
File "<doctest README.rst[...]>", line 2, in <module>
local_2()
File "<doctest README.rst[...]>", line 2, in local_2
local_1()
File "<doctest README.rst[...]>", line 2, in local_1
local_0()
File "<doctest README.rst[...]>", line 6, in local_0
i.reraise()
File "...tblib...decorators.py", line 20, in reraise
reraise(self.exc_type, self.exc_value, self.traceback)
File "...tblib...decorators.py", line 27, in return_exceptions_wrapper
return func(*args, **kwargs)
File "...tblib...decorators.py", line 47, in apply_with_return_error
return args[0](*args[1:])
File "...tests...examples.py", line 2, in func_a
func_b()
File "...tests...examples.py", line 6, in func_b
func_c()
File "...tests...examples.py", line 10, in func_c
func_d()
File "...tests...examples.py", line 14, in func_d
raise Exception("Guessing time !")
Exception: Guessing time !
<BLANKLINE>
其他奇怪的东西
清除回溯有效(Python 3.4及更高版本)
>>> tb = Traceback.from_string("""
... File "skipped.py", line 123, in func_123
... Traceback (most recent call last):
... File "tests/examples.py", line 2, in func_a
... func_b()
... File "tests/examples.py", line 6, in func_b
... func_c()
... File "tests/examples.py", line 10, in func_c
... func_d()
... File "tests/examples.py", line 14, in func_d
... Doesn't: matter
... """)
>>> import traceback, sys
>>> if sys.version_info > (3, 4):
... traceback.clear_frames(tb)
致谢
mitsuhiko/jinja2找到了创建回溯对象的方法。
更改日志
3.0.0 (2023-10-22)
2.0.0 (2023-06-22)
删除了对旧版Python(2.7和3.6)的支持,并在测试网格中添加了Python 3.11。
进行了一些清理和重构(主要来自ruff)。
1.7.0 (2020-07-24)
为Frame和Code对象添加了更多属性以实现pytest兼容性。由Ivanq在#58中贡献。
1.6.0 (2019-12-07)
在腌制异常时,也腌制其回溯和异常链(raise ... from ...)。由Guido Imperiale在#53中贡献。
1.5.0 (2019-10-23)
添加了对Python 3.8的支持。由Victor Stinner在#42中贡献。
移除了对已停用的Python 3.4的支持。
少量CI改进和修复。
1.4.0 (2019-05-02)
移除了对已停用的Python 3.3的支持。
修复了Python 3.7的测试。由Elliott Sales de Andrade在#36中贡献。
修复了与Twisted(期望co_code属性)的兼容性问题。
1.3.2 (2017-04-09)
添加了对PyPy3.5-5.7.1-beta的支持。之前可能会引发AttributeError: 'Frame' object has no attribute 'clear'错误。请参见PyPy问题#2532。
1.3.1 (2017-03-27)
修复了超出递归限制导致的跟踪堆栈处理。修复了#15。
1.3.0 (2016-03-08)
添加了Traceback.from_string。
1.2.0 (2015-12-18)
1.1.0 (2015-07-27)
添加了对Python 2.6的支持。由Arcadiy Ivanov在#8中贡献。
1.0.0 (2015-03-30)
在Tracebacks中添加了to_dict方法和from_dict类方法。由beckjake在#5中贡献。
项目详情
下载文件
下载适合您平台的文件。如果您不确定选择哪个,请了解有关安装包的更多信息。
源代码分发
构建分发
tblib-3.0.0.tar.gz的哈希值
| 算法 | 哈希摘要 | |
|---|---|---|
| SHA256 | 93622790a0a29e04f0346458face1e144dc4d32f493714c6c3dff82a4adb77e6 |
|
| MD5 | 7532a2d525bd0d5d4a78db5dafe4823f |
|
| BLAKE2b-256 | 1adf4f2cd7eaa6d41a7994d46527349569d46e34d9cdd07590b5c5b0dcf53de3 |
tblib-3.0.0-py3-none-any.whl的哈希值
| 算法 | 哈希摘要 | |
|---|---|---|
| SHA256 | 80a6c77e59b55e83911e1e607c649836a69c103963c5f28a46cbeef44acf8129 |
|
| MD5 | a20cd4b91b1d1b8a58bf5bdc495dfda5 |
|
| BLAKE2b-256 | 9b87ce70db7cae60e67851eb94e1a2127d4abb573d3866d2efd302ceb0d4d2a5 |