01.装饰器

1.1 什么是装饰器?（What）

  • 装饰器本质是函数，用来给其他函数添加新的功能
  • 特点：不修改调用方式、不修改源代码

1.2 装饰器的应用场景？（Where）

  • 用户认证，判断用户是否登录
  • 计算函数运行时间（算是一个功能、在项目里用的不多）
  • 插入日志的时候
  • redis缓存

1.3 为什么使用装饰器？（Why）

  • 结合应用场景说需求

1.4 如何使用装饰器？（How）

1.4.1 装饰器求函数运行时间

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import time
def timer(func):   #timer(test1)  func=test1
    def deco(*args,**kwargs):
        start_time = time.time()
        func(*args,**kwargs)      #run test1
        stop_time = time.time()
        print("running time is %s"%(stop_time-start_time))
    return deco

@timer     # test1=timer(test1)
def test1():
    time.sleep(3)
    print("in the test1")
test1()

1.4.2 三级装饰器

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
#! /usr/bin/env python
# -*- coding: utf-8 -*-
import time
def auth(auth_type):
    print("auth func:",auth_type)
    def outer_wrapper(func):
        def wrapper(*args, **kwargs):
            print("wrapper func args:", *args, **kwargs)
            print('运行前')
            func(*args, **kwargs)
            print('运行后')
        return wrapper
    return outer_wrapper

@auth(auth_type="local") # home = wrapper()
def home():
    print("welcome to home  page")
    return "from home"
home()

1.5 装饰器在项目中应用场景

  • 身份验证
  • 写入日志
  • redis缓存

1.5.1 身份验证

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
user,passwd = 'aaa','123'
def auth(func):
    def wrapper(username,password,*args,**kwargs):
        if user == username and password == passwd:
            print("User has passed authentication")
            res = func(username,password,*args,**kwargs)   #这里执行func()相当于执行调用的函数如home()
            return res          #为了获得home()函数返回值，可以将执行结果赋值给res然后返回print(home())结果是"from home"而不是"None"了
        else:
            raise ValueError("非合法用户")
    return wrapper

@auth
def home(username,password):
    print("welcome to home page")
    return "from home"

home('aaa','123')

1.5.2 记录日志

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
#! /usr/bin/env python
# -*- coding: utf-8 -*-
from functools import wraps
import traceback
def decoratore(func):
    @wraps(func)
    def log(*args,**kwargs):
        try:
            print("当前运行方法",func.__name__)
            return func(*args,**kwargs)
        except Exception as e:
            print(traceback.format_exc())  # 这里应该调用log模块来记录到日志里
    return log

@decoratore
def test():
    int('a')
    pass

if __name__ == '__main__':
    test()
    
    ''' 上面运行结果
    当前运行方法 test
    Traceback (most recent call last):
      File "C:/Users/tom/Desktop/alipay_demo/aaa/t2.py", line 11, in log
        return func(*args,**kwargs)
      File "C:/Users/tom/Desktop/alipay_demo/aaa/t2.py", line 18, in test
        int('a')
    ValueError: invalid literal for int() with base 10: 'a'
     22222
    '''

1.5.3 redis缓存

1
2
3
第一步：查询redis缓存是否存在这个key
第二步：如果存在这个key，不用去mysql中查询，直接从redis中取出数据即可（减轻了mysql压力）
第三步：如果查询的key不存在，先到mysql中查询数据，让后设置到redis中，下次查询就有了
1.5.3.1 2B青年实现
  • 2B青年每一个需要使用缓存的数据，我都写一个方法获取数据，再写一个方法处理缓存。
  • 若需要用到缓存的地方越来越多，每一个都需要这么写一套代码，代码冗余繁琐。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
# coding:utf-8
from django.core.cache import cache
import time

# 获取readed缓存
def get_readed_cache():
    # 判断键是否存在
    key = 'readed'
    if cache.has_key(key):
        data = cache.get(key)
    else:
        # 不存在，则通过sql语句获取数据，并写入缓存,这里只是一个举例的sql语句
        data = "select name from tb"
        # 写入缓存
        cache.set(key, data, 3600 - int(time.time() % 3600))
    return data

def test1():
    data = get_readed_cache()
    return data

def test2():
    data = get_readed_cache()
    return data

if __name__ == '__main__':
    test1()
    test2()
1.5.3.2 NB青年
  • NB青年可以使用三级装饰器，在装饰器中判断key如果存在就从reids中获取，如果不存在就从数据库查询，并设置到reids中
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# coding:utf-8
from django.core.cache import cache

# 获取redis缓存的装饰器
def redis_cache(key, timeout):
    def __redis_cache(func):
        def warpper(*args, **kw):
            if cache.has_key(key):  # 判断缓存是否存在
                data = cache.get(key)
            else:
                # 若不存在则执行获取数据的方法
                # 注意返回数据的类型(字符串，数字，字典，列表均可)
                data = func(*args, **kw)   # 从数据库查询到数据设置到redis中
                cache.set(key, data, timeout)
            return data
        return warpper
    return __redis_cache

#键值为test，超时时间为60秒
@redis_cache('test', 60)
def get_test_data():
    # 获取Blog模型随机排序前3条数据
    # (Blog模型是我自己的模型，具体代码根据自己需求获取数据)
    # values执行结果，将返回一个字典。字典可以直接存入redis
    # data = Blog.objects.values('id', 'caption').order_by('?')[:3]
    data = '从数据库查询到了数据'
    return data

if __name__ == '__main__':
    get_test_data()

__END__