Add method to manually run butler tasks (#916)

* Add butler tasks

* Add tests for butler tasks
This commit is contained in:
JonnyWong16 2022-05-16 19:51:39 -07:00 committed by GitHub
parent e0eb619561
commit 153ef66df0
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 66 additions and 6 deletions

View file

@ -524,9 +524,34 @@ class PlexServer(PlexObject):
filepath = utils.download(url, self._token, None, savepath, self._session, unpack=unpack)
return filepath
def butlerTasks(self):
""" Return a list of :class:`~plexapi.base.ButlerTask` objects. """
return self.fetchItems('/butler')
def runButlerTask(self, task):
""" Manually run a butler task immediately instead of waiting for the scheduled task to run.
Note: The butler task is run asynchronously. Check Plex Web to monitor activity.
Parameters:
task (str): The name of the task to run. (e.g. 'BackupDatabase')
Example:
.. code-block:: python
availableTasks = [task.name for task in plex.butlerTasks()]
print("Available butler tasks:", availableTasks)
"""
validTasks = [task.name for task in self.butlerTasks()]
if task not in validTasks:
raise BadRequest(
f'Invalid butler task: {task}. Available tasks are: {validTasks}'
)
self.query(f'/butler/{task}', method=self._session.post)
@deprecated('use "checkForUpdate" instead')
def check_for_update(self, force=True, download=False):
return self.checkForUpdate()
return self.checkForUpdate(force=force, download=download)
def checkForUpdate(self, force=True, download=False):
""" Returns a :class:`~plexapi.base.Release` object containing release info.
@ -1166,3 +1191,28 @@ class StatisticsResources(PlexObject):
self.__class__.__name__,
self._clean(int(self.at.timestamp()))
] if p])
@utils.registerPlexObject
class ButlerTask(PlexObject):
""" Represents a single scheduled butler task.
Attributes:
TAG (str): 'ButlerTask'
description (str): The description of the task.
enabled (bool): Whether the task is enabled.
interval (int): The interval the task is run in days.
name (str): The name of the task.
scheduleRandomized (bool): Whether the task schedule is randomized.
title (str): The title of the task.
"""
TAG = 'ButlerTask'
def _loadData(self, data):
self._data = data
self.description = data.attrib.get('description')
self.enabled = utils.cast(bool, data.attrib.get('enabled'))
self.interval = utils.cast(int, data.attrib.get('interval'))
self.name = data.attrib.get('name')
self.scheduleRandomized = utils.cast(bool, data.attrib.get('scheduleRandomized'))
self.title = data.attrib.get('title')

View file

@ -257,6 +257,16 @@ def test_server_sessions(plex):
assert len(plex.sessions()) >= 0
def test_server_butlerTasks(plex):
assert len(plex.butlerTasks())
def test_server_runButlerTask(plex):
assert plex.runButlerTask("CleanOldBundles") is None
with pytest.raises(BadRequest):
plex.runButlerTask("<This-task-should-not-exist>")
def test_server_isLatest(plex, mocker):
from os import environ
@ -271,12 +281,12 @@ def test_server_isLatest(plex, mocker):
def test_server_installUpdate(plex, mocker):
m = mocker.MagicMock(release="aa")
with utils.patch('plexapi.server.PlexServer.check_for_update', return_value=m):
with utils.patch('plexapi.server.PlexServer.checkForUpdate', return_value=m):
with utils.callable_http_patch():
plex.installUpdate()
def test_server_check_for_update(plex, mocker):
def test_server_checkForUpdate(plex, mocker):
class R:
def __init__(self, **kwargs):
self.download_key = "plex.tv/release/1337"
@ -286,8 +296,8 @@ def test_server_check_for_update(plex, mocker):
self.downloadURL = "http://path-to-update"
self.state = "downloaded"
with utils.patch('plexapi.server.PlexServer.check_for_update', return_value=R()):
rel = plex.check_for_update(force=False, download=True)
with utils.patch('plexapi.server.PlexServer.checkForUpdate', return_value=R()):
rel = plex.checkForUpdate(force=False, download=True)
assert rel.download_key == "plex.tv/release/1337"
assert rel.version == "1337"
assert rel.added == "gpu transcode"