pin_drop当前位置:知识文库 ❯ 图文

Python标准库functools详解

一、functools概述

functools 是Python标准库中用于高阶函数操作的模块。所谓高阶函数,是指以函数为参数或返回值的函数。functools提供了一系列工具,帮助我们更好地处理函数式编程场景,包括偏函数应用、函数缓存、装饰器辅助等。

本教程将重点介绍functools模块中最常用的几个核心工具:partiallru_cachewrapsreduce


二、partial偏函数

partial 用于创建一个"偏函数",即固定原函数的部分参数,生成一个新函数。这在需要重复调用同一函数但某些参数固定不变的场景中非常有用。

1. 基本语法

代码示例

from functools import partial

# partial(func, *args, **kwargs)
# func: 要包装的原函数
# *args: 固定位置参数
# **kwargs: 固定关键字参数

2. 使用示例

代码示例

from functools import partial

# 示例1:固定进制转换的基数
binary = partial(int, base=2)
hex_num = partial(int, base=16)

print(binary('1010'))    # 输出: 10
print(hex_num('1A'))     # 输出: 26

# 示例2:固定乘法因子
double = partial(lambda x, y: x * y, y=2)
triple = partial(lambda x, y: x * y, y=3)

print(double(5))   # 输出: 10
print(triple(5))   # 输出: 15

# 示例3:固定字符串格式化
greet_en = partial("Hello, {}!".format)
greet_cn = partial("你好,{}!".format)

print(greet_en("Alice"))   # 输出: Hello, Alice!
print(greet_cn("小明"))     # 输出: 你好,小明!

3. 实际应用场景

代码示例

from functools import partial
import operator

# 场景1:排序时固定key函数
students = [
    {'name': '张三', 'score': 85, 'age': 20},
    {'name': '李四', 'score': 92, 'age': 19},
    {'name': '王五', 'score': 78, 'age': 21}
]

# 固定按分数排序
sort_by_score = partial(sorted, key=lambda x: x['score'])
print(sort_by_score(students))

# 场景2:map中固定函数
numbers = [1, 2, 3, 4, 5]
power_of_2 = partial(map, lambda x: x ** 2)
print(list(power_of_2(numbers)))  # [1, 4, 9, 16, 25]

三、lru_cache缓存装饰器

lru_cache 是一个装饰器,用于为函数添加缓存功能。LRU代表"Least Recently Used"(最近最少使用),当缓存满时,会自动淘汰最久未被访问的条目。

1. 基本语法

代码示例

from functools import lru_cache

# @lru_cache(maxsize=128, typed=False)
# maxsize: 最大缓存条目数(None表示无限制)
# typed: 如果为True,不同类型的参数分别缓存(如 1 和 1.0)

2. 斐波那契数列优化

代码示例

from functools import lru_cache
import time

# 未使用缓存的斐波那契(递归版本很慢)
def fib_slow(n):
    if n < 2:
        return n
    return fib_slow(n-1) + fib_slow(n-2)

# 使用lru_cache优化
@lru_cache(maxsize=None)
def fib_fast(n):
    if n < 2:
        return n
    return fib_fast(n-1) + fib_fast(n-2)

# 性能对比
start = time.time()
result = fib_fast(100)
print(f"fib_fast(100) = {result}, 耗时: {time.time()-start:.6f}秒")

# 查看缓存信息
print(fib_fast.cache_info())
# 输出: CacheInfo(hits=98, misses=101, maxsize=None, currsize=101)

3. 缓存管理

代码示例

from functools import lru_cache

@lru_cache(maxsize=3)
def expensive_compute(x):
    print(f"计算 {x}...")
    return x ** 3

print(expensive_compute(1))   # 计算 1...  \n 1
print(expensive_compute(2))   # 计算 2...  \n 8
print(expensive_compute(3))   # 计算 3...  \n 27
print(expensive_compute(1))   # 直接从缓存: 1
print(expensive_compute(4))   # 计算 4...(缓存已满,淘汰最旧的)\n 64
print(expensive_compute(2))   # 计算 2...(已被淘汰)\n 8

# 查看缓存统计
print(expensive_compute.cache_info())

# 清空缓存
expensive_compute.cache_clear()
缓存方法 说明 适用场景
lru_cache 基于LRU策略自动淘汰 递归函数、重复计算
手动字典缓存 需要自己管理缓存逻辑 复杂缓存策略需求
第三方缓存库 如cachetools、redis 分布式、持久化缓存

四、wraps装饰器

wraps 是编写装饰器时的必备工具。它用于将被装饰函数的元信息(如 __name____doc__ 等)复制到包装函数上,避免装饰器破坏原函数的元信息。

1. 不使用wraps的问题

代码示例

