Created
July 23, 2020 01:33
-
-
Save johanste/3e60f4aea4af6b5efcfd51700e61a3fc to your computer and use it in GitHub Desktop.
Registering event types for event grid messages...
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| class AiEventMeta(type): | |
| def __init__(cls, name, bases, nmspc): | |
| super().__init__(name, bases, nmspc) | |
| if not hasattr(cls, 'event_type_registry'): | |
| setattr(cls, 'event_type_registry', {}) | |
| if hasattr(cls, 'EVENT_TYPE_NAME'): | |
| cls.event_type_registry[cls.EVENT_TYPE_NAME] = cls | |
| class AiEvent(metaclass=AiEventMeta): | |
| def __init__(self, data): | |
| self.data = data | |
| @classmethod | |
| def from_json(cls, data): | |
| schema_name = data['type'] | |
| if not schema_name in cls.event_type_registry: | |
| return None | |
| return cls.event_type_registry[schema_name](data) | |
| class StorageBlobCreatedAiEvent(AiEvent): | |
| EVENT_TYPE_NAME = 'azure.storage.blobCreated' | |
| class StorageBlobDeletedAiEvent(AiEvent): | |
| EVENT_TYPE_NAME = 'azure.storage.blobDeleted' | |
| event = AiEvent.from_json({ | |
| 'type': 'azure.storage.blobCreated' | |
| }) | |
| print(event.EVENT_TYPE_NAME) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment