Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def query(self, url, method=None, headers=None, timeout=None, **kwargs):
method = method or self._session.get
timeout = timeout or TIMEOUT
log.debug('%s %s %s', method.__name__.upper(), url, kwargs.get('json', ''))
headers = self._headers(**headers or {})
response = method(url, headers=headers, timeout=timeout, **kwargs)
if response.status_code not in (200, 201, 204): # pragma: no cover
codename = codes.get(response.status_code)[0]
errtext = response.text.replace('\n', ' ')
raise BadRequest('(%s) %s %s; %s' % (response.status_code, codename, response.url, errtext))
data = response.text.encode('utf8')
return ElementTree.fromstring(data) if data.strip() else None
def set(self, value):
""" Set a new value for this setitng. NOTE: You must call plex.settings.save() for before
any changes to setting values are persisted to the :class:`~plexapi.server.PlexServer`.
"""
# check a few things up front
if not isinstance(value, self.TYPES[self.type]['type']):
badtype = type(value).__name__
raise BadRequest('Invalid value for %s: a %s is required, not %s' % (self.id, self.type, badtype))
if self.enumValues and value not in self.enumValues:
raise BadRequest('Invalid value for %s: %s not in %s' % (self.id, value, list(self.enumValues)))
# store value off to the side until we call settings.save()
tostr = self.TYPES[self.type]['tostr']
self._setValue = tostr(value)
def url(self, key, includeToken=False):
""" Build a URL string with proper token argument. Token will be appended to the URL
if either includeToken is True or CONFIG.log.show_secrets is 'true'.
"""
if not self._baseurl:
raise BadRequest('PlexClient object missing baseurl.')
if self._token and (includeToken or self._showSecrets):
delim = '&' if '?' in key else '?'
return '%s%s%sX-Plex-Token=%s' % (self._baseurl, key, delim, self._token)
return '%s%s' % (self._baseurl, key)
season (int): Season number (default:None; required if title not specified).
episode (int): Episode number (default:None; required if title not specified).
Raises:
BadRequest: If season and episode is missing.
NotFound: If the episode is missing.
"""
if title:
key = '/library/metadata/%s/allLeaves' % self.ratingKey
return self.fetchItem(key, title__iexact=title)
elif season and episode:
results = [i for i in self.episodes() if i.seasonNumber == season and i.index == episode]
if results:
return results[0]
raise NotFound('Couldnt find %s S%s E%s' % (self.title, season, episode))
raise BadRequest('Missing argument: title or season and episode are required')
def deleteWebhook(self, url):
urls = copy.copy(self._webhooks)
if url not in urls:
raise BadRequest('Webhook does not exist: %s' % url)
urls.remove(url)
return self.setWebhooks(urls)
def query(self, url, method=None, headers=None, timeout=None, **kwargs):
method = method or self._session.get
timeout = timeout or TIMEOUT
log.debug('%s %s %s', method.__name__.upper(), url, kwargs.get('json', ''))
headers = self._headers(**headers or {})
response = method(url, headers=headers, timeout=timeout, **kwargs)
if response.status_code not in (200, 201, 204): # pragma: no cover
codename = codes.get(response.status_code)[0]
errtext = response.text.replace('\n', ' ')
log.warning('BadRequest (%s) %s %s; %s' % (response.status_code, codename, response.url, errtext))
raise BadRequest('(%s) %s %s; %s' % (response.status_code, codename, response.url, errtext))
data = response.text.encode('utf8')
return ElementTree.fromstring(data) if data.strip() else None
def syncItems(self):
""" Returns an instance of :class:`plexapi.sync.SyncList` for current device.
Raises:
:class:`plexapi.exceptions.BadRequest`: when the device doesn`t provides `sync-target`.
"""
if 'sync-target' not in self.provides:
raise BadRequest('Requested syncList for device which do not provides sync-target')
return self._server.syncItems(client=self)
def _cleanSearchFilter(self, category, value, libtype=None):
# check a few things before we begin
if category.endswith('!'):
if category[:-1] not in self.ALLOWED_FILTERS:
raise BadRequest('Unknown filter category: %s' % category[:-1])
elif category not in self.ALLOWED_FILTERS:
raise BadRequest('Unknown filter category: %s' % category)
if category in self.BOOLEAN_FILTERS:
return '1' if value else '0'
if not isinstance(value, (list, tuple)):
value = [value]
# convert list of values to list of keys or ids
result = set()
choices = self.listChoices(category, libtype)
lookup = {c.title.lower(): unquote(unquote(c.key)) for c in choices}
allowed = set(c.key for c in choices)
for item in value:
item = str((item.id or item.tag) if isinstance(item, MediaTag) else item).lower()
# find most logical choice(s) to use in url
if item in allowed: result.add(item); continue
if item in lookup: result.add(lookup[item]); continue
matches = [k for t, k in lookup.items() if item in t]
if matches: map(result.add, matches); continue
season (int): Season number (default:None; required if title not specified).
episode (int): Episode number (default:None; required if title not specified).
Raises:
:class:`plexapi.exceptions.BadRequest`: If season and episode is missing.
:class:`plexapi.exceptions.NotFound`: If the episode is missing.
"""
if title:
key = '/library/metadata/%s/allLeaves' % self.ratingKey
return self.fetchItem(key, title__iexact=title)
elif season is not None and episode:
results = [i for i in self.episodes() if i.seasonNumber == season and i.index == episode]
if results:
return results[0]
raise NotFound('Couldnt find %s S%s E%s' % (self.title, season, episode))
raise BadRequest('Missing argument: title or season and episode are required')
def addItems(self, items):
""" Add items to a playlist. """
if not isinstance(items, (list, tuple)):
items = [items]
ratingKeys = []
for item in items:
if item.listType != self.playlistType: # pragma: no cover
raise BadRequest('Can not mix media types when building a playlist: %s and %s' %
(self.playlistType, item.listType))
ratingKeys.append(str(item.ratingKey))
uuid = items[0].section().uuid
ratingKeys = ','.join(ratingKeys)
key = '%s/items%s' % (self.key, utils.joinArgs({
'uri': 'library://%s/directory//library/metadata/%s' % (uuid, ratingKeys)
}))
result = self._server.query(key, method=self._server._session.put)
self.reload()
return result