How to achieve "resumption semantics" for Python exceptions?

Expurple 0
Viewed 65

I have a validator class with a method that performs multiple checks and may raise different exceptions:

class Validator:
    def validate(something) -> None:
        if a:
            raise ErrorA()
        if b:
            raise ErrorB()
        if c:
            raise ErrorC()

There's a place in the outside (caller) code where I want to customize its behaviour and prevent ErrorB from being raised, without preventing ErrorC. Something like resumption semantics would be useful here. Hovewer, I haven't found a good way to achieve this.

To clarify: I have the control over Validator source code, but prefer to preserve its existing interface as much as possible.

Some possible solutions that I've considered:

  1. The obvious

    try:
    validator.validate(something)
except ErrorB:
    ...

    is no good because it also suppresses ErrorC in cases where both ErrorB and ErrorC should be raised.

  2. Copy-paste the method and remove the check:

    # In the caller module

class CustomValidator(Validator):
    def validate(something) -> None:
        if a:
            raise ErrorA()
        if c:
            raise ErrorC()

    Duplicating the logic for a and c is a bad idea and will lead to bugs if Validator changes.

  3. Split the method into separate checks:

    class Validator:
    def validate(something) -> None:
        self.validate_a(something)
        self.validate_b(something)
        self.validate_c(something)

    def validate_a(something) -> None:
        if a:
            raise ErrorA()

    def validate_b(something) -> None:
        if b:
            raise ErrorB()

    def validate_c(something) -> None:
        if c:
            raise ErrorC()

# In the caller module

class CustomValidator(Validator):
    def validate(something) -> None:
        super().validate_a(something)
        super().validate_c(something)

    This is just a slightly better copy-paste. If some validate_d() is added later, we have a bug in CustomValidator.

  4. Add some suppression logic by hand:

    class Validator:
    def validate(something, *, suppress: list[Type[Exception]] = []) -> None:
        if a:
            self._raise(ErrorA(), suppress)
        if b:
            self._raise(ErrorB(), suppress)
        if c:
            self._raise(ErrorC(), suppress)

    def _raise(self, e: Exception, suppress: list[Type[Exception]]) -> None:
        with contextlib.suppress(*suppress):
            raise e

    This is what I'm leaning towards at the moment. There's a new optional parameter and the raise syntax becomes kinda ugly, but this is an acceptable cost.

  5. Add flags that disable some checks:

    class Validator:
    def validate(something, *, check_a: bool = True,
                 check_b: bool = True, check_c: bool = True) -> None:
        if check_a and a:
            raise ErrorA()
        if check_b and b:
            raise ErrorB()       
        if check_c and c:
            raise ErrorC()

    This is good, because it allows to granually control different checks even if they raise the same exception.

    However, it feels verbose and will require additional maintainance as Validator changes. I actually have more than three checks there.

  6. Yield exceptions by value:

    class Validator:
    def validate(something) -> Iterator[Exception]:
        if a:
            yield ErrorA()
        if b:
            yield ErrorB()
        if c:
            yield ErrorC()

    This is bad, because it's a breaking change for existing callers and it makes propagating the exception (the typical use) way more verbose:

    # Instead of
# validator.validate(something)

e = next(validator.validate(something), None)
if e is not None:
    raise e

    Even if we keep everything backwards-compatible

    class Validator:
    def validate(something) -> None:
        e = next(self.iter_errors(something), None)
        if e is not None:
            raise e

    def iter_errors(something) -> Iterator[Exception]:
        if a:
            yield ErrorA()
        if b:
            yield ErrorB()
        if c:
            yield ErrorC()

    The new suppressing caller still needs to write all this code:

    exceptions = validator.iter_errors(something)
e = next(exceptions, None)
if isinstance(e, ErrorB):
    # Skip ErrorB, don't raise it.
    e = next(exceptions, None)
if e is not None:
    raise e

    Compared to the previous two options:

    validator.validate(something, suppress=[ErrorB])

    validator.validate(something, check_b=False)
exception python
4 Answers

With bare exceptions you are looking at the wrong tool for the job. In Python, to raise an exception means that execution hits an exceptional case in which resuming is not possible. Terminating the broken execution is an express purpose of exceptions.

Execution Model: 4.3. Exceptions

Python uses the “termination” model of error handling: an exception handler can find out what happened and continue execution at an outer level, but it cannot repair the cause of the error and retry the failing operation (except by re-entering the offending piece of code from the top).

To get resumption semantics for exception handling, you can look at the generic tools for either resumption or for handling.

Resumption: Coroutines

Python's resumption model are coroutines: yield coroutine-generators or async coroutines both allow to pause and explicitly resume execution.

def validate(something) -> Iterator[Exception]:
    if a:
        yield ErrorA()
    if b:
        yield ErrorB()
    if c:
        yield ErrorC()

It is important to distinguish between send-style "proper" coroutines and iterator-style "generator" coroutines. As long as no value must be sent into the coroutine, it is functionally equivalent to an iterator. Python has good inbuilt support for working with iterators:

for e in validator.iter_errors(something):
    if isinstance(e, ErrorB):
        continue  # continue even if ErrorB happens
    raise e

Similarly, one could filter the iterator or use comprehensions. Iterators easily compose and gracefully terminate, making them suitable for iterating exception cases.

Effect Handling

Exception handling is just the common use case for the more generic effect handling. While Python has no builtin effect handling support, simple handlers that address only the origin or sink of an effect can be modelled just as functions:

def default_handler(failure: BaseException):
    raise failure

def validate(something, failure_handler = default_handler) -> None:
    if a:
        failure_handler(ErrorA())
    if b:
        failure_handler(ErrorB())
    if c:
        failure_handler(ErrorC())

This allows the caller to change the effect handling by supplying a different handler.

def ignore_b_handler(failure: BaseException):
    if not isinstance(failure, ErrorB):
        raise failure

validate(..., ignore_b_handler)

This might seem familiar to dependency inversion and is in fact related to it.

There are various stages of buying into effect handling, and it is possible to reproduce much if not all features via classes. Aside from technical functionality, one can implement ambient effect handlers (similar to how try "connects" to raise automatically) via thread local or context-local variables.

MisterMiyagi0

Another spin on option 4:

class Validator:
    def validate(something, *, suppress: tuple[Type[Exception], ...] = ()) -> None:
        def _raise(e: Exception) -> None:
            with contextlib.suppress(*suppress):
                raise e

        if a:
            _raise(ErrorA())
        if b:
            _raise(ErrorB())
        if c:
            _raise(ErrorC())

The raise part looks cleaner now. But I'm not sure if I'll use this, because I actually have multiple validate()-like methods, and each would have to define the same closure.

Expurple0

This solution requires some rewriting of the original code to allow for your surppression semantics, but it feels rather clean. You wrap each raise in a context manager that checks a suppression object:

class SuppressionManager:
  def __init__(self, suppress=None):
    self.suppress = suppress or []

  def __enter__(self):
    pass

  def __exit__(self, exc_type, exc_val, exc_tb):
    if exc_type in self.suppress:
      pass
    else:
      raise exc_val from None

Then in your function:

class Validator:
    def validate(something, suppress=None) -> None:
        mgr = SuppressionManager(suppress)
        if a:
            with mgr:
                raise ErrorA()
        if b:
            with mgr:
                raise ErrorB()
        if c:
            with mgr:
                raise ErrorC()

Now noone needs to change anything :) If you don't control the code inside validate, I propose source code inspection, and rewrite the AST to wrap each raise statement, recompile it, and monkey patch it in.

Robin De Schepper0

You can either make a mixin class that uses __init_subclass__ to apply this to the validate method, or, more conventional, use a decorator:

class Validator:
    @suppressed_raise
    def validate(something, *, suppressed_raise) -> None:
        if a:
            suppressed_raise(ErrorA())
        if b:
            suppressed_raise(ErrorB())
        if c:
            suppressed_raise(ErrorC())

import functools
import inspect
import contextlib

def suppressed_raise(f):
 sig = inspect.signature(f)
 @functools.wraps(f)
 def decorated(*args, *, suppress: list[Type[Exception]] = None, **kwargs):
  def raise_(e: Exception) -> None:
   with contextlib.suppress(*(suppress or [])):
    raise e

  bound = sig.bind_partial(*args, **kwargs)
  bound.arguments["suppressed_raise"] = raise_
  return f(**bound.args, **bound.kwargs)

 return decorated

This will rewrite the suppress list, into a ready-to-use suppressed_raise function, so the caller gives a list of exceptions to suppress, and the method can use suppressed_raise to raise their exceptions.

Robin De Schepper0
Related