From 98b1a7614084d62ca4453d3161d07d71954b7495 Mon Sep 17 00:00:00 2001 From: Chuan-Heng Hsiao <2970164+chhsiao1981@users.noreply.github.com> Date: Sun, 23 Aug 2026 13:42:10 -0400 Subject: [PATCH] feat: raise_err. Signed-off-by: Chuan-Heng Hsiao <2970164+chhsiao1981@users.noreply.github.com> --- README.md | 94 +++++++++++++++++-- src/with_err/__init__.py | 6 +- src/with_err/raise_err.py | 57 ++++++++++++ tests/test_raise_err.py | 188 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 337 insertions(+), 8 deletions(-) create mode 100644 src/with_err/raise_err.py create mode 100644 tests/test_raise_err.py diff --git a/README.md b/README.md index a072bc1..a031bd4 100644 --- a/README.md +++ b/README.md @@ -25,7 +25,7 @@ json_loads_e = with_err(json.loads) data, err = json_loads_e('{"a": 1}') assert err is None -assert data = {"a": 1} +assert data == {"a": 1} data, err = json_loads_e('{"a": }') assert isinstance(err, json.decoder.JSONDecodeError) @@ -44,7 +44,7 @@ def json_loads_e(a: str | bytes | bytearray): data, err = json_loads_e('{"a": 1}') assert err is None -assert data = {"a": 1} +assert data == {"a": 1} data, err = json_loads_e('{"a": }') assert isinstance(err, json.decoder.JSONDecodeError) @@ -65,7 +65,7 @@ def json_loads_e(a: str | bytes | bytearray): data, err = json_loads_e('{"a": 1}') assert err is None -assert data = {"a": 1} +assert data == {"a": 1} data, err = json_loads_e('{"a": }') assert isinstance(err, json.decoder.JSONDecodeError) @@ -96,7 +96,7 @@ json_loads_e = with_err(json.decoder.JSONDecodeError)(json.loads) data, err = json_loads_e('{"a": 1}') assert err is None -assert data = {"a": 1} +assert data == {"a": 1} data, err = json_loads_e('{"a": }') assert isinstance(err, json.decoder.JSONDecodeError) @@ -113,13 +113,70 @@ json_loads_e = with_err()(json.loads) data, err = json_loads_e('{"a": 1}') assert err is None -assert data = {"a": 1} +assert data == {"a": 1} data, err = json_loads_e('{"a": }') assert isinstance(err, json.decoder.JSONDecodeError) assert data is None ``` +### Async Functions + +```python +from with_err import with_err + +@with_err +async def async_fetch_data(endpoint: str) -> dict[str, str]: + if endpoint == "bad": + raise ValueError("Failed to reach endpoint") + return {"status": "ok"} + +async_fetch_data_e = with_err(async_fetch_data) +res, err = await async_fetch_data_e("bad") +assert isinstance(err, ValueError) +assert res is None +``` + +### Generators + +```python +from with_err import with_err + + +@with_err +def my_stream(): + yield 1 + raise ValueError('invalid') + +for idx, (each, err) in enumerate(my_stream()): + if idx == 0: + assert each == 1 + assert err is None + else: + assert each is None + assert isinstance(err, ValueError) +``` + +### Async Generators + +```python +from with_err import with_err + + +@with_err +async def my_async_stream(): + yield 1 + raise ValueError('invalid') + +async for each, err in my_stream(): + if each == 1: + assert each == 1 + assert err is None + else: + assert each is None + assert isinstance(err, ValueError) +``` + ### Get `err` Traceback Stack ```python @@ -148,7 +205,32 @@ err_stack = get_err_strs(err) err_str = '\n'.join(err_stack) assert isinstance(err, json.decoder.JSONDecodeError) assert len(err_stack) > 0 -assert 'json.decoder.JSONDecodeError: Expecting value:' in err_str +assert re.search(r', line \d+, in json_loads_e', err_str) assert re.search(r'json/__init__.py", line \d+, in loads', err_str) +assert 'json.decoder.JSONDecodeError: Expecting value:' in err_str +``` + +### Raise `err` + +```python +import json +import re +from with_err import with_err, get_err_strs, raise_err + +def json_loads_e(a: str | bytes | bytearray): + return json.loads(a) + +def gen_err(): + data, err = json_loads_e('{"a": }') + return data, raise_err(err) + +data, err = gen_err() +err_stack = get_err_strs(err) +err_str = '\n'.join(err_stack) +assert isinstance(err, json.decoder.JSONDecodeError) +assert len(err_stack) > 0 +assert re.search(r', line \d+, in gen_err', err_str) assert re.search(r', line \d+, in json_loads_e', err_str) +assert re.search(r'json/__init__.py", line \d+, in loads', err_str) +assert 'json.decoder.JSONDecodeError: Expecting value:' in err_str ``` diff --git a/src/with_err/__init__.py b/src/with_err/__init__.py index d0a0f19..82855f3 100644 --- a/src/with_err/__init__.py +++ b/src/with_err/__init__.py @@ -1,7 +1,9 @@ +from .raise_err import raise_err as raise_err from .utils import get_err_strs as get_err_strs from .with_err import with_err as with_err -all = [ +__all__ = [ + 'get_err_strs', + 'raise_err', 'with_err', - 'get_err_strs' ] diff --git a/src/with_err/raise_err.py b/src/with_err/raise_err.py new file mode 100644 index 0000000..31ae855 --- /dev/null +++ b/src/with_err/raise_err.py @@ -0,0 +1,57 @@ +# https://chatgpt.com/c/6a8b3620-9cb0-83ea-baf6-1ab4a31e473b + +import inspect +import types + + +def raise_err(err: Exception | None): + ''' + `raise_err` does not raise the exception, but augments traceback with the caller's frame. + ''' + if err is None: + return + + parent_frame = _get_parent_frame() + if parent_frame is None: + return err + + tb = err.__traceback__ + + combined_tb = types.TracebackType( + tb_next=tb, + tb_frame=parent_frame, + tb_lasti=parent_frame.f_lasti, + tb_lineno=parent_frame.f_lineno, + ) + + return err.with_traceback(combined_tb) + + +def _get_parent_frame(): + frame = _try_get_currentframe() + + # frame as _get_parent_frame. + if frame is None: + return + + # frame.f_back as raise_err + if frame.f_back is None: + return + + # framer.f_back.f_back as caller frame of raise_err. + return frame.f_back.f_back + + +def _try_get_currentframe(): + frame = inspect.currentframe() + if frame is not None: # frame as _try_get_currentframe + return frame.f_back # f_back as _get_parent_frame + + # XXX hack for nuitka. + try: + raise ValueError('none') + except ValueError: + frame = inspect.currentframe() # frame as _try_get_currentframe + if frame is None: + return None + return frame.f_back # f_back as _get_parent_frame diff --git a/tests/test_raise_err.py b/tests/test_raise_err.py new file mode 100644 index 0000000..8b4205d --- /dev/null +++ b/tests/test_raise_err.py @@ -0,0 +1,188 @@ +import inspect +import json +import os +import re +from dataclasses import dataclass +from typing import Self + +import pytest + +from with_err import get_err_strs, raise_err, with_err +from with_err.raise_err import _get_parent_frame + + +@pytest.fixture(scope="module", autouse=True) +def init(): + # setup + yield + # teardown + + +def call_raise_err(): + return my_raise_err() + + +def my_raise_err(): + frame = _get_parent_frame() + return frame + + +def test_get_parent_frame(): + + frame = call_raise_err() + print(f'test_get_parent_frame: filename: {frame.f_code.co_filename} lineno: {frame.f_lineno}') + + assert os.path.basename(frame.f_code.co_filename) == 'test_raise_err.py' + assert frame.f_code.co_name == "call_raise_err" + + +def err_json_loads(): + json_loads_e = with_err(json.loads) + ret, err = json_loads_e('{"test": }') + return ret, raise_err(err) + + +def test_raise_err(): + ret, err = err_json_loads() + err_strs = get_err_strs(err) + err_str = '\n'.join(err_strs) + print(f'test_raise_err: err_str: {err_str}') + + assert ret is None + assert re.search(r'test_raise_err.py", line \d+, in err_json_loads', err_str) + assert re.search(r'with_err.py", line \d+, in wrapper', err_str) + assert re.search(r'json/__init__.py", line \d+, in loads', err_str) + assert re.search(r'json.decoder.JSONDecodeError: Expecting value: line 1 column 10', err_str) + + +def ok_json_loads(): + json_loads_e = with_err(json.loads) + ret, err = json_loads_e('{"test": 1}') + return ret, raise_err(err) + + +def test_raise_err2(): + ret, err = ok_json_loads() + assert err is None + assert ret == {"test": 1} + + +def mock_currentframe_none(): + return + + +@pytest.fixture(scope="function") +def mock_currentframe(): + orig_currentframe = inspect.currentframe + inspect.currentframe = mock_currentframe_none + # setup + yield + inspect.currentframe = orig_currentframe + # teardown + + +def test_raise_err_mock_currentframe(mock_currentframe): + ret, err = err_json_loads() + err_strs = get_err_strs(err) + err_str = '\n'.join(err_strs) + print(f'test_raise_err: err_str: {err_str}') + + assert ret is None + assert not re.search(r'test_raise_err.py", line \d+, in err_json_loads', err_str) + assert re.search(r'with_err.py", line \d+, in wrapper', err_str) + assert re.search(r'json/__init__.py", line \d+, in loads', err_str) + assert re.search(r'json.decoder.JSONDecodeError: Expecting value: line 1 column 10', err_str) + + +@dataclass +class MockFrame: + f_back: Self | None = None + + +def mock_currentframe_none2(): + return MockFrame() + + +@pytest.fixture(scope="function") +def mock_currentframe2(): + orig_currentframe = inspect.currentframe + inspect.currentframe = mock_currentframe_none2 + # setup + yield + inspect.currentframe = orig_currentframe + # teardown + + +def test_raise_err_mock_currentframe2(mock_currentframe2): + ret, err = err_json_loads() + err_strs = get_err_strs(err) + err_str = '\n'.join(err_strs) + print(f'test_raise_err: err_str: {err_str}') + + assert ret is None + assert not re.search(r'test_raise_err.py", line \d+, in err_json_loads', err_str) + assert re.search(r'with_err.py", line \d+, in wrapper', err_str) + assert re.search(r'json/__init__.py", line \d+, in loads', err_str) + assert re.search(r'json.decoder.JSONDecodeError: Expecting value: line 1 column 10', err_str) + + +def mock_currentframe_none3(): + return MockFrame(f_back=MockFrame(f_back=None)) + + +@pytest.fixture(scope="function") +def mock_currentframe3(): + orig_currentframe = inspect.currentframe + inspect.currentframe = mock_currentframe_none3 + # setup + yield + inspect.currentframe = orig_currentframe + # teardown + + +def test_raise_err_mock_currentframe3(mock_currentframe3): + ret, err = err_json_loads() + err_strs = get_err_strs(err) + err_str = '\n'.join(err_strs) + print(f'test_raise_err: err_str: {err_str}') + + assert ret is None + assert not re.search(r'test_raise_err.py", line \d+, in err_json_loads', err_str) + assert re.search(r'with_err.py", line \d+, in wrapper', err_str) + assert re.search(r'json/__init__.py", line \d+, in loads', err_str) + assert re.search(r'json.decoder.JSONDecodeError: Expecting value: line 1 column 10', err_str) + + +count_mock_currentframe_none4 = 0 + + +def mock_currentframe_none4(): + global count_mock_currentframe_none4 + if count_mock_currentframe_none4 == 0: + count_mock_currentframe_none4 += 1 + return None + + return MockFrame(f_back=MockFrame(f_back=None)) + + +@pytest.fixture(scope="function") +def mock_currentframe4(): + orig_currentframe = inspect.currentframe + inspect.currentframe = mock_currentframe_none4 + # setup + yield + inspect.currentframe = orig_currentframe + # teardown + + +def test_raise_err_mock_currentframe4(mock_currentframe4): + ret, err = err_json_loads() + err_strs = get_err_strs(err) + err_str = '\n'.join(err_strs) + print(f'test_raise_err: err_str: {err_str}') + + assert ret is None + assert not re.search(r'test_raise_err.py", line \d+, in err_json_loads', err_str) + assert re.search(r'with_err.py", line \d+, in wrapper', err_str) + assert re.search(r'json/__init__.py", line \d+, in loads', err_str) + assert re.search(r'json.decoder.JSONDecodeError: Expecting value: line 1 column 10', err_str)