Details of advanced use of decorators in Python

  • 2020-06-19 10:49:37
  • OfStack

In Python, decorator 1 is generally used to modify functions to achieve common functions and achieve code reuse. Add @xxxx to the function definition, and the function injects some behavior. Amazing! However, this is just grammar sugar.

scenario

Suppose that there are 1 working function for different processing of data:


def work_bar(data):
  pass

def work_foo(data):
  pass

What if we want to output logs before/after a function call?

A fool solution


logging.info('begin call work_bar')
work_bar(1)
logging.info('call work_bar done')

What if there are multiple code calls? It scares me to think about it!

Function of packaging

The stupid solution is to have too much code redundancy and write logging 1 time for every function call. This redundant logic can be encapsulated in a new function:


def smart_work_bar(data):
  logging.info('begin call: work_bar')
  work_bar(data)
  logging.info('call doen: work_bar')

In this way, each call to smart_work_bar will work:


smart_work_bar(1)

# ...

smart_work_bar(some_data)

General closure

It looks perfect... However, do you have to implement smart_work_foo again when work_foo has the same need? This is obviously not scientific!

Don't worry, we can use closures:


def log_call(func):
  def proxy(*args, **kwargs):
    logging.info('begin call: {name}'.format(name=func.func_name))
    result = func(*args, **kwargs)
    logging.info('call done: {name}'.format(name=func.func_name))
    return result
  return proxy

This function takes 1 function object as an argument and returns 1 proxy function. When the proxy function is called, the log is output first, then the proxy function is called, then the log is output after the call is completed, and finally the result of the call is returned. In this way, don't you achieve the purpose of generalization? -- log_call is easy to handle for any proxy function func.


smart_work_bar = log_call(work_bar)
smart_work_foo = log_call(work_foo)

smart_work_bar(1)
smart_work_foo(1)

# ...

smart_work_bar(some_data)
smart_work_foo(some_data)

In line 1, log_call receives the parameter work_bar, returns a proxy function proxy, and assigns smart_work_bar. In line 4, call smart_work_bar, the proxy function proxy, prints the log, then calls func, work_bar, and finally prints the log. Notice that in the proxy function, func is tightly associated with the work_bar object passed in, which is called a closure.

By the way, you can overwrite the name of the proxy function, and the new name prefixed by smart_ is still a bit cumbersome:


work_bar = log_call(work_bar)
work_foo = log_call(work_foo)

work_bar(1)
work_foo(1)

Syntactic sugar

Take a look at the following code:


def work_bar(data):
  pass
work_bar = log_call(work_bar)


def work_foo(data):
  pass
work_foo = log_call(work_foo)

Although the code is not redundant, it is not intuitive to see. At this point, grammar sugar arrives


@log_call
def work_bar(data):
  pass

So, note that at point 1 (highlighted), @log_call is simply telling the Python compiler to insert the code work_bar = log_call(work_bar).

Evaluate the decorator

First guess what the decorator eval_now does?


def eval_now(func):
  return func()

It looks strange, no proxy function defined, is it a decorator?


logging.info('begin call work_bar')
work_bar(1)
logging.info('call work_bar done')
0

This code outputs 1, which evaluates the function call. So what's the point? Can't I just say foo = 1? In this simple example, of course you could write it this way. Let's take a more complex example -- initialize a log object:


# some other code before...

# log format
formatter = logging.Formatter(
  '[%(asctime)s] %(process)5d %(levelname) 8s - %(message)s',
  '%Y-%m-%d %H:%M:%S',
)

# stdout handler
stdout_handler = logging.StreamHandler(sys.stdout)
stdout_handler.setFormatter(formatter)
stdout_handler.setLevel(logging.DEBUG)

# stderr handler
stderr_handler = logging.StreamHandler(sys.stderr)
stderr_handler.setFormatter(formatter)
stderr_handler.setLevel(logging.ERROR)

# logger object
logger = logging.Logger(__name__)
logger.setLevel(logging.DEBUG)
logger.addHandler(stdout_handler)
logger.addHandler(stderr_handler)

# again some other code after...

eval_now:


# some other code before...

@eval_now
def logger():
  # log format
  formatter = logging.Formatter(
    '[%(asctime)s] %(process)5d %(levelname) 8s - %(message)s',
    '%Y-%m-%d %H:%M:%S',
  )

  # stdout handler
  stdout_handler = logging.StreamHandler(sys.stdout)
  stdout_handler.setFormatter(formatter)
  stdout_handler.setLevel(logging.DEBUG)

  # stderr handler
  stderr_handler = logging.StreamHandler(sys.stderr)
  stderr_handler.setFormatter(formatter)
  stderr_handler.setLevel(logging.ERROR)

  # logger object
  logger = logging.Logger(__name__)
  logger.setLevel(logging.DEBUG)
  logger.addHandler(stdout_handler)
  logger.addHandler(stderr_handler)

  return logger

# again some other code after...

The purpose of the two pieces of code is the same, but the latter is clearly clearer and has the feel of a block of code. More importantly, function calls are initialized in the local namespace to prevent temporary variables (such as formatter) from contaminating the external namespace (such as the global).

With parameter decorator

