Python装饰器深度解析:从核心原理到高级应用实战

发布时间:2026/8/25 18:03:00
Python装饰器深度解析:从核心原理到高级应用实战 1. 项目概述为什么装饰器是Python开发者的必修课如果你写过一段时间的Python尤其是在接触Web框架比如Flask、Django或者一些测试框架时一定见过那个神奇的“”符号。它看起来像语法糖用起来也简单但很多人对它背后的机制一知半解仅仅停留在“用它来计时”或者“用它来做权限验证”的层面。这就好比你会开车但不知道发动机是怎么工作的一旦车子抛锚你就只能干瞪眼。装饰器Decorator在Python中远不止是一个“语法糖”那么简单。它是Python作为一门动态语言其“一等公民”函数和闭包特性结合后所诞生的一种强大且优雅的设计模式。理解装饰器是你从“Python使用者”迈向“Python开发者”的关键一步。它能让你写出更干净、更模块化、更具表达力的代码。无论是想给函数自动添加日志、缓存计算结果、实现重试机制还是构建复杂的Web路由或权限系统装饰器都是那个藏在幕后的核心工具。这篇文章我会从一个写了十几年Python的老码农的角度带你彻底拆解装饰器。我们不只讲“怎么用”更要深挖“为什么能这么用”把它的实现原理、各种应用场景以及我踩过的那些坑毫无保留地分享给你。目标是让你读完就能透彻理解并能自信地在自己的项目中运用和创造装饰器。2. 装饰器的核心原理剥开语法糖的外衣要真正掌握装饰器我们必须先忘掉那个可爱的“”符号。因为它只是一个便捷的语法其本质是函数式编程和闭包概念的结合体。理解了这个本质你就能看透所有装饰器变体的真相。2.1 函数Python中的“一等公民”这是理解装饰器的基石。在Python里函数和整数、字符串、列表一样都是对象。这意味着函数可以被赋值给变量my_func len函数可以作为参数传递给另一个函数map(func, iterable)函数可以作为另一个函数的返回值def outer(): def inner(): ...; return inner函数可以定义在另一个函数内部这就是嵌套函数。正是这些特性使得“操作函数”成为可能。装饰器本质上就是在“操作函数”它接受一个函数作为输入然后返回一个新的、增强了的函数。2.2 闭包让内部函数“记住”外部环境闭包是另一个关键概念。当一个嵌套函数内部函数引用了其外部函数enclosing function的变量包括参数时就形成了一个闭包。即使外部函数已经执行完毕并返回内部函数仍然可以访问那些被引用的变量。def outer(msg): # msg 是外部函数的局部变量 def inner(): # inner 引用了外部变量 msg形成了闭包 print(msg) return inner # 返回内部函数对象 my_func outer(Hello, Closure!) my_func() # 输出Hello, Closure! # 此时 outer 函数早已执行结束但 inner 依然能访问到当时的 msg 值。装饰器就是利用闭包来“记住”要被装饰的那个原始函数以及装饰器本身可能接收的一些参数。2.3 装饰器的本质一个高阶函数现在我们把上面两个概念结合起来。一个最简单的、不带“”语法的装饰器长这样def my_decorator(func): # 1. 接收一个函数作为参数 def wrapper(): # 2. 内部定义一个新函数闭包 print(Something is happening before the function is called.) func() # 3. 调用原始函数 print(Something is happening after the function is called.) return wrapper # 4. 返回这个新函数 def say_hello(): print(Hello!) # 手动装饰过程用装饰器函数处理原始函数得到一个新函数 decorated_say_hello my_decorator(say_hello) # 调用新函数 decorated_say_hello() # 输出 # Something is happening before the function is called. # Hello! # Something is happening after the function is called.这个过程清晰展示了装饰器的本质my_decorator是一个高阶函数以函数为参数或返回函数它“吞掉”一个函数say_hello“吐出”另一个功能增强的函数wrapper。之后我们调用的decorated_say_hello实际上是wrapper函数。2.4 “”语法糖让装饰过程更优雅手动赋值的方式太啰嗦了。于是Python提供了“”语法它只是一种语法上的简写在函数定义时立即执行装饰过程my_decorator # 等价于 say_hello my_decorator(say_hello) def say_hello(): print(Hello!) say_hello() # 效果和上面完全一样这个“”行实际上是在def say_hello(): ...这一定义执行后立刻运行my_decorator(say_hello)并将结果重新赋值给同名变量say_hello。从此say_hello这个名字指向的不再是原来的函数而是装饰器返回的wrapper函数。注意这里有一个极其重要的细节也是初期容易迷惑的地方。装饰器函数如my_decorator的调用和被执行发生在被装饰函数如say_hello的定义阶段而不是在调用阶段当你导入一个模块时里面所有带“”装饰的函数其装饰过程就已经完成了。理解这一点对调试和理解装饰器的副作用至关重要。2.5 元数据丢失问题与functools.wraps按照上面的写法我们引入了一个新问题原始函数的“身份”丢失了。print(say_hello.__name__) # 输出wrapper print(say_hello.__doc__) # 输出None (如果wrapper没写文档字符串)这会给调试、日志和依赖函数元信息的工具如某些Web框架的路由带来麻烦。为了解决这个问题Python标准库提供了functools.wraps装饰器。它的作用就是将原始函数的重要元数据如__name__,__doc__,__module__等复制到包装函数上。正确写法如下import functools def my_decorator(func): functools.wraps(func) # 关键在这里 def wrapper(): print(Before call) func() print(After call) return wrapper my_decorator def say_hello(): 这是一个打招呼的函数 print(Hello!) print(say_hello.__name__) # 输出say_hello print(say_hello.__doc__) # 输出这是一个打招呼的函数实操心得养成习惯在定义装饰器内部的wrapper函数时第一时间加上functools.wraps(func)。这是一个优秀的实践能避免很多意想不到的坑。3. 装饰器的进阶形态与核心细节掌握了基础原理我们来看看装饰器在实际应用中那些更复杂、但也更强大的形态。3.1 装饰带参数的函数被装饰的函数可能有参数我们的wrapper必须能接收并传递这些参数。这用*args和**kwargs就能完美解决。import functools def log_decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): # 接收任意形式的位置参数和关键字参数 print(f[LOG] Calling {func.__name__} with args{args}, kwargs{kwargs}) result func(*args, **kwargs) # 原样传递给原始函数 print(f[LOG] {func.__name__} returned {result}) return result return wrapper log_decorator def add(a, b): return a b log_decorator def greet(name, greetingHello): return f{greeting}, {name}! print(add(1, 2)) # 输出 # [LOG] Calling add with args(1, 2), kwargs{} # [LOG] add returned 3 # 3 print(greet(Alice, greetingHi)) # 输出 # [LOG] Calling greet with args(Alice,), kwargs{greeting: Hi} # [LOG] greet returned Hi, Alice! # Hi, Alice!这种设计使得装饰器可以通用地装饰任何签名的函数适应性极强。3.2 带参数的装饰器本身有时我们希望装饰器本身也能接收参数来实现更灵活的配置。比如一个重试装饰器可以指定重试次数和延迟时间。这需要再嵌套一层函数。import functools import time def retry(max_attempts3, delay1): 一个带参数的重试装饰器工厂 def decorator(func): # 这一层接收被装饰的函数 functools.wraps(func) def wrapper(*args, **kwargs): last_exception None for attempt in range(1, max_attempts 1): try: print(fAttempt {attempt}/{max_attempts} for {func.__name__}) return func(*args, **kwargs) except Exception as e: last_exception e if attempt max_attempts: print(fFailed, retrying in {delay} second(s)...) time.sleep(delay) # 所有尝试都失败 print(fAll {max_attempts} attempts failed for {func.__name__}) raise last_exception return wrapper return decorator # 返回真正的装饰器函数 # 使用带参数的装饰器 retry(max_attempts2, delay0.5) def call_unstable_api(): import random if random.random() 0.7: # 70%概率失败 raise ConnectionError(API call failed) return Success! # 尝试运行观察重试行为 try: result call_unstable_api() print(result) except ConnectionError as e: print(f最终捕获异常: {e})它的执行顺序是retry(max_attempts2, delay0.5)先调用retry(2, 0.5)它返回decorator函数。然后这个decorator函数再以call_unstable_api为参数被调用即decorator(call_unstable_api)最终返回wrapper函数。所以带参数的装饰器实际上是一个“装饰器工厂”它生产出针对特定参数的装饰器。3.3 类装饰器用类来实现装饰器除了用函数用类来实现装饰器也是完全可行的这有时能让状态管理更清晰。类装饰器依赖于类的__call__方法让实例可以像函数一样被调用。import functools class CountCalls: 一个记录函数调用次数的类装饰器 def __init__(self, func): functools.update_wrapper(self, func) # 类似 wraps 的功能 self.func func self.num_calls 0 # 状态保存在实例属性中 def __call__(self, *args, **kwargs): self.num_calls 1 print(fCall {self.num_calls} of {self.func.__name__}) return self.func(*args, **kwargs) CountCalls def say_hi(): print(Hi!) say_hi() # 输出Call 1 of say_hi \n Hi! say_hi() # 输出Call 2 of say_hi \n Hi! print(say_hi.num_calls) # 输出2可以直接访问状态类装饰器的初始化__init__发生在装饰阶段函数定义时而真正的装饰逻辑在__call__方法中该方法在每次调用被装饰函数时执行。用类装饰器来管理装饰器自身的状态比如计数器、缓存字典会非常方便。3.4 装饰器堆叠多个装饰器如何工作你可以把多个装饰器堆叠在一个函数上它们会从下往上或者说从内到外依次应用。def decorator_a(func): def wrapper(): print(Decorator A - before) func() print(Decorator A - after) return wrapper def decorator_b(func): def wrapper(): print(Decorator B - before) func() print(Decorator B - after) return wrapper decorator_a decorator_b def target_function(): print(Target function running) target_function()输出会是Decorator A - before Decorator B - before Target function running Decorator B - after Decorator A - after可以把它理解为target_function decorator_a(decorator_b(target_function))。decorator_b先包装原始函数然后decorator_a再包装decorator_b返回的包装函数。执行时就像剥洋葱一样从最外层decorator_a的wrapper开始执行逐步向内。注意事项装饰器堆叠时顺序非常重要。不同的顺序可能导致完全不同的行为。例如一个缓存装饰器应该放在最里面最先应用而一个记录日志的装饰器可能放在外面这样日志能记录到包括缓存命中在内的所有调用。4. 装饰器的经典应用场景实战解析理解了原理和形态我们来看看装饰器在实际项目中能解决哪些具体问题。这些场景都是我多年开发中高频使用的每一个都配有详细的代码和解释。4.1 性能分析与调试计时与日志这是装饰器最直观的用途之一用于非侵入式地监控函数性能和行为。场景一函数运行计时器import functools import time def timer(func): 打印函数运行时间的装饰器 functools.wraps(func) def wrapper(*args, **kwargs): start_time time.perf_counter() # 使用高精度计时器 result func(*args, **kwargs) end_time time.perf_counter() run_time end_time - start_time print(fFinished {func.__name__!r} in {run_time:.4f} secs) return result return wrapper timer def slow_calculation(n): 模拟一个耗时的计算 total 0 for i in range(n): total i ** 2 time.sleep(0.001) # 模拟耗时 return total print(slow_calculation(100))场景二智能日志装饰器一个更实用的日志装饰器可以配置日志级别并输出到文件。import functools import logging # 配置日志 logging.basicConfig(levellogging.DEBUG, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[logging.FileHandler(app.log), logging.StreamHandler()]) logger logging.getLogger(__name__) def log_it(levellogging.INFO): 带日志级别参数的日志装饰器 def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): logger.log(level, fCalling {func.__name__} with args{args}, kwargs{kwargs}) try: result func(*args, **kwargs) logger.log(level, f{func.__name__} returned {result}) return result except Exception as e: logger.error(f{func.__name__} raised an exception: {e}, exc_infoTrue) raise # 重新抛出异常 return wrapper return decorator log_it(levellogging.DEBUG) def process_data(data_id, threshold0.5): # ... 一些数据处理逻辑 if data_id 0: raise ValueError(Invalid data ID) return fProcessed data {data_id} with threshold {threshold} process_data(42) process_data(-1) # 这会触发错误日志4.2 功能增强缓存、重试与单例装饰器可以无感地为函数添加新的能力而不修改其内部逻辑。场景三使用functools.lru_cache实现记忆化缓存Python标准库自带了一个极其强大的缓存装饰器对于计算昂贵、纯函数性质的函数相同输入必然得到相同输出它能极大提升性能。from functools import lru_cache import time lru_cache(maxsize128) # 最多缓存128个不同的调用结果 def fibonacci(n): 计算斐波那契数递归版本无缓存时极慢 if n 2: return n return fibonacci(n-1) fibonacci(n-2) start time.perf_counter() print(fibonacci(35)) # 第一次计算较慢因为要递归 first_time time.perf_counter() - start print(fFirst call took {first_time:.4f} secs) start time.perf_counter() print(fibonacci(35)) # 第二次直接从缓存读取瞬间返回 cached_time time.perf_counter() - start print(fCached call took {cached_time:.6f} secs) print(fibonacci.cache_info()) # 查看缓存命中情况lru_cache内部使用字典来存储参数和结果的映射maxsize指定缓存大小采用LRU最近最少使用算法进行淘汰。对于API调用、数据库查询在短时间内数据不变的情况下、复杂配置解析等场景自定义一个缓存装饰器也非常有用。场景四自定义缓存装饰器简化版import functools def simple_cache(func): 一个简单的内存缓存装饰器 cache {} functools.wraps(func) def wrapper(*args, **kwargs): # 创建一个基于位置参数和关键字参数的缓存键 # 注意这里简化处理实际应用中键需要可哈希且要考虑kwargs的顺序问题 key (args, tuple(sorted(kwargs.items()))) if kwargs else args if key not in cache: cache[key] func(*args, **kwargs) return cache[key] return wrapper场景五单例模式装饰器确保一个类只被实例化一次。def singleton(cls): 单例装饰器用于类 instances {} functools.wraps(cls) def wrapper(*args, **kwargs): if cls not in instances: instances[cls] cls(*args, **kwargs) return instances[cls] return wrapper singleton class DatabaseConnection: def __init__(self, connection_string): print(fInitializing connection to {connection_string}) self.connection_string connection_string # ... 其他方法 conn1 DatabaseConnection(mysql://localhost:3306/mydb) conn2 DatabaseConnection(mysql://localhost:3306/mydb) # 不会打印初始化信息 print(conn1 is conn2) # 输出True是同一个对象4.3 输入验证与权限控制在Web开发或API设计中装饰器是进行前置条件检查的绝佳位置。场景六参数类型检查装饰器import inspect import functools def validate_types(func): 通过函数注解进行类型检查的装饰器 sig inspect.signature(func) annotations func.__annotations__ functools.wraps(func) def wrapper(*args, **kwargs): # 将传入的参数绑定到函数签名 bound sig.bind(*args, **kwargs) bound.apply_defaults() # 检查每个有类型注解的参数 for param_name, param_value in bound.arguments.items(): if param_name in annotations: expected_type annotations[param_name] if not isinstance(param_value, expected_type): raise TypeError( fArgument {param_name} must be {expected_type.__name__}, fgot {type(param_value).__name__} ) return func(*args, **kwargs) return wrapper validate_types def greet_person(name: str, age: int) - str: return f{name} is {age} years old. print(greet_person(Alice, 30)) # 正常 print(greet_person(Bob, thirty)) # 触发 TypeError场景七简易权限验证装饰器模拟Web场景import functools # 模拟一个当前用户 current_user {username: alice, roles: [user]} def require_role(role): 检查当前用户是否拥有指定角色 def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): if role not in current_user.get(roles, []): raise PermissionError(fUser {current_user[username]} lacks required role: {role}) return func(*args, **kwargs) return wrapper return decorator class ArticleService: require_role(editor) def publish_article(self, article_id): return fArticle {article_id} published successfully. require_role(user) def view_article(self, article_id): return fDisplaying article {article_id}. service ArticleService() print(service.view_article(101)) # 正常 print(service.publish_article(101)) # 触发 PermissionError4.4 注册与路由机制这是许多框架如Flask、FastAPI的核心模式。装饰器在这里用于将函数注册到某个中央调度器。场景八模拟一个简单的事件处理器注册系统import functools class EventDispatcher: def __init__(self): self._handlers {} def register(self, event_type): 注册事件处理器的装饰器工厂 def decorator(func): if event_type not in self._handlers: self._handlers[event_type] [] self._handlers[event_type].append(func) return func # 注意这里返回原函数通常不需要包装 return decorator def dispatch(self, event_type, *args, **kwargs): 触发某个事件的所有处理器 results [] for handler in self._handlers.get(event_type, []): results.append(handler(*args, **kwargs)) return results dispatcher EventDispatcher() dispatcher.register(user.login) def send_login_notification(user_id): return f[Email] Login notification sent to user {user_id} dispatcher.register(user.login) def update_login_stats(user_id): return f[Stats] Login stats updated for user {user_id} dispatcher.register(order.created) def process_order(order_id): return f[Order] Processing order {order_id} # 触发事件 print(dispatcher.dispatch(user.login, 123)) # 输出[[Email] Login notification sent to user 123, [Stats] Login stats updated for user 123]这种模式非常强大它实现了“关注点分离”。事件处理函数只需要关心自己的业务逻辑而事件的绑定关系通过装饰器声明式地完成由中央调度器统一管理。5. 装饰器在实战中的高级技巧与避坑指南掌握了基本应用后我们来看看一些更深入的话题和实际开发中容易踩的坑。5.1 装饰器对函数签名和自省的影响及应对装饰器会改变函数的元信息除了之前提到的__name__可以用functools.wraps解决还有一个更隐蔽的问题函数签名。像inspect.signature这样的工具默认看到的是包装函数wrapper的签名而不是原始函数的。import inspect def simple_decorator(func): def wrapper(*args, **kwargs): return func(*args, **kwargs) return wrapper simple_decorator def original(a: int, b: str hello) - float: Original function. return 3.14 print(inspect.signature(original)) # 输出(*args, **kwargs)我们看到签名变成了wrapper的(*args, **kwargs)丢失了参数名、类型注解和默认值信息。这对于依赖签名进行文档生成、参数校验或IDE提示的工具来说是灾难性的。解决方案functools.wraps已经部分解决了这个问题但对于签名我们需要使用functools.update_wrapper的升级版或者第三方库decorator通过pip install decorator安装它能更好地保存签名。使用标准库的增强方案import functools def better_decorator(func): functools.wraps(func) # 这一步会复制 __name__, __doc__ 等 def wrapper(*args, **kwargs): return func(*args, **kwargs) # 手动将原始函数的签名赋给包装函数 wrapper.__signature__ inspect.signature(func) return wrapper better_decorator def original(a: int, b: str hello) - float: Original function. return 3.14 print(inspect.signature(original)) # 输出(a: int, b: str hello) - float5.2 装饰器与静态方法、类方法的顺序问题在类里面使用装饰器时如果和staticmethod、classmethod混用顺序至关重要。类装饰器必须应用在最外层。class MyClass: # 错误顺序timer 在内层它接收到的 func 可能不是普通函数 # timer # staticmethod # def bad_static(): # pass # 正确顺序先声明静态方法或类方法再用自定义装饰器装饰 staticmethod timer # timer 装饰器现在接收的是一个普通的静态函数 def good_static(): time.sleep(0.1) print(Static method called) classmethod timer def good_class(cls): time.sleep(0.1) print(fClass method called on {cls}) MyClass.good_static() MyClass.good_class()原理是staticmethod和classmethod是描述符descriptor它们会返回一个特殊的对象。如果你的自定义装饰器在内层它接收到的就是这个描述符对象而不是一个可调用的函数这很可能导致错误。所以记住这个口诀“先特殊staticmethod/classmethod后普通自定义装饰器”。5.3 调试被装饰的函数当你的函数被装饰后如果抛出异常堆栈跟踪traceback会显示装饰器内部的wrapper函数这有时会让调试变得困难尤其是当装饰器层数很多的时候。def buggy_decorator(func): def wrapper(): # 装饰器本身可能引入bug x 1 / 0 # 这里会抛出 ZeroDivisionError return func() return wrapper buggy_decorator def innocent_func(): return OK innocent_func()错误信息会指向wrapper内部的x 1 / 0这一行而不是直接告诉你innocent_func有问题。虽然functools.wraps能帮助显示正确的函数名但行号可能还是装饰器的。调试技巧临时移除装饰器在调试时可以暂时注释掉decorator这一行直接测试原始函数。使用__wrapped__属性functools.wraps装饰器或手动设置wrapper.__wrapped__ func会给包装函数添加一个__wrapped__属性指向原始函数。在调试器中你可以查看这个属性。在装饰器内部做好异常处理和日志就像前面log_it装饰器做的那样在wrapper里用try...except捕获异常并记录详细上下文信息如参数值然后再重新抛出这能极大提升线上问题排查效率。5.4 装饰器的性能考量装饰器会在函数定义时执行一次装饰过程并在每次调用函数时增加一层或多层函数调用开销。对于绝大多数应用这个开销微乎其微。但是如果装饰器内部逻辑非常复杂或者被装饰的函数在一个超密集的循环中被调用数百万次那么这部分开销就需要考虑了。优化建议保持装饰器逻辑轻量装饰器应专注于“横切关注点”如日志、鉴权复杂的业务逻辑不要放在装饰器里。避免在装饰器内部进行昂贵的初始化特别是带参数的装饰器其工厂函数和装饰函数可能在导入时就被执行。昂贵的操作如打开网络连接、读取大文件应延迟到wrapper函数内部或者确保只执行一次并缓存结果。对于性能至上的场景可以考虑在非生产环境开发、测试使用装饰器注入调试代码而在生产环境通过条件导入或配置开关来跳过这些装饰器。有些框架也提供了编译时或启动时的装饰器处理机制来消除运行时开销。6. 从理解到创造设计你自己的装饰器当你透彻理解了装饰器的原理和应用场景后就可以开始设计符合自己业务需求的装饰器了。这里我分享一个设计流程和几个更复杂的例子。6.1 设计流程与思考要点明确目的这个装饰器要解决什么问题缓存、重试、锁、验证、监控、注册……确定输入装饰器本身需要参数吗如重试次数、缓存过期时间。被装饰的函数有什么样的签名用*args, **kwargs来保持通用性。设计包装逻辑在调用原函数之前、之后、或者围绕原函数的异常处理中需要做什么处理返回值与异常装饰器应该返回什么通常就是原函数的返回值。异常应该被捕获并处理如重试还是直接抛出保护元信息务必使用functools.wraps。考虑线程安全如果装饰器内部有共享状态如缓存字典、计数器在多线程环境下是否需要加锁编写文档清晰说明装饰器的用途、参数和行为。6.2 复杂示例一带过期时间的缓存装饰器import functools import time from threading import RLock def timed_cache(seconds): 带绝对过期时间的缓存装饰器 def decorator(func): cache {} cache_lock RLock() # 因为可能多线程访问缓存字典 functools.wraps(func) def wrapper(*args, **kwargs): # 生成缓存键生产环境需要更健壮的键生成函数 key (args, tuple(sorted(kwargs.items())) if kwargs else ()) now time.time() with cache_lock: # 检查缓存是否存在且未过期 if key in cache: result, timestamp cache[key] if now - timestamp seconds: print(fCache HIT for {func.__name__}{args}) return result else: print(fCache EXPIRED for {func.__name__}{args}) del cache[key] # 缓存未命中或已过期调用原函数 print(fCache MISS for {func.__name__}{args}) result func(*args, **kwargs) cache[key] (result, now) return result # 可选提供清理过期缓存的方法 def clear_cache(): with cache_lock: now time.time() expired_keys [k for k, (_, ts) in cache.items() if now - ts seconds] for k in expired_keys: del cache[k] print(fCleared {len(expired_keys)} expired entries.) wrapper.clear_cache clear_cache return wrapper return decorator timed_cache(seconds5) # 缓存5秒 def get_expensive_data(query): print(fComputing expensive data for {query}...) time.sleep(1) # 模拟耗时计算 return fresult_for_{query} print(get_expensive_data(A)) # 计算缓存 print(get_expensive_data(A)) # 命中缓存 time.sleep(6) print(get_expensive_data(A)) # 缓存过期重新计算6.3 复杂示例二数据库事务装饰器模拟在Web应用中我们经常需要确保一个函数内的所有数据库操作在一个事务中完成成功则提交失败则回滚。import functools class DatabaseSession: 模拟数据库会话 def __init__(self): self.in_transaction False def begin(self): print(BEGIN TRANSACTION) self.in_transaction True def commit(self): if self.in_transaction: print(COMMIT) self.in_transaction False def rollback(self): if self.in_transaction: print(ROLLBACK) self.in_transaction False def with_transaction(session_provider): 事务装饰器。 :param session_provider: 一个可调用对象返回一个数据库会话对象该对象需要有 commit 和 rollback 方法。 def decorator(func): functools.wraps(func) def wrapper(*args, **kwargs): session session_provider() session.begin() try: result func(*args, **kwargs, sessionsession) # 将会话作为参数传入 session.commit() return result except Exception as e: session.rollback() print(fTransaction rolled back due to: {e}) raise # 将异常继续向上抛 return wrapper return decorator # 模拟一个获取会话的函数 def get_db_session(): return DatabaseSession() with_transaction(get_db_session) def create_user_and_order(username, session): # 注意函数需要接收 session 参数 print(fCreating user {username}...) # 模拟数据库操作1 # session.execute(...) print(fCreating order for {username}...) # 模拟数据库操作2 # session.execute(...) # 如果这里抛出异常整个事务会回滚 # raise ValueError(Something went wrong!) return fUser {username} and order created successfully. print(create_user_and_order(Alice))这个装饰器展示了如何将资源数据库会话的管理逻辑从业务函数中剥离出来使业务代码更专注于核心逻辑实现了关注点分离和代码复用。装饰器是Python语言赋予开发者的一把利器它体现了Python“优雅”、“明确”、“简单”的设计哲学。从最初的理解函数作为对象到闭包的运用再到设计出各种解决实际问题的装饰器这个过程本身就是对Python编程思想的一次深度修炼。我个人的体会是不要为了用装饰器而用装饰器而是在你发现多个函数有相同的横切逻辑如日志、缓存、验证并且这些逻辑与核心业务无关时它就是最佳的解决方案。多看看优秀开源项目如Flask、Django REST framework、Celery中装饰器的用法能给你带来更多灵感。最后记住那个黄金法则当你写装饰器时第一个动作就应该是functools.wraps(func)它能为你省去未来许多调试的烦恼。