Python tips: 装饰方法实战：类型校验装饰器和锁同步装饰器
在看PEP-0138时无意发现，就萌生起自己实现一个类似的装饰器的念头，毕竟这个装饰器确实很好用。
其中accepts和returns是在原有的基础上进行改造的，详细可以查看代码
#!/usr/bin/python
# -*- coding: utf-8 -*-
"""All are decorator!!!
"""
class ArgumentLengthError(Exception):
""""""
class ArgumentTypeError(Exception):
"""Accepts argument type error."""
class ReturnValueError(Exception):
"""Return value error."""
def accepts(*types):
"""参数类型校验装饰器
若需要装饰的方法是类方法，则必须符合python的规范，类方法第一个参数的名称必须为self
"""
def check_accepts(f):
is_class_method = False
want_len, now_len = len(types), f.func_code.co_argcount
if now_len - want_len == 1 and f.func_code.co_varnames[0] == 'self': # fixed class method
now_len -= 1
is_class_method = True
if want_len != now_len:
raise ArgumentLengthError('want %d args, but now %d' % \
(want_len, now_len))
def new_f(*args, **kwds):
if is_class_method:
check_args = args[1:]
else:
check_args = args
for (a, t) in zip(check_args, types):
if not isinstance(a, t):
raise ArgumentTypeError("arg %r does not match %s" % \
(a, t))
return f(*args, **kwds)
new_f.func_name = f.func_name
new_f.__doc__ = f.__doc__
return new_f
return check_accepts
def returns(rtype):
"""返回值类型校验装饰器"""
def check_returns(f):
def new_f(*args, **kwds):
result = f(*args, **kwds)
if not isinstance(result, rtype):
raise ReturnValueError(
"return value %r does not match %s" % (result,rtype))
return result
new_f.func_name = f.func_name
new_f.__doc__ = f.__doc__
return new_f
return check_returns
def synchronized(lock):
"""锁同步装饰方法
lock必须实现了acquire和release方法
"""
def sync_with_lock(f):
def new_f(*args, **kwargs):
lock.acquire()
try:
return f(*args, **kwargs)
finally:
lock.release()
new_f.func_name = f.func_name
new_f.__doc__ = f.__doc__
return new_f
return sync_with_lock
使用举例：若accepts与其他任何装饰器同时使用的话，必须将accepts放到最内层，要不然，会导致参数校验异常.
@synchronized(__locker)
@returns(tuple)
@accepts(dict)
def update(self, data):
"""更新计划任务"""
tasks = self.get_tasks()
delete_task = None
for task in tasks:
if task[PLANTASK.ID] == data[PLANTASK.ID]:
tasks.insert(tasks.index(task), data)
tasks.remove(task)
delete_task = task
r, msg = self._refresh(tasks, delete_task)
return r, msg, data[PLANTASK.ID]
PS: synchronized 装饰方法在python2.5 + 可以使用with语法来代替了,
def func3():
with LockTester._lock:
LockTester._count += 1
time.sleep(0.1)
return LockTester._count
希望本文对你有用! ^_^
以上是 Python tips: 装饰方法实战：类型校验装饰器和锁同步装饰器 的全部内容， 来源链接： utcz.com/z/387521.html
