functools 模块

functools.lru_cache(maxsize=128, typed=False)
functools.cache                    # Python 3.9+，等价于 lru_cache(maxsize=None)

# Python 3.10+
from functools import lru_cache, cache
import time

# ========== 1. 基本用法：缓存递归（斐波那契数列） ==========
@lru_cache(maxsize=128)
def fib(n: int) -> int:
    """计算第 n 项斐波那契数（带缓存）"""
    if n < 2:
        return n
    return fib(n - 1) + fib(n - 2)

# 无缓存版本（对比）
def fib_no_cache(n: int) -> int:
    if n < 2:
        return n
    return fib_no_cache(n - 1) + fib_no_cache(n - 2)

# 对比性能
start = time.perf_counter()
result = fib(35)
elapsed_cached = time.perf_counter() - start
print(f"fib(35) 带缓存: {result}, 耗时 {elapsed_cached:.6f}s")
print(f"缓存统计: {fib.cache_info()}")

start = time.perf_counter()
result2 = fib_no_cache(35)
elapsed_no_cache = time.perf_counter() - start
print(f"fib(35) 无缓存: {result2}, 耗时 {elapsed_no_cache:.6f}s")
print(f"加速比: {elapsed_no_cache / elapsed_cached:.0f}x")

# ========== 2. typed 参数：区分参数类型 ==========
from functools import lru_cache

@lru_cache(maxsize=128, typed=True)
def double(x):
    return x * 2

print(double(3))      # 6 — 缓存键: (int, 3)
print(double(3.0))    # 6.0 — 缓存键: (float, 3.0)，与上面分开缓存
print(double.cache_info())  # hits=0, misses=2, maxsize=128, currsize=2

# ========== 3. @cache 快捷方式（Python 3.9+） ==========
@cache
def slow_square(n: int) -> int:
    """模拟耗时计算"""
    time.sleep(0.01)
    return n * n

# 首次调用：计算并缓存
slow_square(10)  # 耗时 ~10ms
# 第二次调用：命中缓存
slow_square(10)  # 耗时 < 1μs

# ========== 4. 缓存管理与监控 ==========
@lru_cache(maxsize=4)
def fetch_user(user_id: int) -> str:
    """模拟数据库查询"""
    time.sleep(0.005)
    return f"User_{user_id}"

# 依次查询 5 个用户（第 5 个会淘汰第 1 个）
for uid in [1, 2, 3, 4, 5]:
    fetch_user(uid)
print(f"缓存状态: {fetch_user.cache_info()}")
# hits=0, misses=5, maxsize=4, currsize=4 — 前 4 个被淘汰 1 个

# 再次查询：命中缓存
fetch_user(2)  # 命中
fetch_user(1)  # 未命中！已被淘汰
print(f"缓存状态: {fetch_user.cache_info()}")
# hits=1, misses=6, maxsize=4, currsize=4

# 清空缓存
fetch_user.cache_clear()
print(f"清空后: {fetch_user.cache_info()}")
# hits=0, misses=0, maxsize=4, currsize=0

# Python 3.10+

def bad_decorator(func):
    """没有使用 @wraps 的装饰器"""
    def wrapper(*args, **kwargs):
        """wrapper 的文档字符串"""
        return func(*args, **kwargs)
    return wrapper

def good_decorator(func):
    """使用了 @wraps 的装饰器"""
    from functools import wraps

    @wraps(func)
    def wrapper(*args, **kwargs):
        """wrapper 的文档字符串"""
        return func(*args, **kwargs)
    return wrapper

@bad_decorator
def greet(name: str) -> str:
    """向用户问好"""
    return f"Hello, {name}!"

@good_decorator
def farewell(name: str) -> str:
    """向用户告别"""
    return f"Goodbye, {name}!"

# 对比：没有 @wraps
print(greet.__name__)   # 'wrapper' — 丢失了原函数名！
print(greet.__doc__)    # "wrapper 的文档字符串" — 丢失了原文档！
print(greet.__wrapped__)  # AttributeError — 无法访问原函数

