Decorator

Decoratotr 简述

装饰器的本质其实就是函数，接收一个函数作为参数，然后返回装饰后的函数

Decorator 作用

装饰器可以在不修改原函数的情况下，对原函数添加功能，因为装饰器通常都是返回一个封装函数，这个封装函数在传入的函数前后做一些事情

Decorator 例子

简单的装饰器

def outer(fun):
    def wrapper():
        print('Authentication')
        fun()
        print('Over')
    return wrapper
@outer
def func1():
    print('func1')

func1()

带参数的装饰器

def outer(fun):
    def wrapper(arg):
        print('Authentication')
        fun(arg)
        print('Over')
    return wrapper
@outer
def func1(arg):
    print('func1', arg)

func1('A')

带参数以及返回值的装饰器

def outer(fun):
    def wrapper(arg):
        print('Authentication')
        res = fun(arg)
        print('Over')
        return res
    return wrapper
@outer
def func1(arg):
    print('func1', arg)
    return 'Da'

print(func1('A'))

Decortor 执行流程

在PyCharm下打上断点多Debug几遍就理解了

1. 定义outer函数放入内存
2. 然后到@outer这里，这里可以理解为：func1=outer(func1)
3. outer函数接收func1作为fun，然后通过warpper函数封装了一个包含原func1函数以及新添加内容的新函数，然后返回出去
4. 然后将新函数返回出来，此时的func1等于wrapper函数
5. 最后当我们再次调用func1()时此时调用就是warpper函数

Decorator 应用场景

函数测速

import time

def timer(fun):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        ret = fun(*args, **kwargs)
        print('Run Time: ', time.time() - start_time)
        return ret
    return wrapper

@timer
def a():
    print('Start sleep...')
    time.sleep(3)
    print('Stop sleep...')

@timer
def b():
    print('Start sleep...')
    time.sleep(5)
    print('Stop sleep...')

a()
b()

输出

Start sleep...
Stop sleep...
Run Time:  3.0
Start sleep...
Stop sleep...
Run Time:  5.000348091125488

接口认证

简单写了下针对接口做U/P认证，比我们

def auth(username, passwd):
    print('Starting Authentication...')
    if username == 'Da' and passwd == '123123':
        return 0
    else:
        return 1

def log(user, res):
    print('\n' + user + " Authentication " + res)

def authentication(fun):
    def wrapper(fn, un, pwd):
        if auth(un, pwd) == 0:
            fun(fn, un, pwd)
            log(un, 'OK.')
        else:
            log(un, 'Failed.')
    return wrapper

@authentication
def get_info(fn, un, pwd):
    f = open(fn)
    for i in f.readlines():
        print(i, end='')
    return 0

@authentication
def write_info(fn, un, pwd):
    f = open(fn, 'a+')
    f.write('I am Da\n')
    print('Write data Successful.', end='')



get_info('test', 'Da', '123123')
write_info('test', 'Da', '123123')

看下get_info和write_info验证是否正常以及输出

Starting Authentication...
Name: 'Da'
Age: 21
Gender: Male
Da Authentication OK.

Starting Authentication...
Write data Successful.
Da Authentication OK.

不过这个验证的装饰器还有一些问题，比如在写入数据时write_info函数肯定要接受content变量，这样get_info在读取数据时就会有问题，Em，还要琢磨下~

OK加个判断就好了

def auth(username, passwd):
    print('Starting Authentication...')
    if username == 'Da' and passwd == '123123':
        return 0
    else:
        return 1

def log(user, res):
    print(user + " Authentication " + res)

def authentication(fun):
    def wrapper(fn, un, pwd, content=None):
        if auth(un, pwd) == 0:
            log(un, 'OK.')
        else:
            log(un, 'Failed.')
        if content == None:
            fun(fn, un, pwd)
        else:
            fun(fn, un, pwd, content)

    return wrapper

@authentication
def get_info(fn, un, pwd):
    f = open(fn)
    for i in f.readlines():
        print(i, end='')
    return 0

@authentication
def write_info(fn, un, pwd, content):
    f = open(fn, 'a+')
    f.write(content)
    print('Write data Successful.', end='')

简单缓存

这里使用了wraps主要为了保留住原函数的信息

import time
from functools import wraps
def cache(instance):
    def dec(fn):
        @wraps(fn)
        def wrap(*args, **kwargs):
            pos = ','.join(str(x) for x in args)
            kws = ','.join('{}={}'.format(k, v) for k, v in sorted(kwargs.items()))
            key = '{}::{}::{}'.format(fn.__name__, pos, kws)
            ret = instance.get(key)
            if ret is not None:
                print('Key: {} Hit Cache.'.format(key))
                return ret
            ret = fn(*args, **kwargs)
            print('Key: {} Miss Cache, creating cache.'.format(key))
            instance.set(key, ret)
            return ret
        return wrap
    return dec


class DictCache:
    def __init__(self):
        self.cache = dict()

    def get(self, key):
        return self.cache.get(key)

    def set(self, key, value):
        self.cache[key] = value

cache_instance = DictCache()
@cache(cache_instance)
def long_time(x):
    time.sleep(x)
    return x

print(long_time(3))
print(long_time(3))

监控

def metric(prefix, instance):
    def timeit(fn):
        @wraps(fn)
        def wrap(*args, **kwargs):
            start = time.time()
            ret = fn(*args, **kwargs)
            key = '{}.{}.{}'.format(prefix, fn.__module__, fn.__name__)
            instance.send(key, time.time() - start)
            return ret
        return wrap
    return timeit


class LoggingMetric:
    def send(self, key, value):
        logging.warning('{} => {}'.format(key, value))


@metric(prefix='Lotus',instance=LoggingMetric())
def loog_time(x):
    time.sleep(x)
    return x

print(loog_time(3))

路由

flask源码中的这段 flask app.route

def decorator(f):
    endpoint = options.pop('endpoint', None)
    self.add_url_rule(rule, endpoint, f, **options)
    return f
return decorator

参数检查