我正在构建一个库来缓解python中事件驱动体系结构的开发。我的图书馆有两个类,希望使用它的项目必须执行:Event
和EventSchema
。 EventSchema
定义了如何序列化(使用棉花糖)Event
类。如何扫描在Python中实现某个类库的类?
我有另一个类的EventSerializer
,其任务是:考虑到必须包含一个名为event_type_name
属性,反序列化到适当的Event
实例的JSON。为了达到这个目的,用户必须提供一个字典,其中包含自定义Events
和它们各自的Schemas
之间的地图。
这是这种实例的例子:
# A derived schema
class EventMineSchema(EventSchema):
data = fields.Dict()
id = fields.Integer()
@post_load
def create_event_mine(self, data):
return EventMine(data['data'], data['id'])
# A derived event
class EventMine(Event):
event_type_name = 'event_mine'
def __init__(self, data, id):
Event.__init__(self, self.event_type_name)
self.data = data
self.id = id
def set_status(self, status):
self.data['status'] = status
def get_status(self):
return self.data['status']
def get_id(self):
return self.id
# Initialize the event serializer
EventSerializer.instance().initialize({
EventMine.event_type_name: EventMineSchema()
})
我希望保存这个麻烦让他手动提供这些映射的用户。我希望有一个无参数的initialize
方法,在其实现中,它将扫描Event
和EventSchema
的所有子类,并根据命名约定自动将各个事件与其模式进行映射。
我来自.NET背景,用反射做这件事相当容易。我如何在Python中执行此操作?我试过使用Event.__subclasses__()
的方法,该方法效果很好......如果用户在调用EventSerializer
的初始化之前手动导入了类。如果可能,除了调用库的初始化方法之外,我不希望强制用户做任何事情。
这些是主要的类的定义在行动:
class EventSchema(Schema):
event_type_name = fields.Str()
occurred_on = fields.Date()
class Event:
# A constant that subscriber can use in their "listens_to" events to
# tell they are interested in all the events that happen on their topic
ALL = 'ALL'
event_type_name = 'Default'
def __init__(self, event_type_name='Default', occurred_on=datetime.now()):
self.occurred_on = occurred_on
self.event_type_name = event_type_name
def get_occurred_on(self):
return self.occurred_on
def get_event_type_name(self):
return self.event_type_name
@Singleton
class EventSerializer:
def __init__(self):
current_module = sys.modules[__name__]
a = Event.__subclasses__()
for name, obj in inspect.getmembers(sys.modules[__name__]):
if inspect.isclass(obj):
print(obj)
self.event_serializer_map = {}
def initialize(self, event_serializer_map):
self.event_serializer_map = event_serializer_map
def deserialize(self, event_dict):
event_type_name = event_dict['event_type_name']
if event_type_name not in self.event_serializer_map:
raise ValueError("The event type {} doesn't have a registered serializer")
schema = self.event_serializer_map[event_type_name]
return schema.load(event_dict).data
def serialize(self, event):
event_type_name = event.event_type_name
if event_type_name not in self.event_serializer_map:
raise ValueError("The event type {} doesn't have a registered serializer")
schema = self.event_serializer_map[event_type_name]
return schema.dumps(event).data
为什么不将序列化作为模式的类方法并使用装饰器来枚举? –
此外,作为(部分)答案,元类。 –
@ IgnacioVazquez-Abrams你可以发表一个这样用法的例子来说明吗? –