# 对比：有 @wraps
print(farewell.__name__)  # 'farewell' — 正确保留
print(farewell.__doc__)   # '向用户告别' — 正确保留
print(farewell.__wrapped__)  # <function farewell at ...> — 可访问原函数

functools.wraps(wrapped, assigned=WRAPPER_ASSIGNMENTS, updated=WRAPPER_UPDATES)

# Python 3.10+
from functools import wraps
import time

def retry(max_attempts: int = 3, delay: float = 1.0):
    """带参数的重试装饰器"""
    def decorator(func):
        @wraps(func)  # 保留原函数元数据
        def wrapper(*args, **kwargs):
            import time as _time
            last_exception = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e
                    if attempt < max_attempts:
                        print(f"[重试 {attempt}/{max_attempts}] {func.__name__} 失败: {e}")
                        _time.sleep(delay)
            raise last_exception
        return wrapper
    return decorator

@retry(max_attempts=3, delay=0.1)
def unstable_network_call(url: str) -> str:
    """模拟不稳定的网络请求"""
    import random
    if random.random() < 0.7:
        raise ConnectionError("网络超时")
    return f"Response from {url}"

# 测试
print(unstable_network_call.__name__)  # 'unstable_network_call' — 正确保留
print(unstable_network_call.__doc__)   # '模拟不稳定的网络请求' — 正确保留
# result = unstable_network_call("https://api.example.com")

# Python 3.10+
from functools import total_ordering

@total_ordering
class Version:
    """语义化版本号（支持全序比较）"""

    def __init__(self, major: int, minor: int, patch: int):
        self.major = major
        self.minor = minor
        self.patch = patch

    def __eq__(self, other) -> bool:
        if not isinstance(other, Version):
            return NotImplemented
        return (self.major, self.minor, self.patch) == (other.major, other.minor, other.patch)

    def __lt__(self, other) -> bool:
        """只需定义 __lt__，其余自动推导"""
        if not isinstance(other, Version):
            return NotImplemented
        return (self.major, self.minor, self.patch) < (other.major, other.minor, other.patch)

    def __repr__(self):
        return f"v{self.major}.{self.minor}.{self.patch}"


# 测试：所有比较运算符自动可用
v1 = Version(1, 0, 0)
v2 = Version(1, 2, 0)
v3 = Version(2, 0, 0)

print(v1 < v2)       # True — __lt__
print(v1 <= v2)      # True — 自动推导
print(v1 > v2)       # False — 自动推导
print(v1 >= v2)      # False — 自动推导
print(v1 == v2)      # False — __eq__
print(v1 != v2)      # True — 自动推导（基于 __eq__）
print(v2 < v3)       # True

# 排序演示
versions = [v3, v1, v2]
print(sorted(versions))  # [v1.0.0, v1.2.0, v2.0.0]

# Python 3.10+
from functools import singledispatch
from collections.abc import Iterable
import json

@singledispatch
def serialize(obj) -> str:
    """默认序列化：使用 repr"""
    return repr(obj)

@serialize.register(int)
def _(obj: int) -> str:
    return f"Integer({obj})"

@serialize.register(float)
def _(obj: float) -> str:
    return f"Float({obj:.2f})"

@serialize.register(str)
def _(obj: str) -> str:
    return f'String("{obj}")'

@serialize.register(list)
def _(obj: list) -> str:
    items = ", ".join(serialize(item) for item in obj)
    return f"List([{items}])"

@serialize.register(dict)
def _(obj: dict) -> str:
    items = ", ".join(f"{serialize(k)}: {serialize(v)}" for k, v in obj.items())
    return f"Dict({{{items}}})"

@serialize.register(Iterable)  # 抽象基类：匹配所有可迭代对象
def _(obj: Iterable) -> str:
    items = ", ".join(serialize(item) for item in obj)
    return f"Iterable({items})"

# 测试
print(serialize(42))                    # Integer(42)
print(serialize(3.14159))               # Float(3.14)
print(serialize("hello"))               # String("hello")
print(serialize([1, "two", 3.0]))       # List([Integer(1), String("two"), Float(3.00)])
print(serialize({"a": 1, "b": 2}))      # Dict({String("a"): Integer(1), String("b"): Integer(2)})
print(serialize((1, 2, 3)))             # Iterable(Integer(1), Integer(2), Integer(3))

