Although throwing excptions for control flow is a controversial topic there are some quite popular examples of using this anti-pattern like C#'s async routines throwing the OperationCanceledException to cancel a task or python throwing the StopIteration to control iterators.
I thought I'll try to use such an exception with my logger decorator package (GitHub). I call it ContinuationError. In general the decorator handles logging of such states as
startedbefore entering a functioncompletedwhen a function successfully executedcanceledwhen a function exited prematurelyfaultedwhen something unexpected occured
The exception supports the canceled state by replacing spammy logging with an error:
Before:
logger.canceled(reason="No luck!")
return 5
After:
raise ContinuationError("No luck!", 5)
It expects a reason for why the cancellation was necessary, optionally a return value if the function is expected to return something and also optionally other arguments that are later rendered into a json-message.
class ContinuationError(Exception):
"""Raise this error to gracefully handle a function cancellation."""
def __new__(cls, *args, **details) -> Any:
instance = super().__new__(cls)
instance.details = details | dict(reason=args[0])
if len(args) > 1:
instance.result = args[1]
return instance
def __init__(self, message: str, result: Optional[Any] = None, **details):
super().__init__(message)
The decorator takes care of handling it by checking whether a result was provided and returns it if necessary. The decorator also provides a lambda for creating started details or another lambda for logging the result.
def telemetry(on_started: Optional[OnStarted] = None, on_completed: Optional[OnCompleted] = None, **kwargs):
"""Provides flow telemetry for the decorated function."""
on_started = on_started or (lambda _: {})
on_completed = on_completed or (lambda _: {})
def factory(decoratee):
@contextlib.contextmanager
def logger_scope() -> Logger:
logger = Logger(
module=inspect.getmodule(decoratee).__name__,
scope=decoratee.__name__,
attachment=kwargs.pop("attachment", None),
parent=_scope.get()
)
token = _scope.set(logger)
try:
yield logger
except Exception:
logger.faulted()
raise
finally:
_scope.reset(token)
def inject_logger(logger: Logger, d: Dict):
""" Injects Logger if required. """
for n, t in inspect.getfullargspec(decoratee).annotations.items():
if t is Logger:
d[n] = logger
def params(*decoratee_args, **decoratee_kwargs) -> Dict[str, Any]:
# Zip arg names and their indexes up to the number of args of the decoratee_args.
arg_pairs = zip(inspect.getfullargspec(decoratee).args, range(len(decoratee_args)))
# Turn arg_pairs into a dictionary and combine it with decoratee_kwargs.
return {t[0]: decoratee_args[t[1]] for t in arg_pairs} | decoratee_kwargs
if asyncio.iscoroutinefunction(decoratee):
@functools.wraps(decoratee)
async def decorator(*decoratee_args, **decoratee_kwargs):
with logger_scope() as scope:
inject_logger(scope, decoratee_kwargs)
scope.started(**on_started(params(*decoratee_args, **decoratee_kwargs)))
try:
result = await decoratee(*decoratee_args, **decoratee_kwargs)
scope.completed(**on_completed(result))
return result
except ContinuationError as e:
if hasattr(e, "result"):
scope.canceled(**(on_completed(e.result) | e.details))
return e.result
else:
scope.canceled(**e.details)
else:
@functools.wraps(decoratee)
def decorator(*decoratee_args, **decoratee_kwargs):
with logger_scope() as scope:
inject_logger(scope, decoratee_kwargs)
scope.started(**on_started(params(*decoratee_args, **decoratee_kwargs)))
try:
result = decoratee(*decoratee_args, **decoratee_kwargs)
scope.completed(**on_completed(result))
return result
except ContinuationError as e:
if hasattr(e, "result"):
scope.canceled(**(on_completed(e.result) | e.details))
return e.result
else:
scope.canceled(**e.details)
decorator.__signature__ = inspect.signature(decoratee)
return decorator
return factory
Later one of the logging APIs checks for the exception and decides whether to log a normal message or an actual error:
def _log(self, **kwargs):
status = inspect.stack()[1][3]
details = Logger.serialize_details(**kwargs)
with _create_log_record(
functools.partial(_set_module_name, name=self.module),
functools.partial(_set_func_name, name=self.scope)
):
# Ignore the ContinuationError as an actual error.
is_error = all(sys.exc_info()) and sys.exc_info()[0] is not ContinuationError
self._logger.log(level=self._logger.level, msg=None, exc_info=is_error, extra={
"parent": self.parent.id if self.parent else None,
"node": self.id,
"status": status,
"elapsed": self.elapsed,
"details": details,
"attachment": self.attachment
})
Internally the package is using python's standard logging library.
Example
I use it like this:
import wiretap.src.wiretap as wiretap # becasue it's from the test environment
@wiretap.telemetry(on_started=lambda p: {"value": p["value"], "bar": p["bar"]}, on_completed=lambda r: {"count": r})
def foo(value: int, logger: wiretap.Logger = None, **kwargs) -> int:
logger.running(test=f"{value}")
raise wiretap.ContinuationError("No luck!", 0, foo="bar")
return 3
if __name__ == "__main__":
print(foo(1, bar="baz")) # <-- prints: 0
What do you think of this idea? I guess I probably should check if the decorated function is expected to return something and throw an invalid operation exception when a return value wasn't provided.
