Move EventFactory to watch module
MODIFY
plextraktsync/events.py
plextraktsync/events.py
-import importlib
-
-
class Event(dict):
def __str__(self):
return f"{self.__class__}:{str(self.copy())}"
@@ -92,41 +89,3 @@ def title(self):
class TranscodeSession(Event):
pass
-
-
-class EventFactory:
- EVENTS = {
- "account": "AccountUpdateNotification",
- "activity": "ActivityNotification",
- "backgroundProcessingQueue": "BackgroundProcessingQueueEventNotification",
- "playing": "PlaySessionStateNotification",
- "preference": "Setting",
- "progress": "ProgressNotification",
- "reachability": "ReachabilityNotification",
- "status": "StatusNotification",
- "timeline": "TimelineEntry",
- "transcodeSession.end": "TranscodeSession",
- "transcodeSession.start": "TranscodeSession",
- "transcodeSession.update": "TranscodeSession",
- }
-
- def __init__(self):
- self.module = importlib.import_module(self.__module__)
-
- def get_events(self, message):
- if message["size"] != 1:
- raise ValueError(f"Unexpected size: {message}")
-
- message_type = message["type"]
- if message_type not in self.EVENTS:
- return
- class_name = self.EVENTS[message_type]
- if class_name not in message:
- return
- for data in message[class_name]:
- event = self.create(class_name, **data)
- yield event
-
- def create(self, name, **kwargs):
- cls = getattr(self.module, name)
- return cls(**kwargs)

MODIFY
plextraktsync/watch/EventDispatcher.py
plextraktsync/watch/EventDispatcher.py
-from plextraktsync.events import Error, EventFactory
+from plextraktsync.events import Error
from plextraktsync.factory import logging
+from plextraktsync.watch.EventFactory import EventFactory
class EventDispatcher:

ADD
plextraktsync/watch/EventFactory.py
+import importlib
+
+
+class EventFactory:
+ EVENTS = {
+ "account": "AccountUpdateNotification",
+ "activity": "ActivityNotification",
+ "backgroundProcessingQueue": "BackgroundProcessingQueueEventNotification",
+ "playing": "PlaySessionStateNotification",
+ "preference": "Setting",
+ "progress": "ProgressNotification",
+ "reachability": "ReachabilityNotification",
+ "status": "StatusNotification",
+ "timeline": "TimelineEntry",
+ "transcodeSession.end": "TranscodeSession",
+ "transcodeSession.start": "TranscodeSession",
+ "transcodeSession.update": "TranscodeSession",
+ }
+
+ def __init__(self):
+ self.module = importlib.import_module(self.__module__)
+
+ def get_events(self, message):
+ if message["size"] != 1:
+ raise ValueError(f"Unexpected size: {message}")
+
+ message_type = message["type"]
+ if message_type not in self.EVENTS:
+ return
+ class_name = self.EVENTS[message_type]
+ if class_name not in message:
+ return
+ for data in message[class_name]:
+ event = self.create(class_name, **data)
+ yield event
+
+ def create(self, name, **kwargs):
+ cls = getattr(self.module, name)
+ return cls(**kwargs)

MODIFY
tests/test_events.py
tests/test_events.py
#!/usr/bin/env python3 -m pytest
-from plextraktsync.events import ActivityNotification, EventFactory
+from plextraktsync.events import ActivityNotification
from plextraktsync.watch.EventDispatcher import EventDispatcher
+from plextraktsync.watch.EventFactory import EventFactory
from tests.conftest import load_mock

Move EventFactory to watch module