036、装饰器三部曲（一）：无参装饰器的原理与 10 个实用案例

036、装饰器三部曲（一）：无参装饰器的原理与 10 个实用案例

上周帮一个同事排查线上接口响应时间异常的问题。他写了一个统计函数执行时间的装饰器，结果发现每次调用都多打印了一行“None”，而且时间统计完全不准。我一看代码，他直接在装饰器内部调用了被装饰函数，却没有返回结果——典型的“装饰器新手翻车现场”。这个场景让我决定把装饰器这个主题拆成三篇来讲，今天先搞定最基础的无参装饰器。

装饰器到底是个什么东西

别被“装饰器”这个名字唬住。本质上，装饰器就是一个接受函数作为参数、返回一个新函数的函数。Python里函数是一等公民，可以像变量一样传来传去，这是装饰器能工作的根基。

看一个最朴素的例子：

def my_decorator(func):
def wrapper():
print("调用前，这里可以加日志")
func() # 这里踩过坑：别忘了调用原函数
print("调用后，这里可以加清理逻辑")
return wrapper

def say_hello():
print("你好，世界")

say_hello = my_decorator(say_hello)
say_hello()

输出：

调用前，这里可以加日志
你好，世界
调用后，这里可以加清理逻辑

这就是装饰器的原始形态——手动把函数传进去，再手动把新函数赋回去。Python的@语法糖只是帮我们省掉了最后两行赋值操作。

@语法糖是怎么工作的

用@改写上面的例子：

def my_decorator(func):
def wrapper():
print("调用前")
func()
print("调用后")
return wrapper

@my_decorator
def say_hello():
print("你好，世界")

say_hello()

@my_decorator等价于执行了say_hello = my_decorator(say_hello)。注意这个执行时机——它发生在函数定义时，而不是函数调用时。很多新手以为装饰器每次调用都会执行，其实只在模块加载时执行一次。

被装饰函数的元信息丢失问题

用上面的装饰器装饰后，say_hello.__name__会变成wrapper而不是say_hello。这在调试和文档生成时很坑。

print(say_hello.__name__) # 输出: wrapper

解决方案是用functools.wraps：

import functools

def my_decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
print("调用前")
return func(*args, **kwargs)
return wrapper

@my_decorator
def say_hello():
print("你好，世界")

print(say_hello.__name__) # 输出: say_hello

@functools.wraps(func)会把原函数的__name__、__doc__、__module__等属性复制到wrapper上。别这样写——不写@functools.wraps，除非你明确知道自己在做什么。

10个实用案例

案例1：函数执行时间统计

import time
import functools

def timer(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
end = time.perf_counter()
print(f"{func.__name__} 执行耗时: {end – start:.4f}秒")
return result # 这里踩过坑：不return的话调用方拿不到结果
return wrapper

@timer
def slow_function():
time.sleep(0.5)
return "完成"

result = slow_function()
print(result) # 输出: 完成

案例2：日志记录

import functools
import logging

logging.basicConfig(level=logging.INFO)

def log_call(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
logging.info(f"调用 {func.__name__}，参数: {args}, {kwargs}")
try:
result = func(*args, **kwargs)
logging.info(f"{func.__name__} 返回: {result}")
return result
except Exception as e:
logging.error(f"{func.__name__} 抛出异常: {e}")
raise
return wrapper

@log_call
def divide(a, b):
return a / b

divide(10, 2) # 正常调用
divide(10, 0) # 异常调用

案例3：访问控制与权限校验

import functools

def require_admin(func):
@functools.wraps(func)
def wrapper(user, *args, **kwargs):
if not getattr(user, 'is_admin', False):
raise PermissionError("只有管理员才能执行此操作")
return func(user, *args, **kwargs)
return wrapper

class User:
def __init__(self, name, is_admin=False):
self.name = name
self.is_admin = is_admin

@require_admin
def delete_user(admin_user, user_id):
print(f"管理员 {admin_user.name} 删除了用户 {user_id}")

admin = User("Alice", is_admin=True)
normal = User("Bob")

delete_user(admin, 123) # 正常
delete_user(normal, 456) # 抛出异常

案例4：缓存计算结果

import functools

def cache_result(func):
cache = {}
@functools.wraps(func)
def wrapper(*args, **kwargs):
key = (args, tuple(sorted(kwargs.items())))
if key in cache:
print(f"命中缓存: {key}")
return cache[key]
result = func(*args, **kwargs)
cache[key] = result
return result
return wrapper

@cache_result
def expensive_calc(n):
print(f"正在计算 {n}…")
return n * n

print(expensive_calc(5)) # 计算
print(expensive_calc(5)) # 命中缓存

案例5：重试机制

import functools
import time
import random

def retry(max_attempts=3, delay=1):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(1, max_attempts + 1):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts:
raise
print(f"第{attempt}次失败: {e}，{delay}秒后重试")
time.sleep(delay)
return None
return wrapper
return decorator

@retry(max_attempts=3, delay=0.5)
def unstable_api():
if random.random() < 0.7:
raise ConnectionError("网络不稳定")
return "成功"

print(unstable_api())

案例6：输入参数校验

import functools

def validate_positive(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for i, arg in enumerate(args):
if isinstance(arg, (int, float)) and arg < 0:
raise ValueError(f"参数 {i} 必须为正数，收到 {arg}")
for key, value in kwargs.items():
if isinstance(value, (int, float)) and value < 0:
raise ValueError(f"参数 {key} 必须为正数，收到 {value}")
return func(*args, **kwargs)
return wrapper

@validate_positive
def sqrt_approx(x):
return x ** 0.5

print(sqrt_approx(9)) # 正常
print(sqrt_approx(–1)) # 抛出异常

案例7：单例模式

import functools

def singleton(cls):
instances = {}
@functools.wraps(cls)
def get_instance(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
return instances[cls]
return get_instance

@singleton
class Database:
def __init__(self):
print("初始化数据库连接")

db1 = Database()
db2 = Database()
print(db1 is db2) # 输出: True

案例8：函数注册表

import functools

plugins = {}

def register_plugin(name):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
plugins[name] = wrapper
return wrapper
return decorator

@register_plugin("format_json")
def format_json(data):
import json
return json.dumps(data, indent=2)

@register_plugin("format_yaml")
def format_yaml(data):
return str(data)

print(plugins.keys()) # 输出: dict_keys(['format_json', 'format_yaml'])

案例9：调试与堆栈追踪

import functools
import traceback

def debug_call(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
print(f"进入 {func.__name__}")
try:
result = func(*args, **kwargs)
print(f"离开 {func.__name__}，返回 {result}")
return result
except Exception:
print(f"{func.__name__} 异常:")
traceback.print_exc()
raise
return wrapper

@debug_call
def buggy_function(x):
return 1 / x

buggy_function(0)

案例10：限流控制

import functools
import time

def rate_limit(max_calls=5, period=10):
calls = []
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
now = time.time()
calls[:] = [t for t in calls if now – t < period]
if len(calls) >= max_calls:
raise RuntimeError(f"超过限流限制: {period}秒内最多{max_calls}次")
calls.append(now)
return func(*args, **kwargs)
return wrapper
return decorator

@rate_limit(max_calls=3, period=5)
def api_call():
print("API调用成功")

for _ in range(4):
try:
api_call()
except RuntimeError as e:
print(e)
time.sleep(0.5)

个人经验性建议

下一篇会讲带参数的装饰器，以及如何用类实现装饰器。如果你在写装饰器时遇到过什么奇葩bug，欢迎在评论区分享，说不定下一篇就会拿你的案例来讲解。