Feature/weak diversions

.github/workflows/build.yaml

@@ -3,14 +3,39 @@ name: Build library
 on: [ push ]
 
 jobs:
+  mock-build:
+    name: Ensure the build works
+    runs-on: ubuntu-latest
+
+    if: github.ref != 'refs/heads/main'  # Only upload if change was to the main branch
+
+    steps:
+      - uses: actions/checkout@v3
+
+      - name: Setup Python environment `3.10`
+        uses: actions/setup-python@v4
+        with:
+          python-version: "3.10"
+
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r "requirements.dev.txt"
+
+      - name: Build
+        run: python -m build
+
   publish-and-publish:
     name: Upload release to PyPI
     runs-on: ubuntu-latest
+
+    if: github.ref == 'refs/heads/main'  # Only upload if change was to the main branch
     environment:
       name: build
       url: https://pypi.org/p/divert
     permissions:
       id-token: write
+
     steps:
       - uses: actions/checkout@v3
 
@@ -28,5 +53,4 @@ jobs:
         run: python -m build
 
       - name: Publish package distributions to PyPI
-        if: github.ref == 'refs/heads/main'  # Only upload if change was to the main branch
         uses: pypa/gh-action-pypi-publish@release/v1

.version

@@ -1 +1 @@
-0.1.0
+0.2.0

changelog.md

@@ -1,5 +1,10 @@
 # Flow Control
 
+## 0.2.0
+
+* feature: support for weak diversion variants
+  * a weak variant allows it to be caught by *normal* try-catches when using `except Exception:`
+
 ## 0.1.0
 
 * feature: basic flow edge and jump to edge

divert/core.py

@@ -1,9 +1,11 @@
+from __future__ import annotations
+
 from typing import TypeVar, Generic, Any
 
 Payload = TypeVar("Payload")
 
 
-class Diversion(Exception):
+class Diversion(BaseException):
     """Diversion from the normal execution flow."""
 
     def __init__(self, jumps: int = 1) -> None:
@@ -16,6 +18,19 @@ def raise_again(self, *_: Any) -> bool:
         self._jumps_remaining -= 1
         return self._jumps_remaining > 0
 
+    @classmethod
+    def make_weak(cls, *args: Any, **kwargs: Any) -> Diversion:
+        """Make weak variant of the diversion class.
+
+        By dynamically forming it provides support for generic extensions of the base diversions.
+        The minor performance hit is acceptable vs. having to maintain `WeakDiversion` instances for each extension.
+        """
+
+        class WeakDiversion(cls, Exception):
+            """Weak variation of the diversion, so it can be caught by *normal* user exceptions."""
+
+        return WeakDiversion(*args, **kwargs)
+
 
 class PayloadDiversion(Diversion, Generic[Payload]):
     def __init__(self, payload: Payload, jumps: int = 1) -> None:

divert/flow.py

@@ -11,6 +11,8 @@
 DefaultReturn = TypeVar("DefaultReturn")
 FlowEdgeWrapper = Callable[[Function], OptionalFunction]
 
+DiversionType = TypeVar("DiversionType", bound=Diversion)
+
 
 def custom_flow_edge(default_return: DefaultReturn = None, name: str | None = None) -> FlowEdgeWrapper:
     """Flow edge that allows the default return value to be specified."""
@@ -41,11 +43,16 @@ def flow_edge(function: Function) -> OptionalFunction:
     return custom_flow_edge(default_return=None)(function)
 
 
-def divert(number_of_edges: int = 1) -> None:
+def finalize_diversion(diversion: type[DiversionType], weak: bool) -> type[DiversionType]:
+    """Finalize the diversion class."""
+    return Diversion.make_weak if weak else diversion
+
+
+def divert(number_of_edges: int = 1, weak: bool = False) -> None:
     """Jump to the nearest execution flow edge."""
-    raise Diversion(number_of_edges)
+    raise finalize_diversion(Diversion, weak)(number_of_edges)
 
 
-def divert_payload(payload: Any, number_of_edges: int = 1) -> None:
+def divert_payload(payload: Any, number_of_edges: int = 1, weak: bool = False) -> None:
     """Divert the payload to the nearest execution flow edge."""
-    raise PayloadDiversion(payload, number_of_edges)
+    raise finalize_diversion(PayloadDiversion, weak)(payload, number_of_edges)

divert/targeted.py

@@ -1,6 +1,7 @@
 from typing import Callable, TypeVar
 
 from divert.core import PayloadDiversion, Payload
+from divert.flow import finalize_diversion
 
 Target = TypeVar("Target", bound=Callable | str | None)
 
@@ -19,6 +20,6 @@ def raise_again(self, edge_function: Callable, name: str | None) -> bool:
         return self._target != target_subject
 
 
-def divert_to_target(target: Target, payload: Payload = None) -> None:
+def divert_to_target(target: Target, payload: Payload = None, weak: bool = False) -> None:
     """Divert the payload to a flow edge specified by the function it wraps or the edge name."""
-    raise TargetedDiversion(target, payload)
+    raise finalize_diversion(TargetedDiversion, weak)(target, payload)

test/test_core.py

@@ -1,6 +1,9 @@
 from pytest import mark
 
 from divert.core import Diversion
+from divert.flow import flow_edge, divert_payload
+
+PAYLOAD = 42
 
 
 @mark.parametrize("jumps", argvalues=range(10))
@@ -14,3 +17,22 @@ def test_flow_exception__raise_again(jumps: int) -> None:
 
         else:
             assert current_jump >= jumps, f"Should have continued jumping at {current_jump}"
+
+
+@flow_edge
+def user_function(weak: bool) -> int | None:
+    """Mocked outer call."""
+    try:
+        divert_payload(PAYLOAD, weak=weak)
+    except Exception:
+        assert weak, "Should not have made it here."
+
+
+@mark.parametrize("weak", (True, False))
+def test_avoids_user_try_catch(weak: bool) -> None:
+    """Ensure the user can't accidentally catch the flow exception."""
+    result = user_function(weak)
+    if weak:
+        assert result is None, "Should have been caught by the user try-except."
+    else:
+        assert result == PAYLOAD, "Should have returned the user payload."