Python3 基础教程

Python3 自定义装饰器

🎉摘要:系统掌握Python自定义装饰器的编写方法,涵盖无参、带参、类装饰器四种形式,通过异常兜底、登录校验、缓存等实战案例,帮你理解装饰器原理并直接应用到项目中。

之前聊了 @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 的装饰器就是靠闭包层层传的,没有捷径。

类装饰器 —— 用 __call__ 实现

如果装饰器需要维护状态(比如调用计数、缓存数据),用类来实现会更自然。例如:

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。

保留元信息 —— @wraps 不能省

被装饰后的函数,__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 更暴力,直接在运行时拦住。

说说我的看法
全部评论(
没有评论
关于
本网站专注于 Java、数据库(MySQL、Oracle)、Linux、软件架构及大数据等多领域技术知识分享。涵盖丰富的原创与精选技术文章,助力技术传播与交流。无论是技术新手渴望入门,还是资深开发者寻求进阶,这里都能为您提供深度见解与实用经验,让复杂编码变得轻松易懂,携手共赴技术提升新高度。如有侵权,请来信告知:hxstrive@outlook.com
其他应用
公众号