Py学习  »  Python

Python高级特性:详解装饰器与生成器

学姐带你玩AI • 2 月前 • 63 次点击  

来源:投稿  作者:sunny 

编辑:学姐

在Python中,装饰器(Decorator)和生成器(Generator)如同两件神秘法宝,让开发者能以更优雅的方式解决复杂问题,前者赋予代码超越流程控制的能力,后者则开辟了内存优化的新维度。

一、装饰器

1.1 从一个实际需求说起

假设我们正在开发一个Web框架,需要为所有路由函数添加身份验证和日志记录功能。传统做法需要修改每个函数:

def user_profile(request):
    # 身份验证
    if not check_auth(request):
         return redirect('/login')
    
    # 日志记录
    log_access(request.user)
    
    # 业务逻辑
    return render_profile(request.user)

当需要修改认证逻辑时,必须逐个修改所有函数。这显然违反了DRY(Don't Repeat Yourself)原则。

1.2 装饰器的诞生

装饰器本质上是一个高阶函数,它接收一个函数作为参数,返回一个新的增强函数。

基础版装饰器:

def log_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"调用函数 {func.__name__}")
        return func(*args, **kwargs)
    return wrapper

@log_decorator
def add(a, b):
    return a + b

print(add(2,3))
# 输出:
# 调用函数 add
# 5

代码解析:

  1. @log_decorator 是语法糖,等价于 add = log_decorator(add)
  2. wrapper 函数保留了原函数的参数(*args, **kwargs
  3. 通过闭包特性保留了原函数的引用

1.3 进阶技巧:带参数的装饰器

当需要配置装饰器行为时(如设置日志级别),可以使用三层嵌套:

def repeat(num_times):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for _ in  range(num_times):
                result = func(*args, **kwargs)
            return result
        return wrapper
    return decorator

@repeat(num_times=3)
def greet(name):
    print(f"Hello {name}")

greet("Alice")
# 输出:
# Hello Alice
# Hello Alice
# Hello Alice

关键点:

  • 最外层函数接收装饰器参数
  • 中间层函数接收被装饰函数
  • 最内层函数实现增强逻辑

1.4 类装饰器的奥秘

装饰器不仅限于函数,还可以用类实现:

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.calls = 0

    def __call__(self, *args, **kwargs):
        self.calls += 1
        print(f"调用次数:{self.calls}")
        return self.func(*args, **kwargs)

@CountCalls
def  multiply(a, b):
    return a * b

print(multiply(2,3))  # 调用次数:1 → 6
print(multiply(4,5))  # 调用次数:2 → 20

魔法方法解析:

  • __call__ 方法使类实例成为可调用对象
  • 通过 self.func 保留原函数引用
  • 可以维护状态(如调用次数)

1.5 装饰器的实际应用场景

  1. 性能分析:记录函数执行时间



    
import time

def timer(func):
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        duration = time.perf_counter() - start
        print(f"{func.__name__} 执行耗时: {duration:.4f}秒")
        return result
    return wrapper

@timer
def complex_calculation():
    time.sleep(1.5)

complex_calculation()
# 输出:complex_calculation 执行耗时: 1.5001秒
  1. 缓存优化:记忆化(Memoization)
def memoize(func):
    cache = {}
    def wrapper(*args):
        if args in cache:
            return cache[args]
        result = func(*args)
        cache[args] = result
        return result
    return wrapper

@memoize
def fibonacci(n):
    return n if n 2else fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(30))  # 首次计算耗时,后续调用瞬间返回
  1. 权限控制:基于角色的访问控制


    
def admin_required(func):
    def wrapper(user, *args, **kwargs):
        if user.role != 'admin':
            raise PermissionError("需要管理员权限")
        return func(user, *args, **kwargs)
    return wrapper

class User:
    def __init__(self, role):
        self.role = role

@admin_required
def delete_user(user, user_id):
    print(f"删除用户 {user_id}")

admin = User('admin')
normal_user = User('user')

delete_user(admin, 123)     # 正常执行
delete_user(normal_user, 456)  # 抛出 PermissionError

二、生成器

2.1 从迭代器说起

生成器是迭代器的进化版,先回顾迭代器协议:



    
class Counter:
    def __init__(self, max_num):
        self.max_num = max_num
        self.current = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.current < self.max_num:
            self.current += 1
            return self.current
        else:
            raise StopIteration

for num in Counter(5):
    print(num)
# 输出:1 2 3 4 5

2.2 生成器的诞生:yield关键字

生成器使用 yield 语句替代复杂的迭代器实现:

def counter(max_num):
    current = 0
    while current < max_num:
        current += 1
        yield current
 
for num in counter(5):
    print(num)
# 输出:1 2 3 4 5

关键特性:

  • 惰性求值:每次产生一个值,不保存全部结果
  • 状态保存:yield 会保留函数执行上下文
  • 内存高效:适合处理大数据流

2.3 生成器表达式

类似列表推导式,但使用圆括号:

gen = (x**2 for x in range(10))
print(list(gen))  # [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

2.4 协程与生成器的进阶应用

生成器可以作为简单的协程使用,通过 send() 方法发送值:

def coroutine():
    while True:
        received = yield
        print(f"收到消息: {received}")
 