# 错误示例:未使用wraps
def my_decorator(func):
    def wrapper(*args, **kwargs):
        """wrapper函数的文档字符串"""
        print("调用前...")
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def say_hello(name):
    """向某人打招呼"""
    print(f"Hello, {name}!")

say_hello("World")
print(say_hello.__name__)   # 输出: wrapper(错误!)
print(say_hello.__doc__)    # 输出: wrapper函数的文档字符串(错误!)

2. 正确使用wraps

代码示例

from functools import wraps

# 正确示例:使用wraps
def my_decorator(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        """wrapper函数的文档字符串"""
        print("调用前...")
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def say_hello(name):
    """向某人打招呼"""
    print(f"Hello, {name}!")

say_hello("World")
print(say_hello.__name__)   # 输出: say_hello(正确!)
print(say_hello.__doc__)    # 输出: 向某人打招呼(正确!)

3. 实用的日志装饰器

代码示例

from functools import wraps
import time

def log_execution(func):
    """记录函数执行时间和参数的装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        elapsed = time.time() - start
        print(f"[日志] {func.__name__} 执行耗时: {elapsed:.4f}秒")
        return result
    return wrapper

@log_execution
def process_data(n):
    """模拟数据处理"""
    time.sleep(0.1)
    return sum(range(n))

result = process_data(10000)
print(f"结果: {result}")
print(f"函数名: {process_data.__name__}")    # process_data
print(f"文档: {process_data.__doc__}")      # 模拟数据处理

五、reduce函数

reduce 在Python 3中被移入functools模块。它对一个序列从左到右累积应用二元函数,将序列缩减为单一值。

1. 基本语法

代码示例

from functools import reduce

# reduce(function, iterable[, initializer])
# function: 接收两个参数的函数
# iterable: 要处理的可迭代对象
# initializer: 可选的初始值

2. 使用示例

代码示例

from functools import reduce

# 示例1:求和
numbers = [1, 2, 3, 4, 5]
total = reduce(lambda x, y: x + y, numbers)
print(total)  # 输出: 15

# 示例2:求乘积
product = reduce(lambda x, y: x * y, numbers)
print(product)  # 输出: 120

# 示例3:找最大值
max_val = reduce(lambda x, y: x if x > y else y, numbers)
print(max_val)  # 输出: 5

# 示例4:字符串拼接
words = ['Python', ' ', 'is', ' ', 'awesome']
sentence = reduce(lambda x, y: x + y, words)
print(sentence)  # Python is awesome

# 示例5:使用初始值
total_with_init = reduce(lambda x, y: x + y, numbers, 100)
print(total_with_init)  # 输出: 115

3. reduce vs sum/其他内置函数

代码示例

from functools import reduce
import operator

numbers = [1, 2, 3, 4, 5]

# 推荐使用内置函数(更清晰、更高效)
print(sum(numbers))           # 求和: 15
print(max(numbers))           # 最大值: 5
print(min(numbers))           # 最小值: 1
print(all(n > 0 for n in numbers))  # 全部为正: True

# 使用operator模块替代lambda
print(reduce(operator.add, numbers))      # 求和: 15
print(reduce(operator.mul, numbers))      # 乘积: 120
print(reduce(operator.and_, [True, True, False]))  # 逻辑与: False

六、代码示例与实战应用

实战1:构建API调用缓存

代码示例

from functools import lru_cache
import hashlib

# 模拟一个耗时的API调用
@lru_cache(maxsize=100)
def fetch_user_data(user_id):
    """模拟从API获取用户数据"""
    print(f"正在获取用户 {user_id} 的数据...")
    # 实际场景中这里是HTTP请求
    return {"id": user_id, "name": f"用户{user_id}", "score": user_id * 10}

# 第一次调用:实际请求
user1 = fetch_user_data(1)
print(user1)

# 第二次调用:直接返回缓存
user1_again = fetch_user_data(1)
print(user1_again)

# 不同参数:新的请求
user2 = fetch_user_data(2)
print(user2)

实战2:函数计时装饰器

代码示例

from functools import wraps
import time

def timer(func):
    """计算函数执行时间的装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"[{func.__name__}] 耗时: {elapsed:.6f} 秒")
        return result
    return wrapper

@timer
def compute_sum(n):
    return sum(range(n))

@timer
def compute_product(n):
    result = 1
    for i in range(1, n + 1):
        result *= i
    return result

compute_sum(1000000)
compute_product(1000)

实战3:配置化函数创建

代码示例

from functools import partial

# 场景:为不同环境创建不同配置的函数
def create_db_connection(host, port, db_name, timeout=30):
    """模拟创建数据库连接"""
    return f"连接到 {host}:{port}/{db_name} (超时:{timeout}秒)"

