Use trakt_batch factory to create TraktBatch
MODIFY
plextraktsync/trakt_api.py
plextraktsync/trakt_api.py
@@ -102,19 +102,19 @@ def device_auth(client_id: str, client_secret: str):
@cached_property
def batch_collection_add(self):
- return TraktBatch(self, "collection", add=True, batch_delay=self.batch_delay)
+ return self.trakt_batch("collection", add=True)
@cached_property
def batch_collection_del(self):
- return TraktBatch(self, "collection", add=False, batch_delay=self.batch_delay)
+ return self.trakt_batch("collection", add=False)
@cached_property
def batch_watchlist_add(self):
- return TraktBatch(self, "watchlist", add=True, batch_delay=self.batch_delay)
+ return self.trakt_batch("watchlist", add=True)
@cached_property
def batch_watchlist_del(self):
- return TraktBatch(self, "watchlist", add=False, batch_delay=self.batch_delay)
+ return self.trakt_batch("watchlist", add=False)
@cached_property
@nocache
@@ -382,9 +382,13 @@ def flush(self):
self.batch_watchlist_add.flush(force=True)
self.batch_watchlist_del.flush(force=True)
+ @staticmethod
+ def trakt_batch(*args, **kwargs) -> TraktBatch:
+ return factory.trakt_batch(*args, **kwargs)
+
class TraktBatch:
- def __init__(self, trakt: TraktApi, name: str, add: bool, batch_delay=None):
+ def __init__(self, name: str, add: bool, trakt: TraktApi, batch_delay=None):
if name not in ["collection", "watchlist"]:
raise ValueError(f"TraktBatch name not allowed: {name}")
self.name = name

MODIFY
tests/test_batch.py
tests/test_batch.py
def test_batch_delay_none():
response = load_mock("trakt_sync_collection_response.json")
- b = TraktBatch(trakt, "collection", add=True)
+ b = TraktBatch("collection", add=True, trakt=trakt)
b.trakt_sync = Mock(return_value=response)
assert b.queue_size() == 0
@@ -27,7 +27,7 @@ def test_batch_delay_none():
def test_batch_delay_1():
response = load_mock("trakt_sync_collection_response.json")
- b = TraktBatch(trakt, "collection", add=True, batch_delay=1)
+ b = TraktBatch("collection", add=True, trakt=trakt, batch_delay=1)
b.trakt_sync = Mock(return_value=response)
assert b.queue_size() == 0

Use trakt_batch factory to create TraktBatch