Define a decorator for recording slow function calls:


logging.info('begin call work_bar')
work_bar(1)
logging.info('call work_bar done')
3

Lines 3 and 5 respectively sample the current time before and after the function call. Line 7 calculates the call time, which is more than 1 second and outputs a warning log.


logging.info('begin call work_bar')
work_bar(1)
logging.info('call work_bar done')
4

However, the threshold setting is always situational, and different functions may set different values. If only thresholds could be parameterized:


logging.info('begin call work_bar')
work_bar(1)
logging.info('call work_bar done')
5

However, the @xxxx syntax sugar always calls the decorator with an argument to the decorated function, meaning there is no opportunity to pass the threshold argument. What to do? -- Encapsulates the threshold parameters with 1 closure:


logging.info('begin call work_bar')
work_bar(1)
logging.info('call work_bar done')
6

Thus, log_slow_call(threshold=0.5) calls the return function decorator, which has the closure variable threshold with a value of 0.5. decorator with sleep_seconds.

With the default threshold, the function call cannot be omitted:


@log_slow_call()
def sleep_seconds(seconds):
  time.sleep(seconds)

Virgo may not be happy with the parenthesis in line 1, but this can be improved:


def log_slow_call(func=None, threshold=1):
  def decorator(func):
    def proxy(*args, **kwargs):
      start_ts = time.time()
      result = func(*args, **kwargs)
      end_ts = time.time()

      seconds = start_ts - end_ts
      if seconds > threshold:
      logging.warn('slow call: {name} in {seconds}s'.format(
        name=func.func_name,
        seconds=seconds,
      ))

      return result

    return proxy

  if func is None:
    return decorator
  else:
    return decorator(func)

This is compatible with two different usages, A default threshold (no calls); Use B to customize thresholds (with calls).


logging.info('begin call work_bar')
work_bar(1)
logging.info('call work_bar done')
9

In A, what happens is that log_slow_call(sleep_seconds), that the func parameter is non-null, which is directly called decorator wrapped and returned (the threshold is the default).

In B, the first thing that happens is log_slow_call(threshold=0.5). The func parameter is empty, and the new decorator decorator is returned directly. Then, decorator decorates the function sleep_seconds, or decorator(sleep_seconds). Note that the value associated with threshold at this point is 0.5, completing the customization.

You may have noticed that it's best to use keyword arguments this way -- using positional arguments can be ugly:


# Case B-
@log_slow_call(None, 0.5)
def sleep_seconds(seconds):
  time.sleep(seconds)

Of course, it's a good practice to try to use keyword arguments in function calls, especially when there are many arguments.

Intelligent decorator

As described in the previous section, there are many levels of nesting, and if every similar decorator were implemented this way, it would be more laborious (not enough brain) and more error-prone.

If you have a smart decorator smart_decorator and a decorative decorator log_slow_call, you get the same capability. In this way, the log_slow_call definition will be clearer and easier to implement:


@smart_decorator
def log_slow_call(func, threshold=1):
  def proxy(*args, **kwargs):
    start_ts = time.time()
    result = func(*args, **kwargs)
    end_ts = time.time()

    seconds = start_ts - end_ts
    if seconds > threshold:
    logging.warn('slow call: {name} in {seconds}s'.format(
      name=func.func_name,
      seconds=seconds,
    ))

    return result

  return proxy

Imagination finished, smart_decorator how to achieve? It's also simple:


def smart_decorator(decorator):

  def decorator_proxy(func=None, **kwargs):
    if func is not None:
      return decorator(func=func, **kwargs)

    def decorator_proxy(func):
      return decorator(func=func, **kwargs)

    return decorator_proxy

  return decorator_proxy

With smart_decorator implemented, the idea is set! In this case, log_slow_call is decorator_proxy(the outer layer), and the associated closure variable decorator is the log_slow_call defined at the beginning of this section (called real_log_slow_call to avoid ambiguity). log_slow_call supports the following USES:


# Case A
@log_slow_call
def sleep_seconds(seconds):
  time.sleep(seconds)

In A, decorator_proxy(sleep_seconds)(outer layer) is executed, func is not empty, kwargs is empty; Directly execute decorator(func=func, **kwargs), that is, real_log_slow_call(sleep_seconds), resulting in proxy associated with the default parameter.


# Case B
# Same to Case A
@log_slow_call()
def sleep_seconds(seconds):
  time.sleep(seconds)

In B, first execute decorator_proxy(), func and kwargs are null, return decorator_proxy object (inner layer); Then execute decorator_proxy(sleep_seconds)(inner layer); Finally, execute decorator(func, **kwargs), equivalent to real_log_slow_call(sleep_seconds), with effect and usage A1.


# Case C
@log_slow_call(threshold=0.5)
def sleep_seconds(seconds):
  time.sleep(seconds)

In C, first execute decorator_proxy(threshold=0.5), func is empty but kwargs is not empty, return decorator_proxy object (inner layer); Then execute decorator_proxy(sleep_seconds)(inner layer); Finally, decorator(sleep_seconds, **kwargs) is executed, which is equivalent to real_log_slow_call(sleep_seconds, threshold=0.5). The threshold is customized!


Related articles: