Python装饰器为修改和扩展函数的行为提供了简单而强大的语法。装饰器（decorator）本质上是函数，它接受函数作为参数、增强其功能，并返回新的函数——而不需要永久修改原始函数本身。
前面两篇介绍了装饰器原理及典型应用示例，本文继续介绍6个实用的装饰器，包括缓存、过期缓存、速率限制、失败重试、内存检测以及类型检查功能。无论您是想分析性能、提高效率、验证数据还是管理错误，这些装饰器都可以满足您的需求！选择的示例侧重于典型应用模式和实用性，它们可以在您的日常编程中派上用场，并为您节省大量工作。理解装饰器的灵活性将帮助您编写干净、有弹性和优化的应用程序代码。
失败重试装饰器
当你想要在失败后自动重新尝试继续执行函数时，这个装饰器会派上用场，增强它在涉及暂时失败的情况下的弹性，常用于容易出现间歇性故障的外部服务或网络请求。
import sqlite3
import time
def retry_on_failure(max_attempts, retry_delay=1):
def decorator(func):
def wrapper(*args, **kwargs):
for _ in range(max_attempts):
try:
result = func(*args, **kwargs)
return result
except Exception as error:
print(f"Error occurred: {error}. Retrying...")
time.sleep(retry_delay)
raise Exception("Maximum attempts exceeded. Function failed.")
return wrapper
return decorator
@retry_on_failure(max_attempts=3, retry_delay=2)
def establish_database_connection():
connection = sqlite3.connect("example.db")
db_cursor = connection.cursor()
db_cursor.execute("SELECT * FROM users")
query_result = db_cursor.fetchall()
db_cursor.close()
connection.close()
return query_result
try:
retrieved_data = establish_database_connection()
print("Data retrieved successfully:", retrieved_data)
except Exception as error_message:
print(f"Failed to establish database connection: {error_message}")
执行结果：
Error occurred: no such table: users. Retrying...
Error occurred: no such table: users. Retrying...
Error occurred: no such table: users. Retrying...
Failed to establish database connection: Maximum attempts exceeded. Function failed.
该示例装饰器用于在发生失败时重试函数执行。它具有指定的最大尝试次数和重试之间的延迟。
retry_on_failure()是装饰器工厂，接受最大重试计数和延迟的参数，并返回一个管理重试逻辑的decorator()。在wrapper()函数中，经过装饰的函数在循环中执行，尝试指定的最大次数。如果出现异常，它打印错误消息，引入由retry_delay指定的延迟，然后重试。如果所有尝试都失败，则会引发异常，表明已超过最大尝试数。
在establish_database_connection()函数上面应用@retry_on_failure()，集成重试逻辑，在数据库连接遇到失败的情况下，允许最多3次重试，每次尝试之间有2秒的延迟。在不改变核心函数代码的前提下，成功实现了无缝融合重试的实用功能。
- 应用场景
这个重试装饰器在应用程序开发中非常有用，可以增加针对临时或间歇性错误的弹性。
举例，考虑航班预订应用程序，它调用支付网关API process_payment()来处理客户事务。有时，支付提供商端的网络中断或高负载可能会导致API响应中的短暂错误。process_payment()函数可以使用@retry_on_failure修饰，而不是直接向客户显示失败，以隐式地处理此类场景。现在，当支付失败一次时，它将无缝地重新发送请求多达3次，直到最终报告错误，如果错误仍然存在。
这提供了屏蔽暂时中断的功能，而不会直接将用户暴露在不可靠的基础设施行为中。即使依赖的服务偶尔出现故障，应用程序也保持可靠可用。重试装饰器有助于整齐地限制重试逻辑，而不会将其分散到API代码中。超出应用程序控制范围的故障被优雅地处理，而不是由应用程序故障直接影响用户。
限流装饰器
通过控制调用函数的频率，限流装饰器确保有效的资源管理并防止误用。它在API滥用或资源保护等情况下特别有用，因为限制函数调用是必要的。
import time
def rate_limiter(max_allowed_calls, reset_period_seconds):
def decorate_rate_limited_function(original_function):
calls_count = 0
last_reset_time = time.time()
def wrapper_function(*args, **kwargs):
nonlocal calls_count, last_reset_time
elapsed_time = time.time() - last_reset_time
# If the elapsed time is greater than the reset period, reset the call count
if elapsed_time > reset_period_seconds:
calls_count = 0
last_reset_time = time.time()
# Check if the call count has reached the maximum allowed limit
if calls_count >= max_allowed_calls:
raise Exception("Rate limit exceeded. Please try again later.")
# Increment the call count
calls_count += 1
# Call the original function
return original_function(*args, **kwargs)
return wrapper_function
return decorate_rate_limited_function
# Allowing a maximum of 6 API calls within 10 seconds.
@rate_limiter(max_allowed_calls=6, reset_period_seconds=10)
def make_api_call():
print("API call executed successfully...")
# Make API calls
for _ in range(8):
try:
make_api_call()
except Exception as error:
print(f"Error occurred: {error}")
time.sleep(10)
make_api_call()
输出结果：
API call executed successfully...
API call executed successfully...
API call executed successfully...
API call executed successfully...
API call executed successfully...
API call executed successfully...
Error occurred: Rate limit exceeded. Please try again later.
Error occurred: Rate limit exceeded. Please try again later.
API call executed successfully...
这段代码展示了使用装饰器实现函数调用的速率限制机制。
rate_limititer()函数是速率限制逻辑的核心，它指定了最大调用数和以秒为单位的重置计数周期。decorate_rate_limited_function()使用包装器来管理速率限制，方法是在周期已过时重置计数。它检查计数是否已达到允许的最大值，然后引发异常或增加计数并相应地执行函数。
使用@rate_limititer()应用于make_api_call()，它将函数限制为在任何10秒内调用6次。这在不改变功能逻辑的情况下引入了速率限制，确保调用遵守限制并防止在设置的间隔内过度使用。
- 应用场景
速率限制装饰器在应用程序开发中对于控制api的使用和防止滥用非常有用。
举例，旅行预订应用程序可能依赖于第三方Flight Search API来检查各航空公司的实时座位可用性。虽然大多数使用都是合法的，但有些用户可能会过度调用此API，从而降低整体服务性能。
通过像@rate_limititer(100, 60)这样修饰API模块，将实现限制预订模块每分钟只进行100次Flight API调用。超过调用限制直接通过装饰器被拒绝，甚至没有到达实际的API。这可以避免下游服务过度使用，从而保护服务、基础设施。
类型检查包装器
强制类型检查装饰器通过验证函数参数是否符合指定的数据类型、防止与类型相关的错误和提高代码可靠性来确保数据完整性。在严格遵守数据类型至关重要的情况下，它特别有用。
import inspect
def enforce_type_checking(func):
def type_checked_wrapper(*args, **kwargs):
# Get the function signature and parameter names
function_signature = inspect.signature(func)
function_parameters = function_signature.parameters
# Iterate over the positional arguments
for i, arg_value in enumerate(args):
parameter_name = list(function_parameters.keys())[i]
parameter_type = function_parameters[parameter_name].annotation
if not isinstance(arg_value, parameter_type):
raise TypeError(f"Argument '{parameter_name}' must be of type '{parameter_type.__name__}'")
# Iterate over the keyword arguments
for keyword_name, arg_value in kwargs.items():
parameter_type = function_parameters[keyword_name].annotation
if not isinstance(arg_value, parameter_type):
raise TypeError(f"Argument '{keyword_name}' must be of type '{parameter_type.__name__}'")
# Call the original function
return func(*args, **kwargs)
return type_checked_wrapper
# Example usage
@enforce_type_checking
def multiply_numbers(factor_1: int, factor_2: int) -> int:
return factor_1 * factor_2
# Call the decorated function
result = multiply_numbers(5, 7) # No type errors, returns 35
print("Result:", result)
result = multiply_numbers("5", 7) # Type error: 'factor_1' must be of type 'int'
输出结果：
Result:Traceback (most recent call last):
File "C:\\\\Program Files\\\\Sublime Text 3\\\\test.py", line 36, in <module>
35
result = multiply_numbers("5", 7) # Type error: 'factor_1' must be of type 'int'
File "C:\\\\Program Files\\\\Sublime Text 3\\\\test.py", line 14, in type_checked_wrapper
raise TypeError(f"Argument '{parameter_name}' must be of type '{parameter_type.__name__}'")
TypeError: Argument 'factor_1' must be of type 'int'
execut_type_checking装饰器验证传递给函数的参数是否与指定的类型注释匹配。
在type_checked_wrapper中，它检查修饰函数的签名，检索参数名和类型注释，并确保提供的参数与预期的类型一致。这包括检查位置参数的顺序，以及检查关键字参数的参数名。如果检测到类型不匹配，则会引发TypeError。
这个修饰符的例子是它在multiply_numbers函数中的应用，其中的参数被注释为整数。尝试传递字符串会导致异常，而传递整数则会正常执行。
- 应用场景
使用类型检查装饰器可以及早发现问题并提高可靠性。
举例，考虑web应用程序后端，它的数据访问层函数get_user_data()被注释为期望整数用户id。如果字符串id从前端代码传入则查询失败。
使用此装饰器，则无需在本地添加显式检查并引发异常。现在，任何传递无效类型的上游或消费者代码都将在函数执行期间被自动捕获。装饰器检查注解和参数类型，并在到达数据库层之前抛出相应的错误。
这种方式确保有效的数据格式才能跨层流动，从而防止类型模糊错误。在没有额外检查的情况下，实现类型安全检查，使逻辑更清晰。
内存检测装饰器
当涉及到大型数据密集型应用程序或资源受限环境时，内存检测装饰器可以实现内存检测，它提供对函数内存消耗的洞察。
import tracemalloc
def measure_memory_usage(target_function):
def wrapper(*args, **kwargs):
tracemalloc.start()
# Call the original function
result = target_function(*args, **kwargs)
snapshot = tracemalloc.take_snapshot()
top_stats = snapshot.statistics("lineno")
# Print the top memory-consuming lines
print(f"Memory usage of {target_function.__name__}:")
for stat in top_stats[:5]:
print(stat)
# Return the result
return result
return wrapper
# Example usage
@measure_memory_usage
def calculate_factorial_recursive(number):
if number == 0:
return 1
else:
return number * calculate_factorial_recursive(number - 1)
# Call the decorated function
result_factorial = calculate_factorial_recursive(3)
print("Factorial:", result_factorial)
输出结果：
Memory usage of calculate_factorial_recursive:
C:\\\\Program Files\\\\Sublime Text 3\\\\test.py:29: size=1552 B, count=6, average=259 B
C:\\\\Program Files\\\\Sublime Text 3\\\\test.py:8: size=896 B, count=3, average=299 B
C:\\\\Program Files\\\\Sublime Text 3\\\\test.py:10: size=416 B, count=1, average=416 B
Memory usage of calculate_factorial_recursive:
C:\\\\Program Files\\\\Sublime Text 3\\\\test.py:29: size=1552 B, count=6, average=259 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:226: size=880 B, count=3, average=293 B
C:\\\\Program Files\\\\Sublime Text 3\\\\test.py:8: size=832 B, count=2, average=416 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:173: size=800 B, count=2, average=400 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:505: size=592 B, count=2, average=296 B
Memory usage of calculate_factorial_recursive:
C:\\\\Program Files\\\\Sublime Text 3\\\\test.py:29: size=1440 B, count=4, average=360 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:535: size=1240 B, count=3, average=413 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:67: size=1216 B, count=19, average=64 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:193: size=1104 B, count=23, average=48 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:226: size=880 B, count=3, average=293 B
Memory usage of calculate_factorial_recursive:
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:558: size=1416 B, count=29, average=49 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:67: size=1408 B, count=22, average=64 B
C:\\\\Program Files\\\\Sublime Text 3\\\\test.py:29: size=1392 B, count=3, average=464 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:535: size=1240 B, count=3, average=413 B
C:\\\\Program Files\\\\Python310\\\\lib\\\\tracemalloc.py:226: size=832 B, count=2, average=416 B
Factorial: 6
这段代码展示了如何实现装饰器measure_memory_usage，用于度量函数的内存消耗。当应用该装饰器时，会在调用原始函数之前启动内存跟踪。一旦函数完成执行，就会获取内存快照，并打印占用内存最多的前5行。
通过calculate_factorial_recursive()的示例说明，装饰器允许在不更改函数本身的情况下监视内存使用情况，为优化目的提供了有价值的见解。本质上，它提供了一种快捷的方法来评估和分析任何函数在运行时的内存消耗。
- 应用场景
内存检测装饰器在应用程序开发中对于识别和排除内存膨胀或泄漏问题非常有价值。
举例，考虑具有关键ETL组件（如transform_data()）的数据流管道场景，该组件处理大量信息。虽然这个过程在正常负载下看起来很好，但像黑色星期五销售这样的大容量数据可能会导致过度的内存使用或系统崩溃。
使用装饰处理器@measure_memory_usage可以揭示有用的见解，无需工程师手动调试。它将在峰值数据流期间打印内存最密集的行，而无需任何代码更改。你应该瞄准快速消耗内存的特定代码块，并通过更好的算法或优化加以解决。
内存检测装饰器有助于在关键路径上诊断、透视，以便及早发现异常的内存消耗趋势。可以在发布前通过分析预先识别问题，而不是延迟的产生问题。使用该装饰器减少了调试的麻烦，并将运行时故障降至最低。
缓存装饰器
该装饰器通过存储和检索函数结果、消除重复输入的冗余计算，从而提升应用程序响应时间（特别是对于耗时的计算）来优化性能。
def cached_result_decorator(func):
result_cache = {}
def wrapper(*args, **kwargs):
cache_key = (*args, *kwargs.items())
if cache_key in result_cache:
return f"[FROM CACHE] {result_cache[cache_key]}"
result = func(*args, **kwargs)
result_cache[cache_key] = result
return result
return wrapper
# Example usage
@cached_result_decorator
def multiply_numbers(a, b):
return f"Product = {a * b}"
# Call the decorated function multiple times
print(multiply_numbers(4, 5)) # Calculation is performed
print(multiply_numbers(4, 5)) # Result is retrieved from cache
print(multiply_numbers(5, 7)) # Calculation is performed
print(multiply_numbers(5, 7)) # Result is retrieved from cache
print(multiply_numbers(-3, 7)) # Calculation is performed
print(multiply_numbers(-3, 7)) # Result is retrieved from cache
输出结果：
Product = 20
[FROM CACHE] Product = 20
Product = 35
[FROM CACHE] Product = 35
Product = -21
[FROM CACHE] Product = -21
上面代码示例展示了缓存装饰器实现，用来高效地缓存和重用函数调用结果。
cached_result_decorator()函数接受另一个函数并返回包装器。在这个包装器中，缓存字典（result_cache）存储唯一的调用参数及其相应的结果。
在执行实际的函数之前，wrapper()检查当前参数的结果是否已经在缓存中。如果是，它检索并返回缓存的结果； 否则，它调用函数，将结果存储在缓存中，并返回结果。
@cached_result_decorator语法将此缓存逻辑应用于任何函数，例如multiply_numbers()。这样可以确保在使用相同参数的后续调用时重用缓存的结果，从而防止冗余计算。
- 应用场景
缓存缓存装饰器在应用程序开发中非常有用，可以优化重复函数调用的性能。
举例，考虑推荐引擎调用预测模型函数给用户生成建议。Get_user_recommendations()为每个用户请求准备输入数据并将其提供给模型。可以用@cached_result_decorator来装饰它，以引入缓存层，而不是重新运行计算重复输入。
现在，第一次传递唯一用户参数时，模型运行并缓存结果。具有相同输入的后续调用直接返回缓存的模型输出，跳过模型重新计算。通过避免重复的模型推断计算，可以极大地改善响应用户请求的延迟。
可以监控缓存命中率，以证明降低模型服务器基础设施成本是合理的。通过缓存装饰器解耦这些优化关注点，而不是将它们混合在函数逻辑中，可以提高模块化、可读性，并实现快速的性能提升。缓存将在不影响业务功能的情况下单独启用或禁用。
过期缓存装饰器
专门为过期数据设计的装饰器，在一定时间内缓存函数结果；它结合了缓存和基于时间的过期特性，以确保缓存的数据定期刷新，以防止过期并保持相关性。
import time
def cached_function_with_expiry(expiry_time):
def decorator(original_function):
cache = {}
def wrapper(*args, **kwargs):
key = (*args, *kwargs.items())
if key in cache:
cached_value, cached_timestamp = cache[key]
if time.time() - cached_timestamp < expiry_time:
return f"[CACHED] - {cached_value}"
result = original_function(*args, **kwargs)
cache[key] = (result, time.time())
return result
return wrapper
return decorator
# Example usage
@cached_function_with_expiry(expiry_time=5) # Cache expiry time set to 5 seconds
def calculate_product(x, y):
return f"PRODUCT - {x * y}"
# Call the decorated function multiple times
print(calculate_product(23, 5)) # Calculation is performed
print(calculate_product(23, 5)) # Result is retrieved from cache
time.sleep(5)
print(calculate_product(23, 5)) # Calculation is performed (cache expired)
输出结果：
PRODUCT - 115
[CACHED] - PRODUCT - 115
PRODUCT - 115
这段代码展示了如何实现具有自动过期的缓存装饰器。
函数cached_function_with_expiry()返回装饰器，当应用该装饰器时，它利用名为cache的字典来存储函数结果及其相应的时间戳。wrapper()函数检查当前参数的结果是否在缓存中。如果存在并且在过期时间内，则返回缓存的结果；否则调用该函数。
calculate_product()函数使用该装饰器，最初计算并缓存结果，随后的调用检索缓存的结果，直到时间到期才通过重新计算刷新缓存。从本质上讲，该实现防止了冗余计算，同时在指定的过期时间之后自动刷新结果。
- 应用场景
自动缓存过期装饰器在应用程序开发中非常有用，可以提升数据获取模块的性能。
举例，在旅游网站项目中，它调用后端API get_flight_prices()向用户显示实时价格。虽然缓存减少了对昂贵的航班数据源的调用，但静态缓存导致显示过时的价格。
相反，您可以使用@cached_function_with_expiry(60)每分钟自动刷新一次。现在，第一个用户调用获取实时价格并缓存它们，而60s窗口中的后续请求有效地重用缓存的价格。但是缓存在过期后会自动失效，以保证数据的新鲜。
这即能优化业务流程，也不必担心与过时数据的极端情况。该装饰器完美地处理这种情况，通过可配置的刷新时间使缓存与上游的更改保持同步。没有重新计算的冗余，仍然可以向最终用户提供最佳的更新数据。通用的自动过期缓存装饰器，使用自定义的过期规则能跨代码实现功能重用。
总结
Python装饰器在应用程序开发中被广泛使用，它能够干净地插入常见的横切关注点。实现认证、监控和限流等功能，这些功能已在Django和Flask等框架中成为标准装饰器。通过从头学习这些装饰器，不仅让我们快速掌握其原理，很好应用框架提供的标准装饰器，也能让我们自定义特定业务功能的装饰器。期待你也分享关于装饰器的应用场景。