pyodide-tblib · PyPI · Python 包索引

回溯序列化库。

这些详情尚未由PyPI验证

项目链接

项目描述

异常和回溯的序列化库。

自由软件: BSD许可证

它可以让你

Pickle 回溯和在不同的进程中抛出带有序列化回溯的异常。这在运行跨多个进程的代码时提供了更好的错误处理（想象一下multiprocessing、billiard、futures、celery等）。
从字符串（from_string 方法）创建回溯对象。（不使用序列化）
将回溯序列化为/反序列化为普通字典（from_dict 和 to_dict 方法）。（不使用序列化）
抛出从上述来源创建的回溯。
将异常及其回溯和异常链（raise ... from ...）（仅Python 3）一起序列化。

再次注意，使用序列化支持是完全可选的。如果您决定使用序列化支持，那么您将完全负责安全问题。

安装

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):
...     pass

CustomError的最终子类需要再次进行装饰。

在所有模块已导入且所有异常子类已声明后调用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

这有什么用？想象一下您正在使用多进程方式如下

# 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为创建跟踪对象找到了一种方法。

更改日志

1.7.1 (2022-08-05)

添加了对本地中__tracebackhide__变量的序列化，以与pytest堆栈帧抑制兼容。

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的兼容性问题（twisted.python.failure.Failure期望一个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)

修复了处理生成器和其他内部改进和优化的跟踪异常。由DRayX在#10和#11中贡献。

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中贡献。

项目详情

这些详情尚未由PyPI验证

项目链接

发布历史发布通知 | RSS源

本版本

1.7.1

2022年8月5日

下载文件

下载适用于您平台的文件。如果您不确定选择哪个，请了解更多关于安装包的信息。

源代码分发

pyodide-tblib-1.7.1.tar.gz (33.3 kB 查看哈希)

上传时间 2022年8月5日 源

构建分发

pyodide_tblib-1.7.1-py3-none-any.whl (11.6 kB 查看哈希)

上传时间 2022年8月5日 Python 3

pyodide-tblib-1.7.1.tar.gz的哈希

pyodide-tblib-1.7.1.tar.gz的哈希
算法	哈希摘要
SHA256	`28a94c15350e483501a61cd8795bcbe531e5e1606a06835d2da9a8156b506f22`
MD5	`6dbc752f33df7633df26f335e724fac6`
BLAKE2b-256	`f29013655f38fd67a6a6e1628e10a3e66de3e06232e3df5103f197f2a9a0bb64`

pyodide_tblib-1.7.1-py3-none-any.whl的哈希

pyodide_tblib-1.7.1-py3-none-any.whl的哈希
算法	哈希摘要
SHA256	`6e21535b5be695b4ee2751c5bde21eb5a003ed74aabafaecdca8ef5b3cfa3ea1`
MD5	`354895e8855ab82f793d7db74171a51d`
BLAKE2b-256	`6ff084df460611f1d1c9c348fd4ac2295178083e54962b1be94d571602fafb4c`