functools高级用法
functools高级用法
functools模块提供了用于操作函数和可调用对象的高级工具。通过学习partial、lru_cache、reduce等函数，可以编写更简洁、高效的函数式代码。
partial函数应用
partial函数用于创建偏函数，固定函数的部分参数，生成新的可调用对象。这在需要重复使用相同参数的场景中非常有用。
from functools import partial
import math
# 基本partial使用
def power(base, exponent):
return base ** exponent
# 创建偏函数
square = partial(power, exponent=2)
cube = partial(power, exponent=3)
print("partial函数示例：")
print(f"平方: {square(5)}") # 5^2 = 25
print(f"立方: {cube(5))") # 5^3 = 125
# 偏函数与内置函数
print(f"\n偏函数与内置函数：")
log2 = partial(math.log, base=2)
log10 = partial(math.log, base=10)
print(f"log2(8) = {log2(8)}")
print(f"log10(1000) = {log10(1000)}")
# partial在数据处理中的应用
def process_data(data, multiplier, offset):
return [x * multiplier + offset for x in data]
# 创建处理函数
double_plus_ten = partial(process_data, multiplier=2, offset=10)
halve_minus_five = partial(process_data, multiplier=0.5, offset=-5)
data = [1, 2, 3, 4, 5]
print(f"\n数据处理示例：")
print(f"原始数据: {data}")
print(f"双倍加十: {double_plus_ten(data)}")
print(f"减半减五: {halve_minus_five(data)}")
lru_cache缓存装饰器
lru_cache是Python 3.2+引入的缓存装饰器，基于最近最少使用（LRU）算法缓存函数结果，显著提升重复计算性能。
from functools import lru_cache
import time
# 基本lru_cache使用
@lru_cache(maxsize=128)
def fibonacci(n):
"""计算斐波那契数列"""
if n < 2:
return n
return fibonacci(n-1) + fibonacci(n-2)
# 测试缓存性能
print("lru_cache性能测试：")
start = time.perf_counter()
result = fibonacci(30)
end = time.perf_counter()
print(f"fibonacci(30) = {result}")
print(f"执行时间: {(end-start)*1000:.2f}ms")
# 查看缓存信息
print(f"缓存信息: {fibonacci.cache_info()}")
# 清除缓存
fibonacci.cache_clear()
print(f"清除后缓存: {fibonacci.cache_info()}")
# 带参数的缓存
@lru_cache(maxsize=256)
def expensive_calculation(x, y, operation='add'):
"""模拟耗时计算"""
time.sleep(0.01) # 模拟计算时间
if operation == 'add':
return x + y
elif operation == 'multiply':
return x * y
elif operation == 'power':
return x ** y
# 测试参数缓存
print(f"\n参数缓存测试：")
start = time.perf_counter()
result1 = expensive_calculation(10, 20, 'add')
time1 = time.perf_counter() - start
start = time.perf_counter()
result2 = expensive_calculation(10, 20, 'add') # 使用缓存
time2 = time.perf_counter() - start
print(f"首次调用: {result1}, 耗时: {time1*1000:.2f}ms")
print(f"缓存调用: {result2}, 耗时: {time2*1000:.2f}ms")
print(f"性能提升: {time1/time2:.1f}x")
reduce累积计算
reduce函数对序列中的元素进行累积计算，将二元函数应用于序列，从左到右reduce为单个值。
from functools import reduce
import operator
# 基本reduce使用
numbers = [1, 2, 3, 4, 5]
# 计算乘积
product = reduce(lambda x, y: x * y, numbers)
print(f"reduce乘积: {product}") # 120
# 使用operator模块
sum_result = reduce(operator.add, numbers)
product_result = reduce(operator.mul, numbers)
print(f"reduce求和: {sum_result}")
print(f"reduce乘积: {product_result}")
# 带初始值的reduce
concatenate = reduce(lambda x, y: f"{x}-{y}", numbers, "start")
print(f"reduce连接: {concatenate}")
# 实际应用：字典合并
dict1 = {'a': 1, 'b': 2}
dict2 = {'b': 3, 'c': 4}
dict3 = {'c': 5, 'd': 6}
# 合并多个字典
merged = reduce(lambda d1, d2: {**d1, **d2}, [dict1, dict2, dict3])
print(f"字典合并: {merged}")
# 实际应用：树形结构处理
def flatten(nested_list):
"""展平嵌套列表"""
return reduce(lambda x, y: x + y,
[item if isinstance(item, list) else [item]
for item in nested_list])
nested = [[1, 2], [3, [4, 5]], [6, 7, [8]]]
flattened = flatten(nested)
print(f"展平列表: {flattened}")
装饰器工具
functools提供了多个工具函数用于创建和操作装饰器，包括wraps、total_ordering和cached_property。
from functools import wraps, total_ordering, cached_property
import time
# wraps装饰器保留原函数信息
def timer_decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.perf_counter()
result = func(*args, **kwargs)
end = time.perf_counter()
print(f"{func.__name__}执行时间: {(end-start)*1000:.2f}ms")
return result
return wrapper
@timer_decorator
def slow_function():
"""模拟慢速函数"""
time.sleep(0.1)
return "完成"
print("wraps装饰器示例：")
result = slow_function()
print(f"函数名称: {slow_function.__name__}")
print(f"函数文档: {slow_function.__doc__}")
# total_ordering自动比较方法
@total_ordering
class Version:
def __init__(self, major, minor, patch):
self.major = major
self.minor = minor
self.patch = patch
def __eq__(self, other):
return (self.major, self.minor, self.patch) == \
(other.major, other.minor, other.patch)
def __lt__(self, other):
return (self.major, self.minor, self.patch) < \
(other.major, other.minor, other.patch)
print(f"\ntotal_ordering示例：")
v1 = Version(1, 2, 3)
v2 = Version(1, 3, 0)
v3 = Version(1, 2, 3)
print(f"v1 < v2: {v1 < v2}")
print(f"v1 == v3: {v1 == v3}")
print(f"v1 >= v3: {v1 >= v3}")
print(f"v2 > v1: {v2 > v1}")
# cached_property缓存属性值
class DataAnalyzer:
def __init__(self, data):
self.data = data
@cached_property
def statistics(self):
"""计算并缓存统计信息"""
print("计算统计信息...")
return {
'mean': sum(self.data) / len(self.data),
'max': max(self.data),
'min': min(self.data),
'count': len(self.data)
}
print(f"\ncached_property示例：")
analyzer = DataAnalyzer([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
# 首次访问计算
print(f"统计信息: {analyzer.statistics}")
# 再次访问使用缓存
print(f"再次访问: {analyzer.statistics}")
实际应用：函数式编程模式
functools支持多种函数式编程模式，包括函数组合、柯里化和高阶函数。
from functools import partial, reduce, wraps
from typing import Callable, Any
# 函数组合
def compose(*functions):
"""组合多个函数"""
return reduce(lambda f, g: lambda *args, **kwargs: f(g(*args, **kwargs)), functions)
# 示例函数
def double(x):
return x * 2
def increment(x):
return x + 1
def square(x):
return x ** 2
# 组合函数
double_then_square = compose(square, double)
increment_then_double = compose(double, increment)
print("函数组合示例：")
print(f"double_then_square(3) = {double_then_square(3)}") # (3*2)^2 = 36
print(f"increment_then_double(3) = {increment_then_double(3)}") # (3+1)*2 = 8
# 柯里化
def curry(func):
"""将多参数函数转换为柯里化形式"""
import inspect
params = inspect.signature(func).parameters
def curried(*args):
if len(args) >= len(params):
return func(*args)
else:
return partial(curried, *args)
return curried
@curry
def add_three_numbers(a, b, c):
return a + b + c
print(f"\n柯里化示例：")
add_five = add_three_numbers(5)
add_five_and_ten = add_five(10)
result = add_five_and_ten(15)
print(f"add_three_numbers(5)(10)(15) = {result}")
# 高阶函数与partial
def apply_operation(data, operation, *args, **kwargs):
"""对数据应用操作"""
return [operation(item, *args, **kwargs) for item in data]
# 创建操作函数
to_upper = partial(str.upper)
to_lower = partial(str.lower)
to_length = partial(len)
data = ["hello", "world", "python", "functools"]
print(f"\n高阶函数示例：")
print(f"转大写: {apply_operation(data, to_upper)}")
print(f"转小写: {apply_operation(data, to_lower)}")
print(f"计算长度: {apply_operation(data, to_length)}")
# 管道操作
def create_pipeline(*functions):
"""创建函数管道"""
def pipeline(data):
result = data
for func in functions:
result = func(result)
return result
return pipeline
# 创建数据处理管道
process_text = create_pipeline(
str.strip,
str.lower,
lambda x: x.replace(' ', '_'),
partial(str[:], end=10) # 截取前10个字符
)
text = " Hello World Python Programming "
print(f"\n管道操作示例：")
print(f"原始文本: '{text}'")
print(f"处理后: '{process_text(text)}'")
高级技巧与最佳实践
functools的高级用法包括自定义缓存策略、函数记忆和性能优化技巧。
from functools import lru_cache, wraps, partial
import time
import weakref
# 自定义缓存装饰器
def custom_cache(maxsize=128, ttl=60):
"""带过期时间的缓存装饰器"""
def decorator(func):
cache = {}
timestamps = {}
@wraps(func)
def wrapper(*args):
current_time = time.time()
# 检查缓存是否过期
if args in cache:
if current_time - timestamps[args] < ttl:
return cache[args]
else:
del cache[args]
del timestamps[args]
# 执行函数并缓存结果
result = func(*args)
cache[args] = result
timestamps[args] = current_time
# 限制缓存大小
if len(cache) > maxsize:
oldest_key = min(timestamps.keys(), key=lambda k: timestamps[k])
del cache[oldest_key]
del timestamps[oldest_key]
return result
wrapper.cache_clear = lambda: (cache.clear(), timestamps.clear())
return wrapper
return decorator
@custom_cache(maxsize=64, ttl=30)
def expensive_function(n):
"""模拟耗时函数"""
time.sleep(0.01)
return n ** 2
print("自定义缓存装饰器：")
start = time.perf_counter()
result1 = expensive_function(100)
time1 = time.perf_counter() - start
start = time.perf_counter()
result2 = expensive_function(100) # 使用缓存
time2 = time.perf_counter() - start
print(f"首次调用: {result1}, 耗时: {time1*1000:.2f}ms")
print(f"缓存调用: {result2}, 耗时: {time2*1000:.2f}ms")
# 函数记忆与重试机制
def memoize_with_retry(max_retries=3, delay=1):
"""带重试的函数记忆"""
def decorator(func):
cache = {}
retry_counts = {}
@wraps(func)
def wrapper(*args):
# 检查缓存
if args in cache:
return cache[args]
# 重试机制
for attempt in range(max_retries):
try:
result = func(*args)
cache[args] = result
retry_counts[args] = attempt + 1
return result
except Exception as e:
if attempt == max_retries - 1:
raise
time.sleep(delay * (attempt + 1))
raise RuntimeError(f"函数 {func.__name__} 在 {max_retries} 次重试后失败")
wrapper.get_retry_count = lambda args: retry_counts.get(args, 0)
wrapper.clear_cache = lambda: (cache.clear(), retry_counts.clear())
return wrapper
return decorator
@memoize_with_retry(max_retries=3)
def unreliable_function(n):
"""模拟不可靠函数"""
import random
if random.random() < 0.7: # 70%概率失败
raise ValueError("随机失败")
return n * 2
print(f"\n重试机制示例：")
try:
result = unreliable_function(50)
print(f"函数结果: {result}")
except Exception as e:
print(f"函数失败: {e}")
# 性能分析装饰器
def performance_analyzer(func):
"""性能分析装饰器"""
@wraps(func)
def wrapper(*args, **kwargs):
start_time = time.perf_counter()
result = func(*args, **kwargs)
end_time = time.perf_counter()
# 记录性能指标
wrapper.performance_data = {
'call_count': getattr(wrapper, 'performance_data', {}).get('call_count', 0) + 1,
'total_time': getattr(wrapper, 'performance_data', {}).get('total_time', 0) + (end_time - start_time),
'last_time': end_time - start_time
}
return result
def get_stats():
data = getattr(wrapper, 'performance_data', {})
if not data:
return {}
return {
'average_time': data['total_time'] / data['call_count'],
'call_count': data['call_count'],
'total_time': data['total_time']
}
wrapper.get_performance_stats = get_stats
return wrapper
@performance_analyzer
def analyzed_function(n):
"""被分析的函数"""
return sum(i ** 2 for i in range(n))
print(f"\n性能分析示例：")
for i in range(5):
analyzed_function(1000)
stats = analyzed_function.get_performance_stats()
print(f"调用次数: {stats['call_count']}")
print(f"平均时间: {stats['average_time']*1000:.2f}ms")
print(f"总时间: {stats['total_time']*1000:.2f}ms")
functools模块提供了丰富的函数式编程工具，通过掌握partial、lru_cache、reduce等函数的高级用法，开发者可以编写更简洁、高效的代码，实现复杂的函数组合和性能优化策略。这些技术在数据处理、算法优化和系统设计中具有重要应用价值。