c = coroutine()
next(c)  # 启动生成器,执行到第一个yield
c.send("Hello")  # 收到消息: Hello
c.send("World")  # 收到消息: World

2.5 生成器的实际应用场景

  1. 大数据处理:逐行读取文件
def read_large_file(file_path):
    with open(file_path, 'r'as f:
        for line in f:
            yield line.strip()
 
for line in read_large_file('big_data.txt'):
    process(line)  # 逐行处理,内存友好
  1. 数学序列生成:斐波那契数列
def fibonacci():
    a, b = 01
    while True:
        yield a
        a, b = b, a + b
 
fib = fibonacci()
print([next(fib) for _ in range(10)])  # [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
  1. 管道处理:数据流水线
def multiply_by(n):
    whileTrue:
        num = yield
        yield num * n

def add(x):
    whileTrue:
        num = yield
        yield num + x

pipeline = multiply_by(2) | add(5)  # 需要Python 3.5+的生成器管道支持

for num in pipeline:
    print(num)
# 输入序列:1,2,3 → 输出:7,9,11

三、装饰器与生成器的协同作战

3.1 生成器装饰器模式

为生成器添加预处理/后处理逻辑:

def validate_output(generator):
    for value in generator:
        if value 0:
            raise ValueError("生成器产生负数")
        yield value

@validate_output
def temperature_sensor():
    yield25.5
    yield-3.2# 会触发异常
    yield28.1

for temp in temperature_sensor():
    print(f"当前温度: {temp}℃")
# 输出:
# 当前温度: 25.5℃
# 抛出 ValueError: 生成器产生负数

3.2 异步任务调度

结合生成器和装饰器实现简单任务队列:

import time
from collections import deque

class TaskScheduler:
    def __init__(self):
        self.tasks = deque()
    
    def add_task(self, task):
        self.tasks.append(task)
    
    def run(self):
        while self.tasks:
            task = self.tasks.popleft()
            try:
                next(task)
                self.tasks.append(task)
            except StopIteration:
                print(f"任务完成: {task.name}")

def task(func):
    def wrapper(*args, **kwargs):
        task_gen = func(*args, **kwargs)
        task_gen.name = func.__name__
        next(task_gen)  # 预启动到第一个yield
        return task_gen
    return wrapper

scheduler = TaskScheduler()

@task
def download_file():
    print("开始下载...")
    time.sleep(2)
    print("下载完成")
    yield

@task
def process_data():
    print("处理数据中...")
    time.sleep(1)
    print("处理完成")
    yield

scheduler.add_task(download_file())
scheduler.add_task(process_data())
scheduler.run()
# 输出:
# 开始下载...
# 处理数据中...
# 下载完成
# 处理完成
# 任务完成: download_file
# 任务完成: process_data

四、深入理解底层机制

4.1 装饰器的函数签名问题

装饰器可能破坏原函数的元数据,使用 functools.wraps 修复:

from functools import wraps

def my_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        print("装饰器生效")
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def example():
    """示例函数"""
    pass

print(example.__name__)  # 输出 example(而非 wrapper)
print(example.__doc__)   # 输出 "示例函数"

4.2 生成器的执行流程

通过反汇编观察生成器状态:

import dis

def simple_generator():
    yield1
    yield2

dis.dis(simple_generator)
# 输出包含:
# LOAD_CONST               1 (1)
# YIELD_VALUE
# POP_TOP
# LOAD_CONST               2 (2)
# YIELD_VALUE
# POP_TOP
# LOAD_CONST               0 (None)
# RETURN_VALUE

4.3 生成器的内存优势验证

对比列表和生成器的内存使用:

import sys
 
# 列表方式
list_range = list(range(10**6))
print(f"列表内存占用: {sys.getsizeof(list_range)} bytes")   # 约8MB
 
# 生成器方式
gen_range = (x for x in range(10**6))
print(f"生成器内存占用: {sys.getsizeof(gen_range)} bytes")  # 约128B

五、如何正确使用装饰器、生成器

5.1 装饰器使用原则

  1. 保持装饰器简单:避免在装饰器中实现复杂业务逻辑
  2. 明确装饰器作用:通过命名清晰表达功能(如 @login_required
  3. 使用 wraps 保持元数据
  4. 避免多层嵌套装饰器的执行顺序问题

5.2 生成器使用场景选择

  1. 优先选择生成器表达式而非列表推导式处理大数据
  2. 需要状态保持时使用生成器函数
  3. 复杂数据流处理使用生成器管道
  4. 避免在生成器中执行阻塞IO操作(考虑使用异步框架)

5.3 调试技巧

  1. 使用 inspect 模块检查装饰器:
import inspect
 
print(inspect.getsource(example))  # 查看被装饰函数源码
  1. 生成器调试:
gen = counter(3)
print(next(gen))  # 1
print(gen.gi_frame.f_lasti)  # 查看当前指令位置

- END -

推荐课程

《Python ·  AI&数据科学入门》

图片

点这里👇关注我,回复“ python”了解课程

往期精彩阅读

👉 kaggle比赛baseline合集

👉经典论文推荐合集

👉人工智能必读书籍

👉本专科硕博学习经验

10个赞学姐的午饭就可以有个鸡腿🍗

图片

Python社区是高质量的Python/Django开发社区
本文地址:http://www.python88.com/topic/181700
 
63 次点击