# Python 3.10+
from functools import singledispatchmethod

class DataProcessor:
    """根据输入类型自动选择处理方式"""

    @singledispatchmethod
    def process(self, data):
        raise TypeError(f"不支持的数据类型: {type(data).__name__}")

    @process.register(int)
    def _(self, data: int):
        return f"处理整数: {data ** 2}"

    @process.register(str)
    def _(self, data: str):
        return f"处理字符串: {data.upper()}"

    @process.register(list)
    def _(self, data: list):
        return f"处理列表: 长度={len(data)}, 总和={sum(data)}"

    @process.register(dict)
    def _(self, data: dict):
        return f"处理字典: 键={list(data.keys())}"


# 测试
processor = DataProcessor()
print(processor.process(10))           # 处理整数: 100
print(processor.process("hello"))      # 处理字符串: HELLO
print(processor.process([1, 2, 3]))    # 处理列表: 长度=3, 总和=6
print(processor.process({"a": 1}))     # 处理字典: 键=['a']
# processor.process(3.14)              # TypeError: 不支持的数据类型: float

# Python 3.10+
from functools import partial, partialmethod

# ========== 1. partial 基本用法 ==========
def power(base: float, exponent: float) -> float:
    """计算 base 的 exponent 次方"""
    return base ** exponent

# 固定 exponent 参数，创建平方和立方函数
square = partial(power, exponent=2)
cube = partial(power, exponent=3)

print(square(5))   # 25 — 等价于 power(5, 2)
print(cube(5))     # 125 — 等价于 power(5, 3)

# ========== 2. partial 覆盖已绑定的参数 ==========
greet = partial(print, "Hello", sep=", ")
greet("World")     # Hello, World — 等价于 print("Hello", "World", sep=", ")

# 覆盖 sep 参数
greet("World", sep=" | ")  # Hello | World

# ========== 3. partial 实战：简化回调注册 ==========
import tkinter as tk

def on_button_click(button_id: str, event=None):
    """按钮点击回调"""
    print(f"Button {button_id} clicked!")

# 不使用 partial：需要嵌套 lambda
# btn.config(command=lambda: on_button_click("save"))

# 使用 partial：语义清晰
save_callback = partial(on_button_click, "save")
cancel_callback = partial(on_button_click, "cancel")
# btn.config(command=save_callback)

# ========== 4. partialmethod — 类方法参数预绑定 ==========
class RequestBuilder:
    """HTTP 请求构建器"""

    def send(self, method: str, url: str, **headers):
        return f"{method} {url} | Headers: {headers}"

    # partialmethod 自动处理 self 传递
    get = partialmethod(send, "GET")
    post = partialmethod(send, "POST")
    put = partialmethod(send, "PUT")

    # 可以进一步绑定默认 headers
    json_get = partialmethod(send, "GET", Accept="application/json")


builder = RequestBuilder()
print(builder.get("/api/users"))           # GET /api/users | Headers: {}
print(builder.post("/api/users"))          # POST /api/users | Headers: {}
print(builder.json_get("/api/data"))       # GET /api/data | Headers: {'Accept': 'application/json'}

# Python 3.10+
from functools import reduce
import operator

# ========== reduce 仍然适用的场景 ==========