# 为不同环境创建偏函数
dev_db = partial(create_db_connection, 'localhost', 5432, timeout=10)
prod_db = partial(create_db_connection, 'prod.example.com', 5432, timeout=60)

print(dev_db('dev_database'))
# 输出: 连接到 localhost:5432/dev_database (超时:10秒)

print(prod_db('production_database'))
# 输出: 连接到 prod.example.com:5432/production_database (超时:60秒)

实战4:数据聚合管道

代码示例

from functools import reduce

# 数据处理管道
data = [
    {"name": "张三", "department": "技术部", "salary": 15000},
    {"name": "李四", "department": "技术部", "salary": 18000},
    {"name": "王五", "department": "市场部", "salary": 12000},
    {"name": "赵六", "department": "市场部", "salary": 14000},
]

# 使用reduce按部门分组
def group_by_department(acc, person):
    dept = person["department"]
    if dept not in acc:
        acc[dept] = []
    acc[dept].append(person)
    return acc

dept_groups = reduce(group_by_department, data, {})

for dept, members in dept_groups.items():
    total_salary = sum(m["salary"] for m in members)
    avg_salary = total_salary / len(members)
    print(f"{dept}: 平均薪资 {avg_salary:.0f} 元, 人数 {len(members)}")

七、注意事项与最佳实践

1. lru_cache的缓存键要求

注意lru_cache 要求函数的所有参数都必须是可哈希的(hashable)。如果函数接受列表、字典等不可哈希类型作为参数,会导致 TypeError

代码示例

from functools import lru_cache

# 错误:列表不可哈希
# @lru_cache()
# def process_data(items):  # items是列表
#     return sum(items)

# 正确:将列表转为元组
@lru_cache()
def process_data(items_tuple):
    return sum(items_tuple)

result = process_data((1, 2, 3, 4, 5))
print(result)  # 15

2. 缓存污染与内存泄漏

注意:使用 lru_cache(maxsize=None) 时无限制缓存,如果参数组合非常多,可能导致内存泄漏。建议始终设置合理的 maxsize 值。

3. partial不保留原函数签名

注意partial 创建的函数不会保留原函数的签名信息,这可能影响IDE的自动补全和类型检查。在Python 3.4+中,可以使用 functools.update_wrapper 来改善这种情况。

小贴士

在Python 3.9+中,@cache@lru_cache(maxsize=None) 的简写形式,语义更清晰。如果你的缓存不需要淘汰策略,优先使用 @cache。更多关于functools的详细信息,可以参考Python官方文档


八、课程小结

  • partial偏函数:固定函数的部分参数,生成新函数,适合参数复用场景

  • lru_cache缓存:为函数添加LRU缓存,大幅提升重复调用的性能,特别适合递归函数

  • wraps装饰器:编写装饰器时必须使用,确保被装饰函数的元信息不被破坏

  • reduce归约函数:将序列归约为单一值,虽然很多场景可用内置函数替代,但在复杂归约逻辑中仍然有用

  • 函数式编程:functools是Python函数式编程的重要工具,合理使用可以让代码更简洁、更高效

常见问题

lru_cache缓存的是什么?

lru_cache缓存的是函数的返回值。当函数被相同的参数再次调用时,它不会重新执行函数体,而是直接返回之前缓存的结果。缓存的键由函数的所有参数值组合而成。

partial和普通lambda函数有什么区别?

虽然lambda也能创建匿名函数,但partial更高效。partial是C语言实现的,而且它保留了对原函数的引用。另外,partial比lambda有更好的可读性,因为它的意图更明确。

写装饰器时必须用wraps吗?

虽然不是强制的,但强烈建议使用。不用wraps会导致被装饰函数的__name__、__doc__等属性被替换为wrapper函数的属性,这会影响调试、文档生成和依赖反射的工具(如help()函数)。

reduce函数为什么被移到functools?

Python 3将reduce从内置函数移到functools模块,是因为大多数reduce场景可以用sum()、max()、min()等更直观的内置函数替代。只有复杂的归约逻辑才需要reduce,因此它不再是高频使用的内置函数。

练习1

使用lru_cache装饰器优化一个递归计算阶乘的函数,然后比较使用缓存和不使用缓存时计算 factorial(100) 的性能差异。使用 cache_info() 方法查看缓存命中情况。

练习2

编写一个带重试功能的装饰器(使用wraps),该装饰器在函数抛出异常时自动重试指定次数。使用partial创建一个重试3次的装饰器变体,并在实际场景中应用。

标签: functools 装饰器 Python标准库 lru_cache 高阶函数 函数式编程

本文涉及AI创作

内容由AI创作,请仔细甄别

list快速访问

上一篇: Python标准库itertools详解 下一篇: Python标准库pathlib详解

poll相关推荐