关于python装饰器可能是最全的一篇文章(包括有用的例子)

def mydecorator(function):
    def wrapped(*args, **kwargs):
    # 函数调用之前, 做点什么
    result = function(*args, **kwargs)
    # 函数调用之后, 做点什么
    return result
return wrapped

class DecoratorClass:
    def __init__(self, function):
        self.function = function
    def __call__(self, *args, **kwargs):
        # 在函数调用之前做点什么
        result = self.function(*args, **kwargs)
        # 在函数调用之后做点什么
        return result

def repeat(number=3):
    """
    多次重复执行被装饰的函数, 返回最后一次执行的结果
    :param number: 重复次数, 默认是3
    :return:
    """
    def actual_decorator(function):
        def wrapper(*args, **kwargs):
            result = None
            for _ in range(number):
            result = function(*args, **kwargs)
            return result
        return wrapper
    return actual_decorator

from functools import wraps
def preserving_decrator(function):
    @wraps(function)
    def wrapped(*args, **kwargs):
        """包装函数内部文档"""
        return function(*args, **kwargs)
    return wrapped
@preserving_decrator
def function_with_import_docstring():
    """这是想要保存的重要的函数元数据"""
print(function_with_import_docstring.__name__)
# function_with_import_docstring
print(function_with_import_docstring.__doc__)

xml-rpc是一种基于HTTP使用xml进行通信的rpc协议, 但是python没有静态类型检查, 可以通过装饰器实现.
rpc_info = {}
def xmlrpc(in_=(), out=(type(None),)):
 def _xmlrpc(function):
 # 注册签名
 func_name = function.__name__
 rpc_info[func_name] = (in_, out)
 def __xmlrpc(*args):
 """包装过的函数"""
            # 检查输入的内容
 checkable_args = args[1:]  # 去掉self
 _checktypes(checkable_args, in_)
 # 运行函数
 result = function(*args)
 # 检查输出的内容
 if not type(result) in (tuple, list):
                checkable_result = (result,)
 else:
                checkable_result = result
            _checktypes(checkable_result, out)
 def _checktypes(elements, types):
 """用来检查实参和形参的类型是否符合的子函数"""
 if len(elements) != len(types):
 raise TypeError("argument count is wrong")
            typed = enumerate(zip(elements, types))
 for index, couple in typed:
                arg, of_the_right_type = couple
 if not isinstance(arg, of_the_right_type):
 raise TypeError("arg #%d should be %s" % (index,                    of_the_right_type))
 return __xmlrpc
 return _xmlrpc
class RPCView:            
 @xmlrpc((int, int))  # two int -> None
 def meth1(self, int1, int2):
 print('received %d and %d' % (int1, int2))
 @xmlrpc((str,), (int,))  # string -> int
 def meth2(self, phrase):
 print('received %s' % phrase)
 return 12
print(rpc_info)
# {'meth1': ((<class 'int'>, <class 'int'>), (<class 'NoneType'>,)), 'meth2': ((<class 'str'>,), (<class 'int'>,))}
my = RPCView()
my.meth1(1, 2)
# received 1 and 2
my.meth2(2)
"""
Traceback (most recent call last):
  File "/Users/chenzhang/PycharmProjects/fisher/ceshi.py", line 133, in <module>
    my.meth2(2)
  File "/Users/chenzhang/PycharmProjects/fisher/ceshi.py", line 92, in __xmlrpc
    _checktypes(checkable_args, in_)
  File "/Users/chenzhang/PycharmProjects/fisher/ceshi.py", line 110, in _checktypes
    raise TypeError("arg #%d should be %s" % (index, of_the_right_type))
TypeError: arg #0 should be <class 'str'>
"""

import time
import hashlib
import pickle
cache = {}
def is_obsolete(entry, duration):
 return time.time() - entry['time'] > duration
def compute_key(function, args, kw):
    key = pickle.dumps((function.__name__, args, kw))
 return hashlib.sha1(key).hexdigest()
def memorize(duration=10):
 def _memorize(function):
 def __memorize(*args, **kwargs):
            key = compute_key(function, args, kwargs)
 # 是否已经拥有它了?
 if (key in cache and not is_obsolete(cache[key], duration)):
 print('we got a winner')
 return cache[key]['value']
 # 计算
 result = function(*args, **kwargs)
 # 保存结果
 cache[key] = {
 'value': result,
 'time': time.time()
            }
 return result
 return __memorize
 return _memorize
@memorize()
def very_very_very_complex_stuff(a, b):
    time.sleep(3)
 return a + b
print(very_very_very_complex_stuff(2, 2))
# 4
print(very_very_very_complex_stuff(2, 2))
# we got a winner
# 4
print(cache)
# {'c4ba025dbed84bd8eb75d4beacf6900922117068': {'value': 4, 'time': 1532080242.420179}}
time.sleep(15)
print(very_very_very_complex_stuff(2, 2))
# 4
(3)  代理
代理装饰器使用全局机制来标记和注册函数. 比如根据当前用户的角色是否是admin用户来进行集中式的安全检查和权限判断.
class User:
 def __init__(self, roles):
 self.roles = roles
class Unauthorized(Exception):
 pass
def protect(role):
 def _protect(function):
 def __protect(*args, **kwargs):
            user = globals().get('user')
 if user is None or role not in user.roles:
 raise Unauthorized("I won't tell you")
 return function(*args, **kwargs)
 return __protect
 return _protect
tarek = User(('admin', 'user'))
bill = User(('user',))
class MySecrets:
 @protect('admin')
 def waffle_recipe(self):
 print('use tons of buffer!')
there_are = MySecrets()
user = tarek
there_are.waffle_recipe()
# use tons of buffer!
user = bill
there_are.waffle_recipe()
"""
Traceback (most recent call last):
  File "/Users/chenzhang/PycharmProjects/fisher/ceshi.py", line 38, in <module>
    there_are.waffle_recipe()
  File "/Users/chenzhang/PycharmProjects/fisher/ceshi.py", line 15, in __protect
    raise Unauthorized("I won't tell you")
__main__.Unauthorized: I won't tell you
"""
(4) 上下文提供者
上下文装饰器确保函数运行在正确的上下文中, 比如数据操作多线程共享的时候,用同一个锁来包装只有一个线程访问.
不过通常会用上下文管理器替代.
from threading import RLock
lock = RLock()
def synchronized(function):
 def _synchronized(*args, **kwargs):
        lock.acquire()
 try:
 return function(*args, **kwargs)
 finally:
            lock.release()
 return _synchronized
@synchronized
def thread_safe():
 print('确保锁定资源')