我正在建立一个库来简化python中事件驱动架构的开发。我的库有两个类,希望使用它的项目必须实现这些类:Event
和EventSchema
。 EventSchema
定义如何序列化(使用棉花糖)Event
类。
我还有另一个类,EventSerializer
,其工作是:给定一个必须包含名为event_type_name
的属性的json,将其反序列化为正确的Event
实例。为了实现这一点,用户必须在其自定义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
的初始化之前手动导入了类,则该方法非常有效。除了调用库的initialize方法外,我不希望强迫用户做任何事情。这些是起作用的主要类的定义:
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
最佳答案
一种方法是查看当前导入的内容。
您当前正在查询sys.modules.keys()
的模块名称,或者可以使用globals()
,并且可以使用dir()
询问每个项目。
对于给定的课程,您可以通过例如EventSchema.__subclasses__()
。递归以查找所有后代。
关于python - 如何扫描在Python中实现我的库的特定类的类?,我们在Stack Overflow上找到一个类似的问题:https://stackoverflow.com/questions/46043816/