阅读目录
- 一 什么是装饰器
- 二 装饰器需要遵循的原则
- 三 实现装饰器知识储备
- 四 高阶函数
- 五 函数嵌套
- 六 闭包
- 七 无参装饰器
- 八 装饰器应用示例
- 九 超时装饰器
器即函数
装饰即修饰，意指为其他函数添加新功能
装饰器定义：本质就是函数，功能是为其他函数添加新功能
1.不修改被装饰函数的源代码（开放封闭原则）
2.为被装饰函数添加新功能后，不修改被修饰函数的调用方式
装饰器=高阶函数+函数嵌套+闭包
高阶函数定义:1.函数接收的参数是一个函数名
2.函数的返回值是一个函数名
3.满足上述条件任意一个,都可称之为高阶函数
高阶函数总结1.函数接收的参数是一个函数名 作用:在不修改函数源代码的前提下,为函数添加新功能, 不足:会改变函数的调用方式2.函数的返回值是一个函数名 作用:不修改函数的调用方式 不足:不能添加新功能
1 def father(name):
2 print('from father %s' %name)
3 def son():
4 print('from son')
5 def grandson():
6 print('from grandson')
7 grandson()
8 son()
9
10 father('林海峰')
1 '''
2 闭包:在一个作用域里放入定义变量,相当于打了一个包
3 '''
4 def father(name):
5 def son():
6 # name='alex'
7 print('我爸爸是 [%s]' %name)
8 def grandson():
9 # name='wupeiqi'
10 print('我爷爷是 [%s]' %name)
11 grandson()
12 son()
13
14 father('林海峰')
无参装饰器=高级函数+函数嵌套
基本框架
1 #这就是一个实现一个装饰器最基本的架子 2 def timer(func): 3 def wrapper(): 4 func() 5 return wrapper
加上参数
1 def timer(func): 2 def wrapper(*args,**kwargs): 3 func(*args,**kwargs) 4 return wrapper
加上功能
1 import time
2 def timer(func):
3 def wrapper(*args,**kwargs):
4 start_time=time.time()
5 func(*args,**kwargs)
6 stop_time=time.time()
7 print('函数[%s],运行时间是[%s]' %(func,stop_time-start_time))
8 return wrapper
加上返回值
1 import time
2 def timer(func):
3 def wrapper(*args,**kwargs):
4 start_time=time.time()
5 res=func(*args,**kwargs)
6 stop_time=time.time()
7 print('函数[%s],运行时间是[%s]' %(func,stop_time-start_time))
8 return res
9 return wrapper
使用装饰器
1 def cal(array): 2 res=0 3 for i in array: 4 res+=i 5 return res 6 7 cal=timer(cal) 8 cal(range(10))
语法糖@
1 @timer #@timer就等同于cal=timer(cal) 2 def cal(array): 3 res=0 4 for i in array: 5 res+=i 6 return res 7 8 cal(range(10))
user_list=[
{'name':'alex','passwd':'123'},
{'name':'linhaifeng','passwd':'123'},
{'name':'wupeiqi','passwd':'123'},
{'name':'yuanhao','passwd':'123'},
]
current_user={'username':None,'login':False}
def auth_deco(func):
def wrapper(*args,**kwargs):
if current_user['username'] and current_user['login']:
res=func(*args,**kwargs)
return res
username=input('用户名: ').strip()
passwd=input('密码: ').strip()
for index,user_dic in enumerate(user_list):
if username == user_dic['name'] and passwd == user_dic['passwd']:
current_user['username']=username
current_user['login']=True
res=func(*args,**kwargs)
return res
break
else:
print('用户名或者密码错误,重新登录')
return wrapper
@auth_deco
def index():
print('欢迎来到主页面')
@auth_deco
def home():
print('这里是你家')
def shopping_car():
print('查看购物车啊亲')
def order():
print('查看订单啊亲')
print(user_list)
# index()
print(user_list)
home()
user_list=[
{'name':'alex','passwd':'123'},
{'name':'linhaifeng','passwd':'123'},
{'name':'wupeiqi','passwd':'123'},
{'name':'yuanhao','passwd':'123'},
]
current_user={'username':None,'login':False}
def auth(auth_type='file'):
def auth_deco(func):
def wrapper(*args,**kwargs):
if auth_type == 'file':
if current_user['username'] and current_user['login']:
res=func(*args,**kwargs)
return res
username=input('用户名: ').strip()
passwd=input('密码: ').strip()
for index,user_dic in enumerate(user_list):
if username == user_dic['name'] and passwd == user_dic['passwd']:
current_user['username']=username
current_user['login']=True
res=func(*args,**kwargs)
return res
break
else:
print('用户名或者密码错误,重新登录')
elif auth_type == 'ldap':
print('巴拉巴拉小魔仙')
res=func(*args,**kwargs)
return res
return wrapper
return auth_deco
#auth(auth_type='file')就是在运行一个函数,然后返回auth_deco,所以@auth(auth_type='file')
#就相当于@auth_deco,只不过现在,我们的auth_deco作为一个闭包的应用,外层的包auth给它留了一个auth_type='file'参数
@auth(auth_type='ldap')
def index():
print('欢迎来到主页面')
@auth(auth_type='ldap')
def home():
print('这里是你家')
def shopping_car():
print('查看购物车啊亲')
def order():
print('查看订单啊亲')
# print(user_list)
index()
# print(user_list)
home()
import sys,threading,time
class KThread(threading.Thread):
"""A subclass of threading.Thread, with a kill()
method.
Come from:
Kill a thread in Python:
http://mail.python.org/pipermail/python-list/2004-May/260937.html
"""
def __init__(self, *args, **kwargs):
threading.Thread.__init__(self, *args, **kwargs)
self.killed = False
def start(self):
"""Start the thread."""
self.__run_backup = self.run
self.run = self.__run # Force the Thread to install our trace.
threading.Thread.start(self)
def __run(self):
"""Hacked run function, which installs the
trace."""
sys.settrace(self.globaltrace)
self.__run_backup()
self.run = self.__run_backup
def globaltrace(self, frame, why, arg):
if why == 'call':
return self.localtrace
else:
return None
def localtrace(self, frame, why, arg):
if self.killed:
if why == 'line':
raise SystemExit()
return self.localtrace
def kill(self):
self.killed = True
class Timeout(Exception):
"""function run timeout"""
def timeout(seconds):
"""超时装饰器，指定超时时间
若被装饰的方法在指定的时间内未返回，则抛出Timeout异常"""
def timeout_decorator(func):
"""真正的装饰器"""
def _new_func(oldfunc, result, oldfunc_args, oldfunc_kwargs):
result.append(oldfunc(*oldfunc_args, **oldfunc_kwargs))
def _(*args, **kwargs):
result = []
new_kwargs = { # create new args for _new_func, because we want to get the func return val to result list
'oldfunc': func,
'result': result,
'oldfunc_args': args,
'oldfunc_kwargs': kwargs
}
thd = KThread(target=_new_func, args=(), kwargs=new_kwargs)
thd.start()
thd.join(seconds)
alive = thd.isAlive()
thd.kill() # kill the child thread
if alive:
raise Timeout(u'function run too long, timeout %d seconds.' % seconds)
else:
return result[0]
_.__name__ = func.__name__
_.__doc__ = func.__doc__
return _
return timeout_decorator
@timeout(5)
def method_timeout(seconds, text):
print('start', seconds, text)
time.sleep(seconds)
print('finish', seconds, text)
return seconds
method_timeout(6,'asdfasdfasdfas')
