Python singledispatch的实现示例

from functools import singledispatch

# 1. 定义基础泛型函数（为object类型注册）
@singledispatch
def process_data(data, verbose=False):
    """处理数据的通用函数"""
    if verbose:
        print(f"Processing generic data: {type(data).__name__}")
    return str(data)

# 2. 注册特定类型的实现（三种方式）
# 方式1：显式指定类型
@process_data.register(int)
def _(data: int, verbose=False):
    if verbose:
        print(f"Processing integer: {data}")
    return data * 2

# 方式2：使用类型注解（Python 3.7+）
@process_data.register
def _(data: list, verbose=False):
    if verbose:
        print(f"Processing list with {len(data)} elements")
    return [x * 2 for x in data]

# 方式3：函数式注册（支持lambda和已有函数）
process_data.register(float, lambda data, verbose=False: data * 3)

# 3. 注册联合类型（Python 3.11+）
from typing import Union
@process_data.register
def _(data: Union[tuple, set], verbose=False):
    if verbose:
        print(f"Processing collection: {type(data).__name__}")
    return list(data)

# 调用时自动根据第一个参数类型分派
print(process_data(42))                # 84 (int实现)
print(process_data([1, 2, 3]))         # [2, 4, 6] (list实现)
print(process_data(3.14))              # 9.42 (float实现)
print(process_data("hello"))           # "hello" (默认object实现)
print(process_data((1, 2, 3)))         # [1, 2, 3] (Union实现)

# 检查分派行为
print(process_data.dispatch(list))     # <function _ at 0x...>
print(process_data.registry.keys())    # dict_keys([<class 'object'>, <class 'int'>, <class 'list'>, ...])

from collections.abc import Mapping

@singledispatch
def serialize(obj):
    return f"Generic: {obj}"

@serialize.register(Mapping)
def _(obj):
    return f"Mapping: {dict(obj)}"

# 字典会匹配Mapping实现，尽管未显式注册dict类型
print(serialize({"a": 1}))  # Mapping: {'a': 1}

def _dispatch(self, arg):
    cls = type(arg)
    
    # 1. 检查缓存
    if cls in self._cache:
        return self._cache[cls]
    
    # 2. 生成扩展MRO（包含ABC）
    mro = self._get_extended_mro(cls)
    
    # 3. 查找最佳匹配
    for typ in mro:
        if typ in self._registry:
            self._cache[cls] = self._registry[typ]
            return self._registry[typ]
    
    # 4. 兜底（理论上不会触发，因为注册了object类型）
    return self._registry[object]

def _get_extended_mro(self, cls):
    # 生成包含ABC的扩展MRO
    mro = list(cls.__mro__)
    abc_list = []
    
    # 收集所有相关ABC
    for abc in self._registry:
        if abc is not object and issubclass(cls, abc):
            abc_list.append(abc)
    
    # 排序并去重，确保正确的继承顺序
    abc_list = sorted(abc_list, key=lambda x: len(x.__mro__), reverse=True)
    extended_mro = []
    for typ in mro:
        extended_mro.append(typ)
        # 插入相关ABC到对应位置
        for abc in abc_list:
            if issubclass(typ, abc) and not any(issubclass(base, abc) for base in typ.__bases__):
                extended_mro.append(abc)
    
    return list(dict.fromkeys(extended_mro))  # 去重保持顺序

from collections.abc import Iterable, Container

class P:
    pass

Iterable.register(P)
Container.register(P)

@singledispatch
def g(obj):
    return "base"

g.register(Iterable, lambda obj: "iterable")
g.register(Container, lambda obj: "container")

# 以下调用会抛出RuntimeError: Ambiguous dispatch
# print(g(P()))

def process_data(data):
    if isinstance(data, int):
        return data * 2
    elif isinstance(data, str):
        return data.upper()
    elif isinstance(data, list):
        return [x * 2 for x in data]
    else:
        return str(data)

@singledispatch
def process_data(data):
    return str(data)

@process_data.register(int)
def _(data):
    return data * 2

@process_data.register(str)
def _(data):
    return data.upper()

@process_data.register(list)
def _(data):
    return [x * 2 for x in data]

@singledispatch
def serialize(obj):
    """通用序列化函数"""
    raise TypeError(f"Unsupported type: {type(obj)}")

@serialize.register(int)
@serialize.register(float)
def _(obj):
    return {"type": type(obj).__name__, "value": obj}

@serialize.register(str)
def _(obj):
    return {"type": "str", "value": obj}

@serialize.register(list)
def _(obj):
    return {"type": "list", "value": [serialize(item) for item in obj]}

# 轻松扩展自定义类型
class Person:
    def __init__(self, name, age):
        self.name = name
        self.age = age

@serialize.register(Person)
def _(obj):
    return {"type": "Person", "name": obj.name, "age": obj.age}

from flask import jsonify

@singledispatch
def format_response(data):
    """格式化API响应"""
    return jsonify({"status": "success", "data": str(data)})

@format_response.register(dict)
def _(data):
    return jsonify({"status": "success", **data})

@format_response.register(list)
def _(data):
    return jsonify({
        "status": "success",
        "count": len(data),
        "data": data
    })

@format_response.register(Exception)
def _(data):
    return jsonify({
        "status": "error",
        "message": str(data)
    }), 500

from functools import singledispatchmethod

class DataProcessor:
    @singledispatchmethod
    def process(self, data):
        raise NotImplementedError(f"Unsupported type: {type(data)}")
    
    @process.register
    def _(self, data: int):
        return data * 2
    
    @process.register
    def _(self, data: str):
        return data.upper()
    
    @process.register
    def _(self, data: list):
        return [self.process(item) for item in data]

processor = DataProcessor()
print(processor.process(42))       # 84
print(processor.process([1, "abc"]))  # [2, "ABC"]