之前聊了 @property、@classmethod、@staticmethod 这三个 Python 内置的装饰器,但装饰器真正的威力在于“自己写”。
下面就来系统梳理一下怎么写自定义装饰器,以及落地到业务里怎么用。
很多人一开始接触装饰器,看到那个 @ 符号就觉得神秘。其实它没那么玄乎。
装饰器本质上就是一个函数,只不过它的参数是另一个函数,返回值还是一个函数。
就这么简单,@ 只是 Python 提供的一种语法糖。例如:
# 这两段代码完全等价
# 写法1:用 @ 语法糖
@decorator
def func():
pass
# 写法2:不用语法糖,手动传参
def func():
pass
func = decorator(func)看出来没有?@decorator 的意思就是把下面定义的函数名,当作参数传给 decorator,然后把返回的新函数重新赋值给原来的函数名。
所以装饰器的核心作用就是:不改原函数源码,不改调用方式,在原函数执行前后插入额外的逻辑。
这种“横向切入”的能力,在编程里叫 AOP(面向切面编程)。日志、计时、权限校验、缓存、异常兜底 —— 这些跟核心业务无关但又到处需要的逻辑,用装饰器最合适。
如果你学过 Java 和 Spring,这不就是 Java 的动态代理和 Spring 的 AOP 编程吗。
直接上代码,把骨架结构拆出来:
def my_decorator(func): # 外层:接收被装饰的函数
# 参数 func 就是函数
def wrapper(*args, **kwargs): # 内层:真正干活的包装函数
# ===== 前置逻辑 =====
print(f"准备调用 {func.__name__}")
result = func(*args, **kwargs) # 调用原函数
# ===== 后置逻辑 =====
print(f"{func.__name__} 执行完了")
return result # 把原函数的返回值原样返回
return wrapper # 把包装后的函数返回出去
@my_decorator
def greet(name):
print(f"你好,{name}")
greet("小明")运行输出:
准备调用 greet
你好,小明
greet 执行完了几个关键点值得记住:
*args, **kwargs 是万能参数收集,保证不管原函数长什么样,装饰器都能接得住。
一定要 return 原函数的返回值,不然原函数如果有返回内容就丢了。
wrapper 内部的 func 是通过闭包捕获的,所以内层函数能访问到它。
下面将通过示例来演示如何自定义装饰器。
实际写代码时,基本就是下面这四种变体。掌握了就能覆盖 90% 的场景。
最常用的一类,装饰器本身不需要任何配置参数:
import time
def timer(func):
"""
无参函数装饰器:统计被装饰函数的执行耗时
:param func: 被包裹的目标原始函数
:return wrapper: 包装后的新函数,新增计时逻辑
"""
def wrapper(*args, **kwargs):
"""
内层包装函数,接管原函数的全部位置参数、关键字参数
*args:接收任意长度位置参数
**kwargs:接收任意关键字参数,保证兼容所有原函数入参形式
"""
# 记录函数正式执行前的高精度时间戳
start = time.perf_counter()
# 执行原始业务函数,传入全部参数,接收原函数返回结果
result = func(*args, **kwargs)
# 计算时间差值:当前时间 - 起始时间,得到函数运行总耗时
elapsed = time.perf_counter() - start
print(f"⏱ {func.__name__} 耗时:{elapsed:.4f}秒")
# 将原函数的执行结果原样返回,保证外部调用逻辑不受装饰器影响
return result
# 返回包装函数,替换掉原来的目标函数
return wrapper
# 使用语法糖 @timer 等价执行:compute_sum = timer(compute_sum)
# 给 compute_sum 挂载计时增强逻辑
@timer
def compute_sum(n):
"""计算 0 ~ n-1 所有整数的累加和"""
return sum(range(n))
compute_sum(10000000)
# ⏱ compute_sum 耗时:0.1735秒这里用 perf_counter() 而不是 time(),是因为前者精度更高且不受系统时间调整影响 —— 做性能测试一定要养成这个小习惯。
有时候装饰器本身需要一些配置,比如日志前缀、开关控制。这时候就需要再往外套一层函数来接收参数,例如:
import time
def timer_with_label(label: str):
"""
第一层外层函数:接收装饰器自定义配置参数
:param label: 自定义业务标签,用来区分不同场景的计时日志(如数据库查询、文件读取)
:return actual_decorator: 真正接收目标函数的标准装饰器
"""
def actual_decorator(func):
"""
第二层标准装饰器函数:接收待增强的原始业务函数
:param func: 被装饰的目标函数(query_db / read_file)
:return wrapper: 包装后的增强函数
"""
def wrapper(*args, **kwargs):
"""
第三层执行包装函数:承载前置计时、执行原函数、统计耗时、后置打印日志整套逻辑
*args:兼容原函数任意位置参数
**kwargs:兼容原函数任意关键字参数,保证不破坏原有传参方式
"""
# 记录函数运行前的高精度时间戳
start = time.perf_counter()
# 执行原始业务函数,透传所有入参,接收函数返回值
result = func(*args, **kwargs)
# 计算函数整体运行耗时
cost = time.perf_counter() - start
print(f"[{label}] {func.__name__} 耗时 {cost:.4f}s")
# 原样返回原函数执行结果,对外调用无感知
return result
return wrapper
return actual_decorator
# 语法糖等价:query_db = timer_with_label("数据库查询")(query_db)
# 先传入标签参数生成装饰器,再把query_db交给装饰器包装
@timer_with_label(label="数据库查询")
def query_db(sql):
"""模拟数据库查询操作,休眠0.3秒模拟IO耗时"""
time.sleep(0.3)
return f"结果:{sql}"
@timer_with_label(label="文件读取")
def read_file(path):
"""模拟本地文件读取操作,休眠0.1秒模拟IO耗时"""
time.sleep(0.1)
return "文件内容..."
if __name__ == "__main__":
# 调用被装饰后的数据库查询函数,自动执行计时逻辑
print(query_db("SELECT * FROM users"))
# [数据库查询] query_db 耗时 0.3012s
# 结果:SELECT * FROM users
# 调用被装饰后的文件读取函数
print(read_file("/tmp/data.txt"))
# [文件读取] read_file 耗时 0.1015s
# 文件内容...理解三层嵌套的关键:
调用 timer_with_label(label="数据库查询"),返回 actual_decorator。
actual_decorator(query_db),返回 wrapper。
后续每次调用 query_db(),实则是调用 wrapper。
初学者最容易在这里绕晕。我的记忆技巧是:"带参装饰器就是在无参装饰器外面再包一层参数接收函数" —— 每次需要多传什么,就往外加一层。Python 的装饰器就是靠闭包层层传的,没有捷径。
如果装饰器需要维护状态(比如调用计数、缓存数据),用类来实现会更自然。例如:
class CountCalls:
"""统计函数被调用次数的类装饰器
依靠类的 __call__ 魔法方法,让类实例变成可调用对象,实现装饰器能力
"""
def __init__(self, func):
"""
初始化方法:绑定需要被装饰的目标函数,初始化调用计数器
:param func: 传入被包装的原始业务函数
"""
# 将目标函数保存到实例属性,后续执行时调用
self.func = func
# 实例属性计数器,记录当前函数累计调用次数
self.count = 0
def __call__(self, *args, **kwargs):
"""
魔法方法:当把类实例当作函数加括号调用时,会自动执行该方法
*args:接收调用函数时所有位置参数
**kwargs:接收调用函数时所有关键字参数
"""
self.count += 1
print(f"→ {self.func.__name__} 被调用了 {self.count} 次")
# 执行原始业务函数,透传全部入参,并将原函数执行结果向上返回
return self.func(*args, **kwargs)
# 语法糖 @CountCalls 等价于:process_task = CountCalls(process_task)
# 1. 实例化 CountCalls 对象,把原始 process_task 传入 __init__
# 2. 后续执行 process_task() 本质是执行实例.__call__()
@CountCalls
def process_task():
"""模拟业务任务处理函数"""
print("处理任务...")
# 第一次调用,触发 __call__
process_task()
# 第二次调用
process_task()
# 第三次调用
process_task()运行代码,输出:
→ process_task 被调用了 1 次
处理任务...
→ process_task 被调用了 2 次
处理任务...
→ process_task 被调用了 3 次
处理任务...注意几点:
@CountCalls 等价于 process_task = CountCalls(process_task),创建了一个实例。
实例被调用时触发 __call__,所以写在 __call__ 里的逻辑就是装饰逻辑。
状态(self.count)挂在实例上,天然持久化,比函数闭包里搞个 nonlocal 变量优雅得多。
类装饰器也能带参数,套路跟函数一样——在 __init__ 里接收:
import time
class Retry:
"""
类形式实现的带参重试装饰器
功能:函数抛出异常时自动间隔等待并重试,达到最大次数仍失败则抛出原始异常
"""
def __init__(self, max_attempts: int = 3, delay: float = 1.0):
"""
装饰器初始化阶段,接收装饰器自身配置参数
:param max_attempts: 最大重试总次数
:param delay: 每次失败后重试之间的休眠等待时长,单位秒
"""
self.max_attempts = max_attempts
self.delay = delay
def __call__(self, func):
"""
__call__ 让类实例变成可调用对象;接收被装饰的目标业务函数
:param func: 需要加上重试逻辑的原始函数
:return wrapper: 包裹了重试逻辑的增强函数
"""
# 在作用域内导入time,仅当前包装逻辑使用
import time as _time
def wrapper(*args, **kwargs):
"""
真正执行拦截、重试逻辑的内层包装函数
*args/**kwargs 兼容原始函数任意形式的入参
"""
# 循环遍历每一次尝试,次数从1开始计数
for attempt in range(1, self.max_attempts + 1):
try:
# 正常执行原始业务函数,执行无异常则直接return,终止重试
return func(*args, **kwargs)
except Exception as e:
# 捕获函数运行抛出的任意异常
# 判断:已经是最后一轮重试,不再继续重试,直接向上抛出异常终止流程
if attempt == self.max_attempts:
raise
print(f" [重试] 第 {attempt} 次失败: {e},{self.delay}秒后重试...")
# 休眠指定时间,间隔一段时间再发起下一次调用
_time.sleep(self.delay)
return wrapper
# @Retry(max_attempts=3, delay=0.5)
# 执行流程:
# 1. 先实例化 Retry 对象,传入最大重试次数3、间隔0.5秒
# 2. 实例自动调用 __call__,接收 unstable_request 生成 wrapper
# 3. unstable_request 最终指向包装后的 wrapper
@Retry(max_attempts=3, delay=0.5)
def unstable_request():
"""模拟不稳定的网络请求,大概率随机抛出网络异常"""
import random
# 70% 的概率触发异常,模拟网络抖动、接口报错等临时故障
if random.random() < 0.7:
raise ConnectionError("网络抖动")
# 30% 概率执行成功,返回正常结果
return "请求成功"
# 调用时自动走wrapper的重试逻辑
print(unstable_request())运行代码,输出:
[重试] 第 1 次失败: 网络抖动,0.5秒后重试...
请求成功为什么不在 __init__ 中接收 func 呢?
优先运行 Retry(max_attempts=3, delay=0.5),触发 __init__,此时我们手里只有配置,还不知道要装饰哪个函数,自然没办法在 __init__ 接收 func。
执行完毕得到一个携带配置的 Retry 实例对象。Python 拿着上面得到的实例,去绑定下方函数,
等价执行 “实例(unstable_request)”。对象加括号调用,必然触发 __call__(self, func),这时才拿到目标函数。
__init__ 的职责:接收装饰器自身的配置参数(重试次数、延迟时间、标签、阈值等),初始化配置、构建实例,此时还未关联业务函数。
__call__ 的职责:拿到已经配置好的实例,再接收需要被装饰的业务函数,生成包裹重试逻辑的 wrapper。后续调用函数 func() 时,依旧走 __call__ 内部的 wrapper。
被装饰后的函数,__name__ 和 __doc__ 会变成 wrapper 的信息。这会影响调试、文档生成、以及一些依赖这些元信息的框架(比如 Flask 的路由)。例如:
def bad_decorator(func):
def wrapper(*args, **kwargs):
"""wrapper 的文档"""
return func(*args, **kwargs)
return wrapper
@bad_decorator
def my_func():
"""my_func 的文档"""
pass
print(my_func.__name__) # wrapper ← 不对!
print(my_func.__doc__) # wrapper 的文档 ← 丢了!解决办法:functools.wraps,一行搞定,例如:
from functools import wraps
def good_decorator(func):
@wraps(func) # 把 func 的元信息拷贝给 wrapper
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
return wrapper
@good_decorator
def my_func():
"""my_func 的文档"""
pass
print(my_func.__name__) # my_func ← 对了
print(my_func.__doc__) # my_func 的文档 ← 也对了@wraps 的原理很简单,就是把原函数的 __name__、__doc__、__module__、__qualname__ 等属性复制到 wrapper 上。本质上就是 functools.update_wrapper 的语法糖。
写装饰器一定要加 @wraps,这是基本功。 记住这个原则就行,不用去纠结它内部怎么实现的。
上面讲的是骨架,这里放几个能直接用到真实项目里的例子。
接口挂了不要直接崩,优雅兜底。例如:
from functools import wraps
import traceback
def safe_call(default_return=None, log_error: bool = True):
"""捕获异常并返回默认值,防止函数崩溃影响上层"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except Exception as e:
if log_error:
print(f"[异常] {func.__name__} 出错了")
print(f" 参数: args={args}, kwargs={kwargs}")
print(f" 详情: {type(e).__name__}: {e}")
return default_return
return wrapper
return decorator
# 比如有个调用第三方 API 的函数,对方服务不稳定
@safe_call(default_return={"err": "服务不可用"})
def get_user_info(user_id):
# 模拟偶尔超时
import random
if random.random() < 0.5:
raise TimeoutError("请求超时")
return {"name": f"用户{user_id}", "age": 25}
# 连续调几次看看
for i in range(5):
result = get_user_info(i)
print(result)运行代码,输出:
[异常] get_user_info 出错了
参数: args=(0,), kwargs={}
详情: TimeoutError: 请求超时
{'err': '服务不可用'}
{'name': '用户1', 'age': 25}
{'name': '用户2', 'age': 25}
{'name': '用户3', 'age': 25}
{'name': '用户4', 'age': 25}safe_call 是我实际项目里最常用的装饰器。第三方接口、数据库操作、文件读写 —— 这些不可控的地方全都套一层,比到处写 try/except 干净太多。
Web 后端接口里的典型场景:
from functools import wraps
def require_auth(fail_response=None):
"""校验用户是否已登录"""
def decorator(func):
@wraps(func)
def wrapper(request, *args, **kwargs):
user = request.get("user", {})
if not user.get("is_login"):
print(f" [拦截] 未登录用户尝试访问 {func.__name__}")
return fail_response or {"code": 401, "msg": "请先登录"}
return func(request, *args, **kwargs)
return wrapper
return decorator
@require_auth(fail_response={"code": 401, "msg": "请先登录"})
def get_profile(request):
return {"name": request["user"]["name"], "level": "VIP"}
@require_auth(fail_response={"code": 401, "msg": "请先登录"})
def edit_profile(request, nickname):
request["user"]["nickname"] = nickname
return {"code": 200, "msg": "修改成功"}
# 未登录用户
req1 = {"user": {"name": "张三", "is_login": False}}
print(get_profile(req1)) # {"code": 401, "msg": "请先登录"}
# 已登录用户
req2 = {"user": {"name": "张三", "is_login": True}}
print(get_profile(req2)) # {"name": "张三", "level": "VIP"}运行代码,输出如下:
[拦截] 未登录用户尝试访问 get_profile
{'code': 401, 'msg': '请先登录'}
{'name': '张三', 'level': 'VIP'}这种模式在 Flask/Django 里用得很多。实际项目里你可以把 request 的来源改成框架提供的全局 request 对象,装饰器就更简洁了。
对于计算量大或者调用频繁的函数,加个结果缓存能明显提升性能。例如:
from functools import wraps
def cache_result(func):
"""简单的结果缓存,基于函数参数做 key"""
cache = {}
@wraps(func)
def wrapper(*args, **kwargs):
# 构造缓存键:把 args 和 kwargs 拼成一个可哈希的 tuple
cache_key = (args, tuple(sorted(kwargs.items())))
if cache_key in cache:
print(f" [命中缓存] {func.__name__}{args}")
return cache[cache_key]
result = func(*args, **kwargs)
cache[cache_key] = result
print(f" [写入缓存] {func.__name__}{args}")
return result
return wrapper
@cache_result
def fib(n):
"""递归斐波那契——不缓存会指数爆炸"""
if n <= 1:
return n
return fib(n - 1) + fib(n - 2)
# 第一次算 fib(30),后面再调就直接走缓存了
print(fib(30))不过要注意,这个实现很粗糙 —— 缓存无限增长,参数里有不可哈希对象(list、dict)就会炸。生产环境建议直接上 functools.lru_cache 或 functools.cache(Python 3.9+)。
这里放一个是为了让你理解缓存装饰器的原理。理解了再去看标准库里那些高级实现会轻松很多。
比基础计时器更进一步,超过阈值时打个警告:
import time
from functools import wraps
def monitor(threshold: float = 1.0):
"""监控函数耗时,超过阈值时告警"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
level = "正常"
if elapsed > threshold * 3:
level = "严重延迟"
elif elapsed > threshold:
level = "略慢"
print(f"[{level}] {func.__name__} 耗时 {elapsed:.3f}s (阈值:{threshold}s)")
return result
return wrapper
return decorator
@monitor(threshold=0.1)
def query_user(user_id):
time.sleep(0.08) # 模拟正常查询
return f"用户_{user_id}"
@monitor(threshold=0.1)
def slow_query(user_id):
time.sleep(0.5) # 模拟慢查询
return f"用户_{user_id}"
query_user(1) # [正常] query_user 耗时 0.080s (阈值:0.1s)
slow_query(2) # [严重延迟] slow_query 耗时 0.501s (阈值:0.1s)运行代码,输出:
[正常] query_user 耗时 0.080s (阈值:0.1s)
[严重延迟] slow_query 耗时 0.501s (阈值:0.1s)这个装饰器在实际做性能优化排查时非常有用 —— 先全局套上,跑一遍压测,哪些函数超时就一目了然了。
一个函数可以被多个装饰器修饰,它们的执行顺序要搞清楚:
@deco_a
@deco_b
@deco_c
def target():
print("核心逻辑")等价于:
target = deco_a(deco_b(deco_c(target)))执行顺序遵循 "剥洋葱" 模型:
前置拦截逻辑(函数运行之前):外层装饰器最先执行,顺序:deco_a → deco_b → deco_c
后置收尾逻辑(函数运行结束之后):内层装饰器先收尾,顺序:deco_c → deco_b → deco_a
看个例子就明白了:
from functools import wraps
def deco_a(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(" [A] 前置")
result = func(*args, **kwargs)
print(" [A] 后置")
return result
return wrapper
def deco_b(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(" [B] 前置")
result = func(*args, **kwargs)
print(" [B] 后置")
return result
return wrapper
def deco_c(func):
@wraps(func)
def wrapper(*args, **kwargs):
print(" [C] 前置")
result = func(*args, **kwargs)
print(" [C] 后置")
return result
return wrapper
@deco_a
@deco_b
@deco_c
def say_hello():
print(" hello!")
say_hello()输出:
[A] 前置
[B] 前置
[C] 前置
hello!
[C] 后置
[B] 后置
[A] 后置一目了然。C 在最里层,A 在最外层。把这个顺序记住,调试多个装饰器叠加的代码时才不会晕。
装饰器用在类的方法上,和非类函数有一些微妙差别。
装饰实例方法时,wrapper 的第一个参数会自然地收到 self,因为 Python 会自动把实例作为第一个参数传进来。用 *args, **kwargs 就够了:
from functools import wraps
def log_method(func):
@wraps(func)
def wrapper(*args, **kwargs):
# args[0] 就是 self
class_name = args[0].__class__.__name__
print(f"[调用] {class_name}.{func.__name__}")
return func(*args, **kwargs)
return wrapper
class UserService:
@log_method
def create(self, name):
print(f"创建用户:{name}")
@log_method
def delete(self, user_id):
print(f"删除用户:{user_id}")
svc = UserService()
svc.create("小明")
svc.delete(100)输出:
[调用] UserService.create
创建用户:小明
[调用] UserService.delete
删除用户:100@staticmethod 和 @classmethod 很特殊 —— 它们和自定义装饰器叠加时,注意顺序:
import time
def timer(func):
"""
无参函数装饰器:统计被装饰函数的执行耗时
:param func: 被包裹的目标原始函数
:return wrapper: 包装后的新函数,新增计时逻辑
"""
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
elapsed = time.perf_counter() - start
print(f"⏱ {func.__name__} 耗时:{elapsed:.4f}秒")
return result
return wrapper
class Demo:
# @timer
@staticmethod # ← 先让 Python 处理成静态方法
@timer # ← 再加计时装饰
def do_work():
time.sleep(0.1)
print("干活")
d = Demo()
d.do_work() # 调用静态方法运行代码,输出:
干活
⏱ do_work 耗时:0.1008秒如果将 @timer 放在 @staticmethod 上面,如下:
class Demo:
@timer
@staticmethod # ← 先让 Python 处理成静态方法
def do_work():
time.sleep(0.1)
print("干活")再次运行,抛出错误:
Traceback (most recent call last):
File "d:\python_demo\demo.py", line 27, in <module>
d.do_work() # 调用静态方法
~~~~~~~~~^^
File "d:\share_dir\workspace\5.demo\python_demo\demo.py", line 11, in wrapper
result = func(*args, **kwargs)
TypeError: Demo.do_work() takes 0 positional arguments but 1 was given实际上,只要你把 @staticmethod/@classmethod 放最外层,自定义装饰器放下面,一般不会出问题。反之可能报错,因为自定义装饰器返回的 wrapper 不一定是 descriptor。
虽然平时很少手写单例(有更好的方式),但用装饰器实现单例是一个很好的学习案例:
from functools import wraps
def singleton(cls):
"""类装饰器:把类变成单例"""
instances = {}
@wraps(cls)
def get_instance(*args, **kwargs):
if cls not in instances:
instances[cls] = cls(*args, **kwargs)
print(f" [单例] 创建 {cls.__name__} 实例")
else:
print(f" [单例] 返回已存在的 {cls.__name__} 实例")
return instances[cls]
return get_instance
@singleton
class DatabaseConnection:
def __init__(self, host="localhost"):
self.host = host
print(f"初始化连接: {host}")
# 多次"实例化"
conn1 = DatabaseConnection("db1.example.com")
conn2 = DatabaseConnection("db2.example.com")
conn3 = DatabaseConnection()
print(f"conn1 是 conn2? {conn1 is conn2}") # True
print(f"同一连接: {conn1.host}") # db1.example.com运行代码,输出:
初始化连接: db1.example.com
[单例] 创建 DatabaseConnection 实例
[单例] 返回已存在的 DatabaseConnection 实例
[单例] 返回已存在的 DatabaseConnection 实例
conn1 是 conn2? True
同一连接: db1.example.com这个例子也展示了 "装饰器不一定只能修饰函数" —— 修饰类也是完全合法的。
把装饰器和 Python 的类型注解结合起来,能做出一个简易的参数校验框架:
from functools import wraps
# 导入内置签名解析工具,解析函数形参、绑定实际传入参数
import inspect
def validate_types(func):
"""
无参装饰器:基于函数的类型注解,在运行时自动校验入参数据类型
工作逻辑:读取函数标注的参数期望类型,调用函数时挨个比对实参类型,
类型不匹配直接抛出TypeError异常拦截执行
:param func: 被装饰、需要开启参数类型校验的目标函数
:return wrapper: 包装了类型校验逻辑的增强函数
"""
@wraps(func)
def wrapper(*args, **kwargs):
# __annotations__:内置属性,取出函数上标记的全部类型注解 {参数名: 目标类型}
hints = func.__annotations__
# 获取目标函数的签名结构(形参名称、参数顺序、默认值等定义信息)
sig = inspect.signature(func)
# 将本次调用传入的 args、kwargs 按照函数签名绑定,映射为「形参名: 实际传入值」
bound = sig.bind(*args, **kwargs)
# 自动填充函数定义中带默认值但本次调用未传递的参数
bound.apply_defaults()
# 遍历本次调用所有绑定好的「参数名-实际值」键值对
for name, value in bound.arguments.items():
# 当前参数设置了类型注解,才需要做校验
if name in hints:
# 取出该参数期望的合法类型
expected_type = hints[name]
# 判断实际传入值的类型是否属于期望类型
if not isinstance(value, expected_type):
raise TypeError(
f"参数 '{name}' 期望 {expected_type.__name__},"
f"实际收到 {type(value).__name__}({value!r})"
)
return func(*args, **kwargs)
return wrapper
# 挂载类型校验装饰器:greet = validate_types(greet)
# 标注三个参数的期望类型:name字符串、age整数、score浮点型(带默认值)
@validate_types
def greet(name: str, age: int, score: float = 0.0):
"""测试打印用户基础信息"""
print(f"{name}, {age}岁, 分数{score}")
# 合法入参,全部类型匹配,校验放行正常执行
greet("小明", 18, 95.5)
# 非法入参:age传入字符串而非int,装饰器会在校验阶段抛出异常
try:
greet("小明", "十八")
except TypeError as e:
print(f"校验失败:{e}")运行代码,输出:
小明, 18岁, 分数95.5
校验失败:参数 'age' 期望 int,实际收到 str('十八')这对于没有类型检查的开发阶段很有用 —— 比 Pylance/MyPy 更暴力,直接在运行时拦住。