源码地址：https://github.com/edgedb/edgedb/blob/master/edb/common/value_dispatch.py
鸣谢原文：一日一技：使用装饰器简化大量 if…elif…代码
源码只对同步函数有效果，我将其稍作修改后，让其针对同步，异步，类中的实例方法也具备效果
先看食用方式
这里展示最复杂的例子，即：装饰类中的异步实例方法
如要查看简单的例子，请点击上方的一日一技文章，里面展示了该方式最简单的使用例子
# 示例：
import asyncio
from match_case_model import value_dispatch, value_dispatch_async, value_dispatch_class_async, value_dispatch_class
# 四个装饰器分别对应 value_dispatch: 同步函数， value_dispatch_async: 异步函数，
# value_dispatch_class_async: 异步实例方法， value_dispatch_class: 同步实例方法
class Demo:
@value_dispatch_class_async
async def eat(self, fruit, *args, **kwargs):
"""
主函数，需要选择4个装饰器之一来装饰，这是类中的异步实例方法，所以使用该装饰器
:param fruit: 分支选择参数
:param args, kwargs: 选择分支失败后用来接收原本传给分支的额外参数
"""
return f"I don't want a {fruit}..."
@eat.register('apple') # 注册 fruit == 'apple' 时的函数
async def _eat_apple(self, fruit, feel):
"""
分支函数（被@主函数.register("XXX")装饰的函数）
:param fruit: 调用主函数时传入的分支选择参数，该函数体内可能用不上这个参数，但是必须要接收该参数
:param feel: 分支函数需要的额外参数，只要有一个分支函数接收该参数，那么所有分支函数都要接收该参数
"""
return f"I love {fruit},{feel}"
# @eat.register('eggplant') # 一个函数可以接受多个注册，但推荐使用下面那个方式
# @eat.register('squash')
# async def _eat_what(self, fruit, feel):
# return f"I didn't know {fruit} is a fruit!"
@eat.register_for_all(('eggplant', 'squash')) # 一个函数接受多个注册的方式
async def _eat_what(self, fruit, feel): # 这个分支无需使用feel参数，但还是需要接收
return f"I didn't know {fruit} is a fruit!"
if __name__ == '__main__':
a = Demo()
print(asyncio.run(a.eat("apple", 'happy')))
print(asyncio.run(a.eat("eggplant", 'happy')))
print(asyncio.run(a.eat("origin", 'happy')))
#----------------------------------
# >>> I love apple,happy
# >>> I didn't know eggplant is a fruit!
# >>> I don't want a origin...
修改后的源码
实际是对源码复制了4份，分别做修改后以应对4钟不同的使用场景。
"""
源码：https://github.com/edgedb/edgedb/blob/master/edb/common/value_dispatch.py
按参数值进行分派 适用于 python < 3.10
python >= 3.10 请直接使用 match case 语句
"""
import functools
def value_dispatch(func):
"""
适用于同步函数
"""
registry = {}
@functools.wraps(func)
def wrapper(arg0, *args, **kwargs):
"""
arg0: 即是调用函数时，传递的那个 选择分组的参数
"""
try:
delegate = registry[arg0]
except KeyError:
pass
else:
# 这里传入arg0参数，需要在定义分组函数时，接收该参数，尽管有时并不会使用该参数
return delegate(arg0, *args, **kwargs)
return func(arg0, *args, **kwargs)
__register_tools(wrapper, registry)
return wrapper
def value_dispatch_async(func):
"""
适用于异步函数
"""
registry = {}
@functools.wraps(func)
async def wrapper(arg0, *args, **kwargs):
"""
arg0: 即是调用函数时，传递的那个 选择分组的参数
"""
try:
delegate = registry[arg0]
except KeyError:
pass
else:
# 这里传入arg0参数，需要在定义分组函数时，接收该参数，尽管有时并不会使用该参数
return await delegate(arg0, *args, **kwargs)
return await func(arg0, *args, **kwargs)
__register_tools(wrapper, registry)
return wrapper
def value_dispatch_class(func):
"""
适用于同步的实例方法
"""
registry = {}
@functools.wraps(func)
def wrapper(self, arg0, *args, **kwargs):
"""
arg0: 即是调用函数时，传递的那个 选择分组的参数
"""
try:
delegate = registry[arg0]
except KeyError:
pass
else:
# 这里传入arg0参数，需要在定义分组函数时，接收该参数，尽管有时并不会使用该参数
return delegate(self, arg0, *args, **kwargs)
return func(self, arg0, *args, **kwargs)
__register_tools(wrapper, registry)
return wrapper
def value_dispatch_class_async(func):
"""
适用于异步的实例方法
"""
registry = {}
@functools.wraps(func)
async def wrapper(self, arg0, *args, **kwargs):
"""
arg0: 即是调用函数时，传递的那个 选择分组的参数
"""
try:
delegate = registry[arg0]
except KeyError:
pass
else:
# 这里传入arg0参数，需要在定义分组函数时，接收该参数，尽管有时并不会使用该参数
return await delegate(self, arg0, *args, **kwargs)
return await func(self, arg0, *args, **kwargs)
__register_tools(wrapper, registry)
return wrapper
def __register_tools(wrapper, registry):
def register(value):
def wrap(func):
if value in registry:
raise ValueError(
f'@value_dispatch: there is already a handler '
f'registered for {value!r}'
)
registry[value] = func
return func
return wrap
def register_for_all(values):
def wrap(func):
for value in values:
if value in registry:
raise ValueError(
f'@value_dispatch: there is already a handler '
f'registered for {value!r}'
)
registry[value] = func
return func
return wrap
wrapper.register = register
wrapper.register_for_all = register_for_all