Functools -- singledispatchmethod
它是 singledispatch 的 类方法版本。
from functools import singledispatchmethod
功能:
- 根据“方法第一个参数的类型(除 self 外)”自动分发到不同方法
- 实现“面向对象版函数重载”
- 让类的方法根据输入类型自动选择实现
from functools import singledispatchmethod
class Processor:
@singledispatchmethod
def process(self, value):
raise NotImplementedError(f"不支持类型: {type(value)}")
@process.register
def _(self, value: int):
return f"处理整数:{value}"
@process.register
def _(self, value: str):
return f"处理字符串:{value}"
p = Processor()
print(p.process(10))
print(p.process("abc"))
输出:
处理整数:10
处理字符串:abc
👉 根据 value 的类型自动选择不同方法。
你有:
- dict
- list
- str
- bytes
- pandas.DataFrame
- pydantic.BaseModel
函数可以自动分发到对应处理逻辑。
一个类处理多种数据:
p.process(str_data)
p.process(json_dict)
p.process(pandas_dataframe)
非常适合你做的数据同步、股票数据处理框架。
不同的模块可以注册新的类型处理器:
@processor.process.register
def _(self, value: MyCustomType):
...
无需修改主类。
避免几十行的:
if isinstance(x, dict):
elif isinstance(x, list):
elif isinstance(x, str):
...
使用 singledispatchmethod 更优雅。
class Loader:
@singledispatchmethod
def load(self, data):
raise TypeError("不支持的类型")
@load.register
def _(self, data: str):
return f"加载字符串: {data}"
@load.register
def _(self, data: list):
return f"加载列表: {data}"
@load.register
def _(self, data: dict):
return f"加载字典: {data}"
class Handler:
@singledispatchmethod
def handle(self, value):
...
@handle.register(int)
def _(self, value):
return f"INT: {value}"
适合动态场景。
例如:
def process(self, a, b):
分发只看 a,不看 b。
class Animal: pass
class Dog(Animal): pass
class P:
@singledispatchmethod
def speak(self, x):
...
@speak.register
def _(self, x: Animal):
print("Animal speak")
p = P()
p.speak(Dog()) # Dog 自动匹配父类 Animal
| 功能 | singledispatchmethod | match-case |
|---|---|---|
| 按类型分发 | ✅ | ✅ |
| 类方法 | ✅ | ❌ |
| 插件扩展 | ✅ | ❌ |
| 使用继承链 | ✅ | ❌ |
| 可集中管理逻辑 | ✅ | ✅ |
| 扩展性 | ⭐⭐⭐⭐⭐ | ⭐⭐ |
singledispatchmethod 更适合大型项目和扩展性要求高的系统。
class StockParser:
@singledispatchmethod
def parse(self, data):
raise TypeError("不支持的数据类型")
@parse.register
def _(self, data: dict):
return f"解析 dict: {data['code']}"
@parse.register
def _(self, data: str):
return f"解析 JSON 字符串: {data}"
@parse.register
def _(self, data: list):
return [self.parse(item) for item in data]
用法:
parser = StockParser()
parser.parse({"code": "600000"})
parser.parse('[{"code": "600000"}]')
parser.parse([{"code": "002594"}, {"code": "600519"}])
class ParamParser:
@singledispatchmethod
def normalize(self, v):
return v
@normalize.register
def _(self, v: str):
return v.strip()
@normalize.register
def _(self, v: list):
return [self.normalize(i) for i in v]
from functools import singledispatchmethod
class C:
@singledispatchmethod
def method(self, x): ...
@method.register
def _(self, x: int): ...
@method.register
def _(self, x: str): ...
特点:
- 类方法自动“按类型分发”
- 扩展性强,可插件式开发
- 替代大量 if-else
- 首个非-self 参数决定分发
- 支持继承链查找