我正在構建一個庫來緩解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你可以發表一個這樣用法的例子來說明嗎? –