CODEWITHSUNDEEP
pythonsocketsgoogle-app-enginedecoratorstatsd
Bodhi
Bodhi

Reputation: 1427

How should I catch exceptions in a decorator function that can raise, if the decorator function is in a library I can't modify?

I'm working the python statsd library on Google App Engine (GAE). Unfortunately, GAE can raise ApplicationError: 4 Unknown error. from time to time when using sockets. The error is an apiproxy_errors.ApplicationError.

The statsd client is already setup to catch socket.error, but not the ApplicationError that sockets can raise on GAE.

I'm specifically working with timer, which returns an instance of Timer: https://github.com/jsocol/pystatsd/blob/master/statsd/client.py#L13

The __call__ method of Timer allows it to be used as a decorator, like so:

from statsd import StatsClient

statsd = StatsClient()

@statsd.timer('myfunc')
def myfunc(a, b):
    """Calculate the most complicated thing a and b can do."""

I don't have easy ability to modify the Timer.__call__ method itself to simply also catch ApplicationError.

How should I write a wrapper or additional decorator that still allows clean decoration like @my_timer_wrapper('statsd_timer_name') but which catches additional exceptions that may occur in the wrapped/decorated timer method?

This is in a foundation module in my codebase that will be used in many places (wherever we want to time something). So although this SO answer might work, I really want to avoid forcing all uses of @statsclient.timer in my codebase to themselves be defined within try-except blocks.

I'm thinking of doing something like the following:

def my_timer_wrapper(wrapped_func, *args, **kwargs):
  @functools.wraps(wrapped_func)
  class Wat(object):
    def __call__(self, *args, **kwargs):
      timer_instance = stats_client.timer(*args, **kwargs)
      try:
        return timer_instance.__call__(wrapped_func)(*args, **kwargs)
      except Exception:
        logger.warning("Caught exception", exc_info=True)
        def foo():
          pass
        return foo

  return Wat()

which would then be used like:

@my_timer_wrapper('stastd_timer_name')
def timed_func():
  do_work()

Is there a better or more pythonic way?

Upvotes: 3

Views: 1286

Answers (1)

jsbueno
jsbueno

Reputation: 110301

It looks like it is a case for an "as straightforward as possible" new decorator adding an extra try/except around your timer decorator.

The only matter being that decorators that take parameters needing 2 levels of nested functions to be defined, almost always makes them look complicated, even when they are not: 

from functools import wraps

def  shielded_timer(statsd, 
                    exceptions=(apiproxy_errors.ApplicationError),
                    name=None):

    def decorator(func):
        timer_decorator = statsd.timer(name or func.__name__)
        new_func = timer_decorator(func)
        @wraps(func)
        def wrapper(*args, **kw):
            try:
                return new_func(*args, **kw)
            except BaseException as error:
                if isinstance (error, exceptions):
                    # Expected error (ApplicationError by default) ocurred
                    pass
                else:
                    raise
        return wrapper
    return decorator



#####################
statsd = StatsClient()
@shielded_timer(statsd)
def my_func(a,b):
    ...

As you can see it is easy enough to even include extra niceties - in this case I've made the wanted exceptions configurable at decoration time, and optionally, it uses the name of the decorated function automatically in the call to statsd.timer.

Upvotes: 1

Related Questions