# 1. 深度合并字典（无内置函数可替代）
def deep_merge(d1: dict, d2: dict) -> dict:
    """递归合并两个字典"""
    result = dict(d1)
    for key, value in d2.items():
        if key in result and isinstance(result[key], dict) and isinstance(value, dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = value
    return result

configs = [
    {"db": {"host": "localhost", "port": 5432}},
    {"db": {"port": 5433}, "cache": {"ttl": 300}},
    {"cache": {"ttl": 600}},
]
final = reduce(deep_merge, configs)
print(final)
# {'db': {'host': 'localhost', 'port': 5433}, 'cache': {'ttl': 600}}

# 2. 链式函数组合（compose）
def compose(f, g):
    """返回 f ∘ g"""
    return lambda x: f(g(x))

# 多个函数的流水线
pipeline = reduce(
    compose,
    [
        lambda x: x.strip(),
        lambda x: x.lower(),
        lambda x: x.replace(" ", "-"),
    ],
)
print(pipeline("  Hello World  "))  # hello-world

# 3. 使用 operator 模块避免 lambda
from operator import add, mul

numbers = [1, 2, 3, 4, 5]
print(reduce(add, numbers))   # 15 — 等价于 sum
print(reduce(mul, numbers))   # 120 — 等价于 math.prod

# Python 3.10+
from functools import cached_property
import time

class Report:
    def __init__(self, data: list[int]):
        self.data = data

    @property
    def average(self) -> float:
        """每次访问都重新计算"""
        print("  [计算中...]")
        return sum(self.data) / len(self.data)

    @cached_property
    def cached_average(self) -> float:
        """首次访问计算，之后缓存"""
        print("  [计算中...]")
        return sum(self.data) / len(self.data)

# 测试
r = Report([1, 2, 3, 4, 5])
print("=== @property ===")
print(r.average)  # [计算中...] 3.0
print(r.average)  # [计算中...] 3.0 — 每次都重新计算！
print(r.average)  # [计算中...] 3.0

print("=== @cached_property ===")
print(r.cached_average)  # [计算中...] 3.0
print(r.cached_average)  # 直接返回 3.0 — 不重新计算！
print(r.cached_average)  # 直接返回 3.0

# Python 3.10+
from functools import cached_property
import threading
import time

class ThreadSafeReport:
    """线程安全的 cached_property 封装"""

    def __init__(self, data: list[int]):
        self.data = data
        self._lock = threading.Lock()

    @cached_property
    def average(self) -> float:
        """注意：cached_property 本身不是线程安全的"""
        time.sleep(0.1)  # 模拟耗时计算
        return sum(self.data) / len(self.data)

    # 如需线程安全，改用显式锁
    def get_average_safe(self) -> float:
        if "average_safe" not in self.__dict__:
            with self._lock:
                if "average_safe" not in self.__dict__:  # 双重检查
                    time.sleep(0.1)
                    self.__dict__["average_safe"] = sum(self.data) / len(self.data)
        return self.__dict__["average_safe"]

# Python 3.10+
# 问题演示：方法上的 lru_cache 导致内存泄漏

from functools import lru_cache
import gc

class LeakyProcessor:
    def __init__(self, name: str):
        self.name = name

    @lru_cache(maxsize=128)
    def process(self, data: int) -> int:
        """缓存绑定到实例 + 参数，self 被缓存引用"""
        return data * 2


# 模拟频繁创建和销毁实例
for i in range(1000):
    p = LeakyProcessor(f"worker-{i}")
    p.process(i)  # self 进入缓存键，缓存持有实例引用
    # 实例 p 超出作用域，但缓存中的 self 引用阻止 GC！

# 解决方案 1：使用 cached_property 替代
class FixedProcessor:
    def __init__(self, name: str, data: int):
        self.name = name
        self.data = data

    @lru_cache(maxsize=128)
    def process(self) -> int:
        """仅依赖 self 属性，不缓存外部参数"""
        return self.data * 2

# Python 3.10+
from functools import lru_cache

@lru_cache(maxsize=None)
def edit_distance(s1: str, s2: str) -> int:
    """
    计算两个字符串的编辑距离（Levenshtein 距离）
    允许插入、删除、替换三种操作
    """
    if not s1:
        return len(s2)
    if not s2:
        return len(s1)

    if s1[-1] == s2[-1]:
        return edit_distance(s1[:-1], s2[:-1])

    return 1 + min(
        edit_distance(s1[:-1], s2),       # 删除
        edit_distance(s1, s2[:-1]),       # 插入
        edit_distance(s1[:-1], s2[:-1]),  # 替换
    )


# 测试
word1 = "kitten"
word2 = "sitting"
print(f"编辑距离('{word1}', '{word2}') = {edit_distance(word1, word2)}")
print(f"缓存统计: {edit_distance.cache_info()}")
# 无缓存时该算法复杂度为 O(3^n)，带缓存后降至 O(n*m)

# 对比：不使用缓存
def edit_distance_slow(s1: str, s2: str) -> int:
    if not s1:
        return len(s2)
    if not s2:
        return len(s1)
    if s1[-1] == s2[-1]:
        return edit_distance_slow(s1[:-1], s2[:-1])
    return 1 + min(
        edit_distance_slow(s1[:-1], s2),
        edit_distance_slow(s1, s2[:-1]),
        edit_distance_slow(s1[:-1], s2[:-1]),
    )

import time
start = time.perf_counter()
result = edit_distance_slow("kitten", "sitting")
elapsed = time.perf_counter() - start
print(f"无缓存耗时: {elapsed:.4f}s")

# Python 3.10+
from functools import singledispatch
from collections.abc import Iterable, Mapping
import json
from datetime import datetime, date


@singledispatch
def to_json_serializable(obj, **kwargs) -> object:
    """将任意对象转换为 JSON 可序列化的形式"""
    try:
        # 尝试直接序列化
        json.dumps(obj)
        return obj
    except (TypeError, ValueError):
        return str(obj)  # 最终兜底


@to_json_serializable.register(datetime)
def _(obj: datetime, **kwargs) -> str:
    date_format = kwargs.get("date_format", "%Y-%m-%dT%H:%M:%S")
    return obj.strftime(date_format)


@to_json_serializable.register(date)
def _(obj: date, **kwargs) -> str:
    return obj.isoformat()


@to_json_serializable.register(bytes)
def _(obj: bytes, **kwargs) -> str:
    encoding = kwargs.get("encoding", "utf-8")
    return obj.decode(encoding, errors="replace")


@to_json_serializable.register(set)
def _(obj: set, **kwargs) -> list:
    return sorted(obj)  # set 不可序列化，转为排序列表


@to_json_serializable.register(Mapping)
def _(obj: Mapping, **kwargs) -> dict:
    return {
        to_json_serializable(k, **kwargs): to_json_serializable(v, **kwargs)
        for k, v in obj.items()
    }


@to_json_serializable.register(Iterable)
def _(obj: Iterable, **kwargs) -> list:
    # 注意：str 也是 Iterable，但 str 已在上面注册为具体类型
    return [to_json_serializable(item, **kwargs) for item in obj]


# 测试完整的数据管道
complex_data = {
    "name": "Alice",
    "created_at": datetime(2026, 6, 5, 14, 30, 0),
    "birthday": date(1995, 8, 15),
    "avatar": b"binary\xffdata",
    "tags": {"python", "developer", "backend"},
    "scores": [95, 87, 92],
    "metadata": {
        "last_login": datetime(2026, 6, 4, 9, 0, 0),
        "preferences": {"theme", "dark", "compact"},
    },
}

result = to_json_serializable(complex_data)
print(json.dumps(result, indent=2, ensure_ascii=False))

# Python 3.10+
from functools import partial, lru_cache
import time
import hashlib


def cached_api_call(cache_size: int = 256):
    """
    通用的 API 调用缓存装饰器工厂
    返回一个已配置的装饰器
    """
    def decorator(func):
        @lru_cache(maxsize=cache_size)
        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)
        return wrapper
    return decorator


# 使用 partial 创建不同缓存策略的变体
cache_large = partial(cached_api_call, cache_size=1024)
cache_small = partial(cached_api_call, cache_size=64)


@cache_large()
def fetch_user_profile(user_id: int) -> dict:
    """获取用户资料（大缓存，用户数据相对稳定）"""
    time.sleep(0.01)  # 模拟 API 调用
    return {"id": user_id, "name": f"User_{user_id}"}


@cache_small()
def fetch_stock_price(symbol: str) -> float:
    """获取股票价格（小缓存，价格变化频繁）"""
    time.sleep(0.01)  # 模拟 API 调用
    return hash(symbol) % 1000 / 10.0


# 测试
print(fetch_user_profile(1))
print(fetch_user_profile(1))  # 命中缓存
print(f"用户缓存状态: {fetch_user_profile.cache_info()}")

print(fetch_stock_price("AAPL"))
print(fetch_stock_price("AAPL"))  # 命中缓存
print(f"股票缓存状态: {fetch_stock_price.cache_info()}")