Suppose I write a simple Python class, Delay, whose point is to encapsulate a delayed (lazy) computation:

class Delay:
    def __init__(self, fn, *args, **kwargs):
        self.partial = (fn, args, kwargs)
        self.result = None
    def __call__(self):
        if self.partial is not None:
            (fn, args, kwargs) = self.partial
            self.result = fn(*args, **kwargs)
            self.partial = None
        return self.result

This is straightforward enough, but now let's think about how it will be used:

# Somewhere in some initialization module of the codebase:
def load_historical_data(filename):
    with open(filename, 'r') as f:
        return json.load(f)
def init_runtime(param_dict):
    ...
    # (Prepare to) load some data that won't be needed right away.
    runtime['historical_data'] = Delay(load_historical_data,
                                       param_dict['hist_filename'])
    ...
    return runtime

# Some far remote corner of the codebase...
def _analyze_history_job_firstpass(runtime, kwargs*):
    ...
    histdata = runtime['historical_data']()
    ...

One problem with the above paradigm arises when there is a bug in the init_runtime function—for example, if the param_dict['hist_filename'] isn't a valid filename. In this case, when _analyze_history_job_firstpass eventually gets called, it will raise an exception due to the file not being found, but nothing about that exception will point back to init_runtime, which makes debugging difficult.

It would be ideal if the Delay class could detect the exception during its __call__ method and instead raise an alternate exception that documented both the call stack that created fn and the call stack that called fn. How can one achieve this? (If there are multiple ways, what are the advantages/disadvantages of each?)

CodePudding user response：

An arguably friendlier approach would be to produce a forced exception and save it into the object during initialization, and then raise the saved exception when handling an exception that actually occurs during the execution of the delayed call:

class Delay:
    def __init__(self, fn, *args, **kwargs):
        self.partial = (fn, args, kwargs)
        self.result = None
        try:
            # artificially create an exception that can be re-raised later
            raise ValueError(f'{fn.__name__} failed with args {args} and kwargs {kwargs}')
        except ValueError as e:
            self.init_exception = e

    def __call__(self):
        if self.partial is not None:
            (fn, args, kwargs) = self.partial
            try:
                self.result = fn(*args, **kwargs)
            except Exception:
                # raise the saved exception
                raise self.init_exception
            self.partial = None
        return self.result

def load_historical_data(filename):
    with open(filename, 'r') as f:
        return f.read()

def init_runtime(filename):
    runtime = Delay(load_historical_data, filename)
    return runtime

def _analyze_history_job_firstpass(runtime):
    return runtime()

_analyze_history_job_firstpass(init_runtime('foobar'))

This produces an error output with tracebacks of both the call stack that created fn and the call stack that called fn:

Traceback (most recent call last):
  File "main.py", line 15, in __call__
    self.result = fn(*args, **kwargs)
  File "main.py", line 23, in load_historical_data
    with open(filename, 'r') as f:
FileNotFoundError: [Errno 2] No such file or directory: 'foobar'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "main.py", line 33, in <module>
    _analyze_history_job_firstpass(init_runtime('foobar'))
  File "main.py", line 31, in _analyze_history_job_firstpass
    return runtime()
  File "main.py", line 18, in __call__
    raise self.init_exception
  File "main.py", line 7, in __init__
    raise ValueError(f'{fn.__name__} failed with args {args} and kwargs {kwargs}')
ValueError: load_historical_data failed with args ('foobar',) and kwargs {}

Demo: https://replit.com/@blhsing/PassionateKnowingAlgorithms#main.py

CodePudding user response：

If I understood question correctly, you want to get callstack or traceback of exceptions from init_runtime and analyze it in _analyze_history_job_firstpass. (Or did you want to aggregate multiple exceptions?)

Here are some alternatives I can think of:

1. creating decorator (another example)
2. make caller log/get/send traceback in exception
3. validate file path first and log/get/send traceback in exception

custom decorator

import traceback
from functools import wraps

def trace(orig_func):
    @wraps(orig_func)
    def wrapper(*args, **kwargs):
        trace = traceback.format_exc()
        _analyze_history_job_firstpass(trace)
        return trace
        
    return wrapper
  
 class Delay:
    @trace
    def foo(self):
        return

caller does the job

def init_runtime(param_dict):
    
    try:
        runtime['historical_data'] = Delay(load_historical_data,
                                           param_dict['hist_filename'])
    except Exception:
        trace = traceback.format_exc()
        logger.warn(trace)
        _analyze_history_job_firstpass(trace)
    
    return runtime

I'd like to know more details about your intention then i can think of better ways.