当遇到比如回调信息，比如订单中不同商店的特有字段处理，可以使用元类进行自动处理
假如现在你是一个第三方系统对接不同的电商平台，这些电商平台绝大部分字段是共用的，但是少部分字段为某个平台独有
怎么办？对每个平台进行单独校验配置？
这种情况可以使用元类对需要处理的数据进行汇总，然后自动处理
# 元类
网上有很多介绍，这里不累赘，可以参考Python元类小结.md
# 自动处理框架实现
我们需要什么，首先，一个元类是毋庸置疑的，每个平台对应都需要一个单独的处理Handler，Handler中实现每个平台特有的业务逻辑，以及一些属性，这些Handler需要一个父类，这个父类的metaclass就是一开始定义的元类
总结如下:
- 实现一个元类
- 实现一个父类
- 实现不同平台对应的class并继承上面的父类
# 基础框架伪代码大概向下面这个样子
import inspect
import logging
from typing import Any
# 全局映射字典
PLATFORM_FIELDS_DICT = {}
# 全局映射列表
PLATFORM_FIELDS_LIST = []
logger = logging.getLogger(__name__)
class TaskException(Exception):
pass
class TaskBaseMeta(type):
def __new__(cls, name, bases, attrs): # noqa
# 如果实现的是第一个父类，则直接生成
if name == "TaskBaseHandler":
return super().__new__(cls, name, bases, attrs)
# 获取子类中Meta
meta = attrs.get("Meta", None)
# 如果可以获取到Meta信息同时是一个类，清除原有Meta数据
if meta is not None and inspect.isclass(meta):
del attrs["Meta"]
# Meta是我们要求子类必须实现的，所以如果不符合条件，则需要抛出异常
else:
raise Exception("回调Handler缺少Meta信息")
meta_required_attrs = ("code", "name")
for attr in meta_required_attrs:
if not getattr(meta, attr, None):
raise Exception(f"Meta必须拥有{attr}属性")
# 要求子类必须实现的方法
required_method = ("xxxxx",)
for method in required_method:
func = attrs.get(f"{method}", None)
if not func or not isinstance(func, classmethod):
raise Exception(f"子类必须实现 {method} 类方法")
# 将子类对象生成
obj = super().__new__(cls, name, bases, attrs)
# 这里是对子类Meta中定义的字段转为子类本身属性
contribute_attrs = ["code", "name"]
for attr in contribute_attrs:
setattr(obj, attr, getattr(meta, attr, None))
# 如果类似回调任务，每次返回一个平台，仅需要使用单独某个class，可以使用字典记录数据处理handler
PLATFORM_FIELDS_DICT.update({name: obj})
# 如果一个平台返回多个特有字段数据，需要挨个匹配，可以使用列表记录数据处理handler
PLATFORM_FIELDS_LIST.append(obj)
# 最后返回这个子类
return obj
@classmethod
def exec(cls, *args, **Kwargs): # noqa
# 从 PLATFORM_FIELDS_DICT 或 PLATFORM_FIELDS_LIST 中获取obj
# 用上面的obj去处理数据
# 处理好的数据对象或者数据返回
return obj
class TaskBaskHandler(metaclass=TaskBaseMeta):
"""回调信息处理基类"""
def __init__(self, *args, **kwargs):
pass
@classmethod
def xxx(cls, *args, **kwargs):
"""可以实现多个类方法，便于校验或其他逻辑"""
pass
def xxxx(self):
"""可以实现多个方法，在返回对象时，可以供后面的逻辑调用"""
pass
class TAOBAOHandler(TaskBaskHandler):
"""某电商平台特有的处理逻辑"""
@classmethod
def _exec(cls, *args, **kwargs):
"""实现某个平台的特有业务逻辑"""
return xxx
# 定义该平台本身的一些数据或需要处理的信息
class Meta:
code = "taobao"
name = "某宝"
# 使用方法
上面如何使用？我们应该直接调用元类的exec方法，接下来就会按照规则去生成匹配某个平台对应的class
obj = TaskBaseMeta.exec(*args, **kwargs)
是不是很简单，下面可以看两个具体实现
# 具体业务实现 1：回调处理
你与合作方共同完成某个订单，但是订单中某些数据在对方明天修改了，为了保证数据一致性，对方会主动向你发起回调请求
但是每个合作方的回调请求，数据签名方式，认证方式，数据字段，处理逻辑可能都不一样，也不应该去一个每个合作方维护一个接口，或者多个if
这时候可以使用上面说的自动处理逻辑
import inspect
import logging
from typing import Any
# 全局映射字典
PLATFORM_FIELDS_DICT = {}
# 全局映射列表
PLATFORM_FIELDS_LIST = []
logger = logging.getLogger(__name__)
class TaskException(Exception):
pass
class TaskBaseMeta(type):
def __new__(cls, name, bases, attrs): # noqa
# 如果实现的是第一个父类，则直接生成
if name == "TaskBaseHandler":
return super().__new__(cls, name, bases, attrs)
# 获取子类中Meta
meta = attrs.get("Meta", None)
# 如果可以获取到Meta信息同时是一个类，清除原有Meta数据
if meta is not None and inspect.isclass(meta):
del attrs["Meta"]
# Meta是我们要求子类必须实现的，所以如果不符合条件，则需要抛出异常
else:
raise Exception("回调Handler缺少Meta信息")
if not meta.code:
raise Exception("Meta中必须有code属性")
# 要求子类必须实现的方法
required_method = ("handler",)
for method in required_method:
func = attrs.get(f"{method}", None)
if not func or not isinstance(func, classmethod):
raise Exception(f"子类必须实现 {method} 类方法")
# 将子类对象生成
obj = super().__new__(cls, name, bases, attrs)
# 这里是对子类Meta中定义的字段转为子类本身属性
contribute_attrs = ["code", "name"]
for attr in contribute_attrs:
setattr(obj, attr, getattr(meta, attr, None))
# 基于meta的code作为key，待会匹配自动处理Handler
PLATFORM_FIELDS_DICT.update({meta.code: obj})
# 最后返回这个子类
return obj
@classmethod
def exec(cls, code, *args, **kwargs): # noqa
# 从 PLATFORM_FIELDS_DICT 或 PLATFORM_FIELDS_LIST 中获取obj
obj = PLATFORM_FIELDS_DICT.get(code) # 这里code由外部调用时传入，就是自定义处理Handler里Meta定义的code，两者必须相同
# 用上面的obj去处理数据
data = obj.handler(*args, **kwargs)
# 处理好的数据对象或者数据返回
return data
class TaskBaskHandler(metaclass=TaskBaseMeta):
"""回调信息处理基类"""
def __init__(self, *args, **kwargs):
pass
@classmethod
def match_xxx(cls, *args, **kwargs):
"""基本校验逻辑"""
pass
def xxxx(self):
"""可以实现多个方法，在返回对象时，可以供后面的逻辑调用"""
pass
class TAOBAOHandler(TaskBaskHandler):
"""某电商平台特有的处理逻辑"""
@classmethod
def handler(cls, *args, **kwargs):
"""实现某个平台的特有业务逻辑"""
return xxx
# 定义该平台本身的一些数据或需要处理的信息
class Meta:
code = "taobao"
name = "某宝"