我正在建立一个库来简化python中事件驱动架构的开发。我的库有两个类,希望使用它的项目必须实现这些类:EventEventSchemaEventSchema定义如何序列化(使用棉花糖)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方法,该方法在实现时会扫描EventEventSchema的所有子类,并根据命名约定自动将各自的事件及其模式映射。

我来自.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/

10-12 16:48
查看更多