Python之装饰器的使用

闭包

对于函数内声明的变量一般都为临时变量，其生命周期随函数结束而结束。

但闭包是个例外

闭包满足的条件：
1、必须要有内嵌函数
2、内函数必须引用外函数的变量
3、外函数必须返回的内函数

def outer(x):a = 300def inner():print(x+a)return inner	# 这里不能打括号，不然就是返回调用了
d = outer(100)
d()		# 输出400

对于闭包函数对象，其内置属性__closure__非空

print(d.__closure__)	# 会记录其引用了的变量
(<cell at 0x00000256060C2FD0: int object at 0x0000025604DB7F70>, <cell at 0x00000256060C2FA0: int object at 0x00000256757955D0>)

装饰器

装饰器本质就是闭包函数，需要把一个callable（函数、类）对象作为参数传入进去，所以装饰器只能用在函数或类上面

装饰器是一种设计模式，主要用于在不改变函数或者类的源代码的基础上，给其添加一些额外的功能

装饰器并非python独有，只要能实现闭包的语言都能实现装饰器

函数实现

# 装饰器模板，在此基础上添加新功能
def runtime(func):def inner(*args,**kwargs):		# 这样可以接收任意个参数result = func(*args,**kwargs)return resultreturn inner

# 装饰器，用于统计函数执行时间
def runtime(func):def inner(*args,**kwargs):start = time.time()result = func(*args,**kwargs)end = time.time()print(f"执行{func.__name__}花费{end-start}s")return resultreturn inner@runtime		# 给函数应用装饰器，去掉该装饰语句后就不会执行装饰器的效果；	func1 = runtime(func1)
def func1():time.sleep(2)print("func1")func1()			# 调用的时候没有任何区别，无需改变源代码，然实际调用的是inner()函数
#---输出---
func1
执行func1花费2.012355089187622s

原理

当代码执行到@runtime时，runtime函数就已经被执行（func1函数也被传入到runtime函数中）

此时调用func1函数时，实际调用为inner()函数，证据是print(func1.__name__)为inner

类实现(带参数)

函数实现想带参数也只需在定义时声明和装饰时传入就好

class A:def __init__(self, username):	# 通过init函数传参self.username=usernamedef __call__(self, func):		# 通过call函数进行实现装饰器def inner(*args,**kwargs):result = func(*args,**kwargs)return resultreturn inner@A(username="root")		# 通过类进行装饰和传参
def func1():pass

装饰类

def runtime(cls):def inner(*args,**kwargs):		# 装饰器函数返回的就是类了，而不是返回值return cls(*args,**kwargs)return inner@runtime
class A:passa1 = A()	# 与装饰函数一样，此时A调用的cls这个内部函数

应用

引入日志，用于1.排错、分析和定位故障；2.用户行为分析；3.页面埋点(相当于数据采集)
增加计时逻辑来检测性能
给函数加入事务的能力
权限控制
……

权限控制

# 权限控制
username = input("输入你的用户名：")def login_require(func):def inner(*args,**kwargs):if username != "root":		# 当用户为root时才能执行add函数print("权限拒绝！")returnresult = func(*args,**kwargs)return resultreturn inner@login_require
def add(x,y):print(x+y)add(1,2)

计时和添加日志

# 对add函数添加两个装饰器，一个记录花费时间，一个添加日志
import time
def run_time(func):def inner(*args,**kwargs):start_time=time.time()result = func(*args,**kwargs)end_time = time.time()print(f"{func.__name__}程序运行共花费{end_time-start_time}s")return resultreturn innerimport logging
def log_add(func):def inner(*args,**kwargs):LOG_FORMAT = "%(asctime)s - %(filename)s[%(levelname)s] : %(message)s"logging.basicConfig(format=LOG_FORMAT)logging.warning(f"-->{func.__name__}<--程序被执行！")result = func(*args,**kwargs)return resultreturn inner@run_time
@log_add
def add(x,y):return x+yresult = add(1,2)
print(result)# 输出效果
inner程序运行共花费0.0009965896606445312s
2023-08-25 11:43:02,963 - 装饰器练习.py[WARNING] : -->add<--函数被执行！
3

系统识别

# 写一个局域网扫描函数，--subprocess，再写一个装饰器，判断当前系统是否为linux，是才能执行# 实现ping函数，ping成功返回true，否则为false
import platform    # For getting the operating system name
import subprocess  # For executing a shell commanddef require_linux(func):def inner(*args,**kwargs):if platform.system().lower() == 'linux':return func(*args,**kwargs)else:print(f"操作系统不是linux，拒绝执行{func.__name__}！")returnreturn innerdef ping(host):"""Returns True if host (str) responds to a ping request.Remember that a host may not respond to a ping (ICMP) request even if the host name is valid."""# Option for the number of packets as a function ofparam = '-n' if platform.system().lower()=='windows' else '-c'# Building the command. Ex: "ping -c 1 google.com"command = ['ping', param, '1', host]return subprocess.call(command) == 0@require_linux
def scan_lan(net_segment):list = net_segment.split('.')seg = '.'.join(list[0:3])for i in range(1,255):host = seg + "." + str(i)if ping(host):with open("actived.txt",'a') as fp:		# 将能到达的服务器ip写入文件fp.write(f"{host}\n")else:with open("unreachable.txt",'a') as fp:fp.write(f"{host}\n")returnscan_lan(net_segmeng)# --------效果--------
# Windows:
C:\Users\zhihe>python3 D:\pycharm2020.3\My_Project\装饰器.py
操作系统不是linux，拒绝执行scan_lan！
# Linux:
[root@zh-ali py练习]# tail unreachable.txt
172.26.252.1
172.26.252.2
172.26.252.3
……

redis_require

写一个装饰器，对于要查询的同学信息，在给定同学id后先到redis上查找，查找不到再去mysql查找，并将查找到的值写入到redis。

import redis
import pymysql'''
对于要查询的同学信息，在给定同学id后先到redis上查找，查找不到再去mysql查找，并将查找到的值写入到redis。
'''def redis_req(func):def inner(id):r = redis.Redis(host='192.168.10.21', port=6379, db=1, decode_responses=True)if r.hkeys(id):print("通过redis查询：")print(f"同学姓名：{r.hget(id, 'name')}")print(f"同学年龄：{r.hget(id, 'age')}")print(f"同学成绩：{r.hget(id, 'grade')}")data = (int(id),r.hget(id, 'name'),int(r.hget(id, 'age')),int(r.hget(id, 'grade')))	# 保证data格式与mysql返回格式一致else:data = func(id)print("通过mysql查询：")if data:				# data不为空时才输出并写入到redisprint(data)r.hset(id, "name", data[1])r.hset(id, "age", data[2])r.hset(id, "grade", data[3])else:print("该id不存在！")r.close()return datareturn inner@redis_req
def mysql(id):db = pymysql.connect(host='192.168.10.21',# port=3306,user='zh',password='123456',database='zh')cursor = db.cursor()cursor.execute(f"select * from stu where id={id}")data = cursor.fetchone()db.close()return dataid = input("输入你要查询的同学id：")
data = mysql(id)
print(data)